MCPcopy Create free account
hub / github.com/Snapchat/KeyDB / clusterRedirectBlockedClientIfNeeded

Function clusterRedirectBlockedClientIfNeeded

src/cluster.cpp:6142–6206  ·  view source on GitHub ↗

This function is called by the function processing clients incrementally * to detect timeouts, in order to handle the following case: * * 1) A client blocks with BLPOP or similar blocking operation. * 2) The master migrates the hash slot elsewhere or turns into a slave. * 3) The client may remain blocked forever (or up to the max timeout time) * waiting for a key change that will never ha

Source from the content-addressed store, hash-verified

6140 * longer handles, the client is sent a redirection error, and the function
6141 * returns 1. Otherwise 0 is returned and no operation is performed. */
6142int clusterRedirectBlockedClientIfNeeded(client *c) {
6143 serverAssert(GlobalLocksAcquired());
6144 if (c->flags & CLIENT_BLOCKED &&
6145 (c->btype == BLOCKED_LIST ||
6146 c->btype == BLOCKED_ZSET ||
6147 c->btype == BLOCKED_STREAM))
6148 {
6149 dictEntry *de;
6150 dictIterator *di;
6151
6152 /* If the cluster is down, unblock the client with the right error.
6153 * If the cluster is configured to allow reads on cluster down, we
6154 * still want to emit this error since a write will be required
6155 * to unblock them which may never come. */
6156 if (g_pserver->cluster->state == CLUSTER_FAIL) {
6157 clusterRedirectClient(c,NULL,0,CLUSTER_REDIR_DOWN_STATE);
6158 return 1;
6159 }
6160
6161 /* All keys must belong to the same slot, so check first key only. */
6162 di = dictGetIterator(c->bpop.keys);
6163 if ((de = dictNext(di)) != NULL) {
6164 robj *key = (robj*)dictGetKey(de);
6165 int slot = keyHashSlot((char*)ptrFromObj(key), sdslen(szFromObj(key)));
6166 clusterNode *node = g_pserver->cluster->slots[slot];
6167
6168 /* if the client is read-only and attempting to access key that our
6169 * replica can handle, allow it. */
6170 if ((c->flags & CLIENT_READONLY) &&
6171 (c->lastcmd->flags & CMD_READONLY) &&
6172 nodeIsSlave(myself) && myself->slaveof == node)
6173 {
6174 node = myself;
6175 }
6176
6177 /* if the client is read-only and attempting to access key that our
6178 * replica can handle, allow it. */
6179 if ((c->flags & CLIENT_READONLY) &&
6180 !(c->lastcmd->flags & CMD_WRITE) &&
6181 nodeIsSlave(myself) && myself->slaveof == node)
6182 {
6183 node = myself;
6184 }
6185
6186 /* We send an error and unblock the client if:
6187 * 1) The slot is unassigned, emitting a cluster down error.
6188 * 2) The slot is not handled by this node, nor being imported. */
6189 if (node != myself &&
6190 g_pserver->cluster->importing_slots_from[slot] == NULL)
6191 {
6192 if (node == NULL) {
6193 clusterRedirectClient(c,NULL,0,
6194 CLUSTER_REDIR_DOWN_UNBOUND);
6195 } else {
6196 clusterRedirectClient(c,node,slot,
6197 CLUSTER_REDIR_MOVED);
6198 }
6199 dictReleaseIterator(di);

Callers 1

clientsCronHandleTimeoutFunction · 0.85

Calls 9

GlobalLocksAcquiredFunction · 0.85
clusterRedirectClientFunction · 0.85
keyHashSlotFunction · 0.85
ptrFromObjFunction · 0.85
sdslenFunction · 0.85
szFromObjFunction · 0.85
dictGetIteratorFunction · 0.70
dictNextFunction · 0.70
dictReleaseIteratorFunction · 0.70

Tested by

no test coverage detected