Check if there are clients blocked in WAIT that can be unblocked since * we received enough ACKs from slaves. */
| 4698 | /* Check if there are clients blocked in WAIT that can be unblocked since |
| 4699 | * we received enough ACKs from slaves. */ |
| 4700 | void processClientsWaitingReplicas(void) { |
| 4701 | long long last_offset = 0; |
| 4702 | int last_numreplicas = 0; |
| 4703 | |
| 4704 | listIter li; |
| 4705 | listNode *ln; |
| 4706 | |
| 4707 | listRewind(g_pserver->clients_waiting_acks,&li); |
| 4708 | while((ln = listNext(&li))) { |
| 4709 | client *c = (client*)ln->value; |
| 4710 | std::unique_lock<fastlock> ul(c->lock); |
| 4711 | |
| 4712 | /* Every time we find a client that is satisfied for a given |
| 4713 | * offset and number of replicas, we remember it so the next client |
| 4714 | * may be unblocked without calling replicationCountAcksByOffset() |
| 4715 | * if the requested offset / replicas were equal or less. */ |
| 4716 | if (last_offset && last_offset >= c->bpop.reploffset && |
| 4717 | last_numreplicas >= c->bpop.numreplicas) |
| 4718 | { |
| 4719 | unblockClient(c); |
| 4720 | addReplyLongLong(c,last_numreplicas); |
| 4721 | } else { |
| 4722 | int numreplicas = replicationCountAcksByOffset(c->bpop.reploffset); |
| 4723 | |
| 4724 | if (numreplicas >= c->bpop.numreplicas) { |
| 4725 | last_offset = c->bpop.reploffset; |
| 4726 | last_numreplicas = numreplicas; |
| 4727 | unblockClient(c); |
| 4728 | addReplyLongLong(c,numreplicas); |
| 4729 | } |
| 4730 | } |
| 4731 | } |
| 4732 | } |
| 4733 | |
| 4734 | /* Return the replica replication offset for this instance, that is |
| 4735 | * the offset for which we already processed the master replication stream. */ |
no test coverage detected