1 /* blocked.c - generic support for blocking operations like BLPOP & WAIT. 2 * 3 * Copyright (c) 2009-2012, Salvatore Sanfilippo <antirez at gmail dot com> 4 * All rights reserved. 5 * 6 * Redistribution and use in source and binary forms, with or without 7 * modification, are permitted provided that the following conditions are met: 8 * 9 * * Redistributions of source code must retain the above copyright notice, 10 * this list of conditions and the following disclaimer. 11 * * Redistributions in binary form must reproduce the above copyright 12 * notice, this list of conditions and the following disclaimer in the 13 * documentation and/or other materials provided with the distribution. 14 * * Neither the name of Redis nor the names of its contributors may be used 15 * to endorse or promote products derived from this software without 16 * specific prior written permission. 17 * 18 * THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS "AS IS" 19 * AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE 20 * IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR PURPOSE 21 * ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT OWNER OR CONTRIBUTORS BE 22 * LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL, SPECIAL, EXEMPLARY, OR 23 * CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT LIMITED TO, PROCUREMENT OF 24 * SUBSTITUTE GOODS OR SERVICES; LOSS OF USE, DATA, OR PROFITS; OR BUSINESS 25 * INTERRUPTION) HOWEVER CAUSED AND ON ANY THEORY OF LIABILITY, WHETHER IN 26 * CONTRACT, STRICT LIABILITY, OR TORT (INCLUDING NEGLIGENCE OR OTHERWISE) 27 * ARISING IN ANY WAY OUT OF THE USE OF THIS SOFTWARE, EVEN IF ADVISED OF THE 28 * POSSIBILITY OF SUCH DAMAGE. 29 * 30 * --------------------------------------------------------------------------- 31 * 32 * API: 33 * 34 * getTimeoutFromObjectOrReply() is just an utility function to parse a 35 * timeout argument since blocking operations usually require a timeout. 36 * 37 * blockClient() set the REDIS_BLOCKED flag in the client, and set the 38 * specified block type 'btype' filed to one of REDIS_BLOCKED_* macros. 39 * 40 * unblockClient() unblocks the client doing the following: 41 * 1) It calls the btype-specific function to cleanup the state. 42 * 2) It unblocks the client by unsetting the REDIS_BLOCKED flag. 43 * 3) It puts the client into a list of just unblocked clients that are 44 * processed ASAP in the beforeSleep() event loop callback, so that 45 * if there is some query buffer to process, we do it. This is also 46 * required because otherwise there is no 'readable' event fired, we 47 * already read the pending commands. We also set the REDIS_UNBLOCKED 48 * flag to remember the client is in the unblocked_clients list. 49 * 50 * processUnblockedClients() is called inside the beforeSleep() function 51 * to process the query buffer from unblocked clients and remove the clients 52 * from the blocked_clients queue. 53 * 54 * replyToBlockedClientTimedOut() is called by the cron function when 55 * a client blocked reaches the specified timeout (if the timeout is set 56 * to 0, no timeout is processed). 57 * It usually just needs to send a reply to the client. 58 * 59 * When implementing a new type of blocking opeation, the implementation 60 * should modify unblockClient() and replyToBlockedClientTimedOut() in order 61 * to handle the btype-specific behavior of this two functions. 62 */ 63 64 #include "redis.h" 65 66 /* Get a timeout value from an object and store it into 'timeout'. 67 * The final timeout is always stored as milliseconds as a time where the 68 * timeout will expire, however the parsing is performed according to 69 * the 'unit' that can be seconds or milliseconds. 70 * 71 * Note that if the timeout is zero (usually from the point of view of 72 * commands API this means no timeout) the value stored into 'timeout' 73 * is zero. */ 74 int getTimeoutFromObjectOrReply(redisClient *c, robj *object, mstime_t *timeout, int unit) { 75 long long tval; 76 77 if (getLongLongFromObjectOrReply(c,object,&tval, 78 "timeout is not an integer or out of range") != REDIS_OK) 79 return REDIS_ERR; 80 81 if (tval < 0) { 82 addReplyError(c,"timeout is negative"); 83 return REDIS_ERR; 84 } 85 86 if (tval > 0) { 87 if (unit == UNIT_SECONDS) tval *= 1000; 88 tval += mstime(); 89 } 90 *timeout = tval; 91 92 return REDIS_OK; 93 } 94 95 /* Block a client for the specific operation type. Once the REDIS_BLOCKED 96 * flag is set client query buffer is not longer processed, but accumulated, 97 * and will be processed when the client is unblocked. */ 98 void blockClient(redisClient *c, int btype) { 99 c->flags |= REDIS_BLOCKED; 100 c->btype = btype; 101 server.bpop_blocked_clients++; 102 } 103 104 /* This function is called in the beforeSleep() function of the event loop 105 * in order to process the pending input buffer of clients that were 106 * unblocked after a blocking operation. */ 107 void processUnblockedClients(void) { 108 listNode *ln; 109 redisClient *c; 110 111 while (listLength(server.unblocked_clients)) { 112 ln = listFirst(server.unblocked_clients); 113 redisAssert(ln != NULL); 114 c = ln->value; 115 listDelNode(server.unblocked_clients,ln); 116 c->flags &= ~REDIS_UNBLOCKED; 117 c->btype = REDIS_BLOCKED_NONE; 118 119 /* Process remaining data in the input buffer. */ 120 if (c->querybuf && sdslen(c->querybuf) > 0) { 121 server.current_client = c; 122 processInputBuffer(c); 123 server.current_client = NULL; 124 } 125 } 126 } 127 128 /* Unblock a client calling the right function depending on the kind 129 * of operation the client is blocking for. */ 130 void unblockClient(redisClient *c) { 131 if (c->btype == REDIS_BLOCKED_LIST) { 132 unblockClientWaitingData(c); 133 } else if (c->btype == REDIS_BLOCKED_WAIT) { 134 unblockClientWaitingReplicas(c); 135 } else { 136 redisPanic("Unknown btype in unblockClient()."); 137 } 138 /* Clear the flags, and put the client in the unblocked list so that 139 * we'll process new commands in its query buffer ASAP. */ 140 c->flags &= ~REDIS_BLOCKED; 141 c->flags |= REDIS_UNBLOCKED; 142 c->btype = REDIS_BLOCKED_NONE; 143 server.bpop_blocked_clients--; 144 listAddNodeTail(server.unblocked_clients,c); 145 } 146 147 /* This function gets called when a blocked client timed out in order to 148 * send it a reply of some kind. */ 149 void replyToBlockedClientTimedOut(redisClient *c) { 150 if (c->btype == REDIS_BLOCKED_LIST) { 151 addReply(c,shared.nullmultibulk); 152 } else if (c->btype == REDIS_BLOCKED_WAIT) { 153 addReplyLongLong(c,replicationCountAcksByOffset(c->bpop.reploffset)); 154 } else { 155 redisPanic("Unknown btype in replyToBlockedClientTimedOut()."); 156 } 157 } 158 159