Function moduleHandleBlockedClients

app/redis-6.2.6/src/module.c:5565–5658

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

 * When this happens the RedisModuleBlockedClient structure in the queue
 * will have the 'client' field set to NULL. */
void moduleHandleBlockedClients(void) {
    listNode *ln;
    RedisModuleBlockedClient *bc;

    pthread_mutex_lock(&moduleUnblockedClientsMutex);
    /* Here we unblock all the pending clients blocked in modules operations
     * so we can read every pending "awake byte" in the pipe. */
    char buf[1];
    while (read(server.module_blocked_pipe[0],buf,1) == 1);
    while (listLength(moduleUnblockedClients)) {
        ln = listFirst(moduleUnblockedClients);
        bc = ln->value;
        client *c = bc->client;
        listDelNode(moduleUnblockedClients,ln);
        pthread_mutex_unlock(&moduleUnblockedClientsMutex);

        /* Release the lock during the loop, as long as we don't
         * touch the shared list. */

        /* Call the reply callback if the client is valid and we have
         * any callback. However the callback is not called if the client
         * was blocked on keys (RM_BlockClientOnKeys()), because we already
         * called such callback in moduleTryServeClientBlockedOnKey() when
         * the key was signaled as ready. */
        uint64_t reply_us = 0;
        if (c && !bc->blocked_on_keys && bc->reply_callback) {
            RedisModuleCtx ctx = REDISMODULE_CTX_INIT;
            ctx.flags |= REDISMODULE_CTX_BLOCKED_REPLY;
            ctx.blocked_privdata = bc->privdata;
            ctx.blocked_ready_key = NULL;
            ctx.module = bc->module;
            ctx.client = bc->client;
            ctx.blocked_client = bc;
            monotime replyTimer;
            elapsedStart(&replyTimer);
            bc->reply_callback(&ctx,(void**)c->argv,c->argc);
            reply_us = elapsedUs(replyTimer);
            moduleFreeContext(&ctx);
        }
        /* Update stats now that we've finished the blocking operation.
         * This needs to be out of the reply callback above given that a
         * module might not define any callback and still do blocking ops.
         */
        if (c && !bc->blocked_on_keys) {
            updateStatsOnUnblock(c, bc->background_duration, reply_us);
        }

        /* Free privdata if any. */
        if (bc->privdata && bc->free_privdata) {
            RedisModuleCtx ctx = REDISMODULE_CTX_INIT;
            if (c == NULL)
                ctx.flags |= REDISMODULE_CTX_BLOCKED_DISCONNECTED;
            ctx.blocked_privdata = bc->privdata;
            ctx.module = bc->module;
            ctx.client = bc->client;
            bc->free_privdata(&ctx,bc->privdata);
            moduleFreeContext(&ctx);
        }