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 * If the blocking operation waits for certain keys to change state, the 63 * clusterRedirectBlockedClientIfNeeded() function should also be updated. 64 */ 65 66 #include "redis.h" 67 68 /* Get a timeout value from an object and store it into 'timeout'. 69 * The final timeout is always stored as milliseconds as a time where the 70 * timeout will expire, however the parsing is performed according to 71 * the 'unit' that can be seconds or milliseconds. 72 * 73 * Note that if the timeout is zero (usually from the point of view of 74 * commands API this means no timeout) the value stored into 'timeout' 75 * is zero. */ 76 int getTimeoutFromObjectOrReply(redisClient *c, robj *object, mstime_t *timeout, int unit) { 77 long long tval; 78 79 if (getLongLongFromObjectOrReply(c,object,&tval, 80 "timeout is not an integer or out of range") != REDIS_OK) 81 return REDIS_ERR; 82 83 if (tval < 0) { 84 addReplyError(c,"timeout is negative"); 85 return REDIS_ERR; 86 } 87 88 if (tval > 0) { 89 if (unit == UNIT_SECONDS) tval *= 1000; 90 tval += mstime(); 91 } 92 *timeout = tval; 93 94 return REDIS_OK; 95 } 96 97 /* Block a client for the specific operation type. Once the REDIS_BLOCKED 98 * flag is set client query buffer is not longer processed, but accumulated, 99 * and will be processed when the client is unblocked. */ 100 void blockClient(redisClient *c, int btype) { 101 c->flags |= REDIS_BLOCKED; 102 c->btype = btype; 103 server.bpop_blocked_clients++; 104 } 105 106 /* This function is called in the beforeSleep() function of the event loop 107 * in order to process the pending input buffer of clients that were 108 * unblocked after a blocking operation. */ 109 void processUnblockedClients(void) { 110 listNode *ln; 111 redisClient *c; 112 113 while (listLength(server.unblocked_clients)) { 114 ln = listFirst(server.unblocked_clients); 115 redisAssert(ln != NULL); 116 c = ln->value; 117 listDelNode(server.unblocked_clients,ln); 118 c->flags &= ~REDIS_UNBLOCKED; 119 120 /* Process remaining data in the input buffer. */ 121 if (c->querybuf && sdslen(c->querybuf) > 0) { 122 processInputBuffer(c); 123 } 124 } 125 } 126 127 /* Unblock a client calling the right function depending on the kind 128 * of operation the client is blocking for. */ 129 void unblockClient(redisClient *c) { 130 if (c->btype == REDIS_BLOCKED_LIST) { 131 unblockClientWaitingData(c); 132 } else if (c->btype == REDIS_BLOCKED_WAIT) { 133 unblockClientWaitingReplicas(c); 134 } else { 135 redisPanic("Unknown btype in unblockClient()."); 136 } 137 /* Clear the flags, and put the client in the unblocked list so that 138 * we'll process new commands in its query buffer ASAP. */ 139 c->flags &= ~REDIS_BLOCKED; 140 c->flags |= REDIS_UNBLOCKED; 141 c->btype = REDIS_BLOCKED_NONE; 142 server.bpop_blocked_clients--; 143 listAddNodeTail(server.unblocked_clients,c); 144 } 145 146 /* This function gets called when a blocked client timed out in order to 147 * send it a reply of some kind. */ 148 void replyToBlockedClientTimedOut(redisClient *c) { 149 if (c->btype == REDIS_BLOCKED_LIST) { 150 addReply(c,shared.nullmultibulk); 151 } else if (c->btype == REDIS_BLOCKED_WAIT) { 152 addReplyLongLong(c,replicationCountAcksByOffset(c->bpop.reploffset)); 153 } else { 154 redisPanic("Unknown btype in replyToBlockedClientTimedOut()."); 155 } 156 } 157 158 /* Mass-unblock clients because something changed in the instance that makes 159 * blocking no longer safe. For example clients blocked in list operations 160 * in an instance which turns from master to slave is unsafe, so this function 161 * is called when a master turns into a slave. 162 * 163 * The semantics is to send an -UNBLOCKED error to the client, disconnecting 164 * it at the same time. */ 165 void disconnectAllBlockedClients(void) { 166 listNode *ln; 167 listIter li; 168 169 listRewind(server.clients,&li); 170 while((ln = listNext(&li))) { 171 redisClient *c = listNodeValue(ln); 172 173 if (c->flags & REDIS_BLOCKED) { 174 addReplySds(c,sdsnew( 175 "-UNBLOCKED force unblock from blocking operation, " 176 "instance state changed (master -> slave?)\r\n")); 177 unblockClient(c); 178 c->flags |= REDIS_CLOSE_AFTER_REPLY; 179 } 180 } 181 } 182