gvsafronov/futriix
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Improve perf of reading cluster bitfield

Browse Source 
Former-commit-id: 9371c005aa7ffc2060b1b787e4268bc25336ca15
This commit is contained in:
John Sully 2019-11-17 14:52:12 -05:00
parent c34fee9ed4
commit baffeff5c7
1 changed files with 55 additions and 36 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

91
src/cluster.cpp
View File

@ -4180,57 +4180,76 @@ void clusterReplyMultiBulkSlots(client *c) {
dictIterator *di = dictGetSafeIterator(g_pserver->cluster->nodes); dictIterator *di = dictGetSafeIterator(g_pserver->cluster->nodes);
while((de = dictNext(di)) != NULL) { while((de = dictNext(di)) != NULL) {
clusterNode *node = (clusterNode*)dictGetVal(de); clusterNode *node = (clusterNode*)dictGetVal(de);
int j = 0, start = -1; int start = -1;
/* Skip slaves (that are iterated when producing the output of their /* Skip slaves (that are iterated when producing the output of their
* master) and masters not serving any slot. */ * master) and masters not serving any slot. */
if (!nodeIsMaster(node) || node->numslots == 0) continue; if (!nodeIsMaster(node) || node->numslots == 0) continue;
static_assert((CLUSTER_SLOTS % (sizeof(uint32_t)*8)) == 0, "code below assumes the bitfield is a multiple of sizeof(unsinged)");
for (j = 0; j < CLUSTER_SLOTS; j++) { for (unsigned iw = 0; iw < (CLUSTER_SLOTS/sizeof(uint32_t)/8); ++iw)
int bit, i; {
uint32_t wordCur = reinterpret_cast<uint32_t*>(node->slots)[iw];
if ((bit = clusterNodeGetSlotBit(node,j)) != 0) { if (iw != ((CLUSTER_SLOTS/sizeof(uint32_t)/8)-1))
if (start == -1) start = j; {
if (start == -1 && wordCur == 0)
continue;
if (start != -1 && (wordCur+1)==0)
continue;
} }
if (start != -1 && (!bit || j == CLUSTER_SLOTS-1)) {
int nested_elements = 3; /* slots (2) + master addr (1). */
void *nested_replylen = addReplyDeferredLen(c);
if (bit && j == CLUSTER_SLOTS-1) j++; unsigned ibitStartLoop = iw*sizeof(uint32_t)*8;
/* If slot exists in output map, add to it's list. for (unsigned j = ibitStartLoop; j < (iw+1)*sizeof(uint32_t)*8; j++) {
* else, create a new output map for this slot */ int i;
if (start == j-1) { int bit = (int)(wordCur & 1);
addReplyLongLong(c, start); /* only one slot; low==high */ wordCur >>= 1;
addReplyLongLong(c, start); if (bit != 0) {
} else { if (start == -1) start = j;
addReplyLongLong(c, start); /* low */
addReplyLongLong(c, j-1); /* high */
} }
start = -1; if (start != -1 && (!bit || j == CLUSTER_SLOTS-1)) {
int nested_elements = 3; /* slots (2) + master addr (1). */
void *nested_replylen = addReplyDeferredLen(c);
/* First node reply position is always the master */ if (bit && j == CLUSTER_SLOTS-1) j++;
addReplyArrayLen(c, 3);
addReplyBulkCString(c, node->ip);
addReplyLongLong(c, node->port);
addReplyBulkCBuffer(c, node->name, CLUSTER_NAMELEN);
/* Remaining nodes in reply are replicas for slot range */ /* If slot exists in output map, add to it's list.
for (i = 0; i < node->numslaves; i++) { * else, create a new output map for this slot */
/* This loop is copy/pasted from clusterGenNodeDescription() if (start == j-1) {
* with modifications for per-slot node aggregation */ addReplyLongLong(c, start); /* only one slot; low==high */
if (nodeFailed(node->slaves[i])) continue; addReplyLongLong(c, start);
} else {
addReplyLongLong(c, start); /* low */
addReplyLongLong(c, j-1); /* high */
}
start = -1;
/* First node reply position is always the master */
addReplyArrayLen(c, 3); addReplyArrayLen(c, 3);
addReplyBulkCString(c, node->slaves[i]->ip); addReplyBulkCString(c, node->ip);
addReplyLongLong(c, node->slaves[i]->port); addReplyLongLong(c, node->port);
addReplyBulkCBuffer(c, node->slaves[i]->name, CLUSTER_NAMELEN); addReplyBulkCBuffer(c, node->name, CLUSTER_NAMELEN);
nested_elements++;
/* Remaining nodes in reply are replicas for slot range */
for (i = 0; i < node->numslaves; i++) {
/* This loop is copy/pasted from clusterGenNodeDescription()
* with modifications for per-slot node aggregation */
if (nodeFailed(node->slaves[i])) continue;
addReplyArrayLen(c, 3);
addReplyBulkCString(c, node->slaves[i]->ip);
addReplyLongLong(c, node->slaves[i]->port);
addReplyBulkCBuffer(c, node->slaves[i]->name, CLUSTER_NAMELEN);
nested_elements++;
}
setDeferredArrayLen(c, nested_replylen, nested_elements);
num_masters++;
} }
setDeferredArrayLen(c, nested_replylen, nested_elements);
num_masters++;
} }
} }
serverAssert(start == -1);
} }
dictReleaseIterator(di); dictReleaseIterator(di);
setDeferredArrayLen(c, slot_replylen, num_masters); setDeferredArrayLen(c, slot_replylen, num_masters);
} }