This function will check the moduleUnblockedClients queue in order to * call the reply callback and really unblock the client. * * Clients end into this list because of calls to RM_UnblockClient(), * however it is possible that while the module was doing work for the * blocked client, it was terminated by Redis (for timeout or other reasons). * When this happens the RedisModuleBlockedClient
| 5673 | * When this happens the RedisModuleBlockedClient structure in the queue |
| 5674 | * will have the 'client' field set to NULL. */ |
| 5675 | void moduleHandleBlockedClients(int iel) { |
| 5676 | RedisModuleBlockedClient *bc; |
| 5677 | serverAssert(GlobalLocksAcquired()); |
| 5678 | |
| 5679 | pthread_mutex_lock(&moduleUnblockedClientsMutex); |
| 5680 | /* Here we unblock all the pending clients blocked in modules operations |
| 5681 | * so we can read every pending "awake byte" in the pipe. */ |
| 5682 | char buf[1]; |
| 5683 | while (read(g_pserver->module_blocked_pipe[0],buf,1) == 1); |
| 5684 | listIter li; |
| 5685 | listNode *ln; |
| 5686 | listRewind(moduleUnblockedClients, &li); |
| 5687 | while ((ln = listNext(&li))) { |
| 5688 | bc = (RedisModuleBlockedClient*)ln->value; |
| 5689 | client *c = bc->client; |
| 5690 | if ((c != nullptr) && (iel != c->iel)) |
| 5691 | continue; |
| 5692 | |
| 5693 | std::unique_lock<fastlock> ul; |
| 5694 | listDelNode(moduleUnblockedClients,ln); |
| 5695 | pthread_mutex_unlock(&moduleUnblockedClientsMutex); |
| 5696 | |
| 5697 | if (c) |
| 5698 | { |
| 5699 | AssertCorrectThread(c); |
| 5700 | ul = std::unique_lock<fastlock>(c->lock); |
| 5701 | } |
| 5702 | |
| 5703 | /* Release the lock during the loop, as long as we don't |
| 5704 | * touch the shared list. */ |
| 5705 | |
| 5706 | /* Call the reply callback if the client is valid and we have |
| 5707 | * any callback. However the callback is not called if the client |
| 5708 | * was blocked on keys (RM_BlockClientOnKeys()), because we already |
| 5709 | * called such callback in moduleTryServeClientBlockedOnKey() when |
| 5710 | * the key was signaled as ready. */ |
| 5711 | uint64_t reply_us = 0; |
| 5712 | if (c && !bc->blocked_on_keys && bc->reply_callback) { |
| 5713 | RedisModuleCtx ctx = REDISMODULE_CTX_INIT; |
| 5714 | ctx.flags |= REDISMODULE_CTX_BLOCKED_REPLY; |
| 5715 | ctx.blocked_privdata = bc->privdata; |
| 5716 | ctx.blocked_ready_key = NULL; |
| 5717 | ctx.module = bc->module; |
| 5718 | ctx.client = bc->client; |
| 5719 | ctx.blocked_client = bc; |
| 5720 | monotime replyTimer; |
| 5721 | elapsedStart(&replyTimer); |
| 5722 | bc->reply_callback(&ctx,(void**)c->argv,c->argc); |
| 5723 | reply_us = elapsedUs(replyTimer); |
| 5724 | moduleFreeContext(&ctx); |
| 5725 | } |
| 5726 | /* Update stats now that we've finished the blocking operation. |
| 5727 | * This needs to be out of the reply callback above given that a |
| 5728 | * module might not define any callback and still do blocking ops. |
| 5729 | */ |
| 5730 | if (c && !bc->blocked_on_keys) { |
| 5731 | updateStatsOnUnblock(c, bc->background_duration, reply_us); |
| 5732 | } |
no test coverage detected