Check if there are clients blocked in WAIT that can be unblocked since * we received enough ACKs from slaves. */
| 3231 | /* Check if there are clients blocked in WAIT that can be unblocked since |
| 3232 | * we received enough ACKs from slaves. */ |
| 3233 | void processClientsWaitingReplicas(void) { |
| 3234 | long long last_offset = 0; |
| 3235 | int last_numreplicas = 0; |
| 3236 | |
| 3237 | listIter li; |
| 3238 | listNode *ln; |
| 3239 | |
| 3240 | listRewind(server.clients_waiting_acks,&li); |
| 3241 | while((ln = listNext(&li))) { |
| 3242 | client *c = ln->value; |
| 3243 | |
| 3244 | /* Every time we find a client that is satisfied for a given |
| 3245 | * offset and number of replicas, we remember it so the next client |
| 3246 | * may be unblocked without calling replicationCountAcksByOffset() |
| 3247 | * if the requested offset / replicas were equal or less. */ |
| 3248 | if (last_offset && last_offset >= c->bpop.reploffset && |
| 3249 | last_numreplicas >= c->bpop.numreplicas) |
| 3250 | { |
| 3251 | unblockClient(c); |
| 3252 | addReplyLongLong(c,last_numreplicas); |
| 3253 | } else { |
| 3254 | int numreplicas = replicationCountAcksByOffset(c->bpop.reploffset); |
| 3255 | |
| 3256 | if (numreplicas >= c->bpop.numreplicas) { |
| 3257 | last_offset = c->bpop.reploffset; |
| 3258 | last_numreplicas = numreplicas; |
| 3259 | unblockClient(c); |
| 3260 | addReplyLongLong(c,numreplicas); |
| 3261 | } |
| 3262 | } |
| 3263 | } |
| 3264 | } |
| 3265 | |
| 3266 | /* Return the slave replication offset for this instance, that is |
| 3267 | * the offset for which we already processed the master replication stream. */ |
no test coverage detected