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
| 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. */ |
| 6142 | int 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); |
no test coverage detected