Redis单线程不行了，快来割VM/ BIO/ IO多线程的韭菜！(附源码）

/* Global server state structure */struct redisServer {...    list *io_newjobs; /* List of VM I/O jobs yet to be processed */    list *io_processing; /* List of VM I/O jobs being processed */    list *io_processed; /* List of VM I/O jobs already processed */    list *io_ready_clients; /* Clients ready to be unblocked. All keys loaded */    pthread_mutex_t io_mutex; /* lock to access io_jobs/io_done/io_thread_job */    pthread_mutex_t io_swapfile_mutex; /* So we can lseek + write */    pthread_attr_t io_threads_attr; /* attributes for threads creation */...}

void queueIOJob(iojob *j) {    redisLog(REDIS_DEBUG,"Queued IO Job %p type %d about key '%s'\n",        (void*)j, j->type, (char*)j->key->ptr);    listAddNodeTail(server.io_newjobs,j);    if (server.io_active_threads < server.vm_max_threads)        spawnIOThread();}

typedef struct iojob {    int type;   /* Request type, REDIS_IOJOB_* */    redisDb *db;/* Redis database */    robj *key;  /* This I/O request is about swapping this key */    robj *id;   /* Unique identifier of this job:                   this is the object to swap for REDIS_IOREQ_*_SWAP, or the                   vmpointer objct for REDIS_IOREQ_LOAD. */    robj *val;  /* the value to swap for REDIS_IOREQ_*_SWAP, otherwise this                 * field is populated by the I/O thread for REDIS_IOREQ_LOAD. */    off_t page; /* Swap page where to read/write the object */    off_t pages; /* Swap pages needed to save object. PREPARE_SWAP return val */    int canceled; /* True if this command was canceled by blocking side of VM */    pthread_t thread; /* ID of the thread processing this entry */} iojob;#define REDIS_IOJOB_LOAD 0          /* Load from disk to memory */#define REDIS_IOJOB_PREPARE_SWAP 1  /* Compute needed pages */#define REDIS_IOJOB_DO_SWAP 2       /* Swap from memory to disk */void *IOThreadEntryPoint(void *arg) {    iojob *j;    listNode *ln;    REDIS_NOTUSED(arg);    pthread_detach(pthread_self());    while(1) {        /* Get a new job to process */        lockThreadedIO();        if (listLength(server.io_newjobs) == 0) {            /* No new jobs in queue, exit. */            ...                        unlockThreadedIO();            return NULL;        }                ln = listFirst(server.io_newjobs);        j = ln->value;        listDelNode(server.io_newjobs,ln);        /* Add the job in the processing queue */                j->thread = pthread_self();        listAddNodeTail(server.io_processing,j);        ln = listLast(server.io_processing); /* We use ln later to remove it */        unlockThreadedIO();                ...        /* Process the Job */        if (j->type == REDIS_IOJOB_LOAD) {            vmpointer *vp = (vmpointer*)j->id;            j->val = vmReadObjectFromSwap(j->page,vp->vtype);        } else if (j->type == REDIS_IOJOB_PREPARE_SWAP) {            j->pages = rdbSavedObjectPages(j->val);        } else if (j->type == REDIS_IOJOB_DO_SWAP) {            if (vmWriteObjectOnSwap(j->val,j->page) == REDIS_ERR)                j->canceled = 1;        }        /* Done: insert the job into the processed queue */        ...                lockThreadedIO();        listDelNode(server.io_processing,ln);        listAddNodeTail(server.io_processed,j);        unlockThreadedIO();        /* Signal the main thread there is new stuff to process */        redisAssert(write(server.io_ready_pipe_write,"x",1) == 1);    }    return NULL; /* never reached */}

// config.c...    else if (!strcasecmp(c->argv[2]->ptr,"appendfsync")) {        if (!strcasecmp(o->ptr,"no")) {            server.appendfsync = APPENDFSYNC_NO;        } else if (!strcasecmp(o->ptr,"everysec")) {            server.appendfsync = APPENDFSYNC_EVERYSEC;        } else if (!strcasecmp(o->ptr,"always")) {            server.appendfsync = APPENDFSYNC_ALWAYS;        } else {            goto badfmt;        }    }...

// aof.cvoid backgroundRewriteDoneHandler(int statloc) {......    else if (server.appendfsync == APPENDFSYNC_EVERYSEC)        aof_background_fsync(newfd);......}在 aof_background_fsync 函数中，发现了后台任务相关函数：// aof.cvoid aof_background_fsync(int fd) {    bioCreateBackgroundJob(REDIS_BIO_AOF_FSYNC,(void*)(long)fd,NULL,NULL);}

// bio.h/* Exported API */void bioInit(void);void bioCreateBackgroundJob(int type, void *arg1, void *arg2, void *arg3);unsigned long long bioPendingJobsOfType(int type);void bioWaitPendingJobsLE(int type, unsigned long long num);time_t bioOlderJobOfType(int type);/* Background job opcodes */#define REDIS_BIO_CLOSE_FILE    0 /* Deferred close(2) syscall. */#define REDIS_BIO_AOF_FSYNC     1 /* Deferred AOF fsync. */#define REDIS_BIO_NUM_OPS       2

// bio.c/* Initialize the background system, spawning the thread. */void bioInit(void) {    pthread_attr_t attr;    pthread_t thread;    size_t stacksize;    int j;    /* Initialization of state vars and objects */    for (j = 0; j < REDIS_BIO_NUM_OPS; j++) {        pthread_mutex_init(&bio_mutex[j],NULL);        pthread_cond_init(&bio_condvar[j],NULL);        bio_jobs[j] = listCreate();        bio_pending[j] = 0;    }    /* Set the stack size as by default it may be small in some system */    pthread_attr_init(&attr);    pthread_attr_getstacksize(&attr,&stacksize);    if (!stacksize) stacksize = 1; /* The world is full of Solaris Fixes */    while (stacksize < REDIS_THREAD_STACK_SIZE) stacksize *= 2;    pthread_attr_setstacksize(&attr, stacksize);    /* Ready to spawn our threads. We use the single argument the thread     * function accepts in order to pass the job ID the thread is     * responsible of. */    for (j = 0; j < REDIS_BIO_NUM_OPS; j++) {        void *arg = (void*)(unsigned long) j;        if (pthread_create(&thread,&attr,bioProcessBackgroundJobs,arg) != 0) {            redisLog(REDIS_WARNING,"Fatal: Can't initialize Background Jobs.");            exit(1);        }    }}

/*
* 提示：该行代码过长，系统自动注释不进行高亮。一键复制会移除系统注释 
* //aof.c/* A background append only file rewriting (BGREWRITEAOF) terminated its work. * Handle this. */void backgroundRewriteDoneHandler(int statloc) {    int exitcode = WEXITSTATUS(statloc);    int bysignal = WIFSIGNALED(statloc);    if (!bysignal && exitcode == 0) {        int newfd, oldfd;        int nwritten;        char tmpfile[256];        long long now = ustime();                ...        /* Flush the differences accumulated by the parent to the         * rewritten AOF. */        snprintf(tmpfile,256,"temp-rewriteaof-bg-%d.aof",            (int)server.bgrewritechildpid);        newfd = open(tmpfile,O_WRONLY|O_APPEND);        if (newfd == -1) {            redisLog(REDIS_WARNING,                "Unable to open the temporary AOF produced by the child: %s", strerror(errno));            goto cleanup;        }        nwritten = write(newfd,server.bgrewritebuf,sdslen(server.bgrewritebuf));        if (nwritten != (signed)sdslen(server.bgrewritebuf)) {            if (nwritten == -1) {                redisLog(REDIS_WARNING,                    "Error trying to flush the parent diff to the rewritten AOF: %s", strerror(errno));            } else {                redisLog(REDIS_WARNING,                    "Short write trying to flush the parent diff to the rewritten AOF: %s", strerror(errno));            }            close(newfd);            goto cleanup;        }        redisLog(REDIS_NOTICE,            "Parent diff successfully flushed to the rewritten AOF (%lu bytes)", nwritten);        /* The only remaining thing to do is to rename the temporary file to         * the configured file and switch the file descriptor used to do AOF         * writes. We don't want close(2) or rename(2) calls to block the         * server on old file deletion.         *         * There are two possible scenarios:         *         * 1) AOF is DISABLED and this was a one time rewrite. The temporary         * file will be renamed to the configured file. When this file already         * exists, it will be unlinked, which may block the server.         *         * 2) AOF is ENABLED and the rewritten AOF will immediately start         * receiving writes. After the temporary file is renamed to the         * configured file, the original AOF file descriptor will be closed.         * Since this will be the last reference to that file, closing it         * causes the underlying file to be unlinked, which may block the         * server.         *         * To mitigate the blocking effect of the unlink operation (either         * caused by rename(2) in scenario 1, or by close(2) in scenario 2), we         * use a background thread to take care of this. First, we         * make scenario 1 identical to scenario 2 by opening the target file         * when it exists. The unlink operation after the rename(2) will then         * be executed upon calling close(2) for its descriptor. Everything to         * guarantee atomicity for this switch has already happened by then, so         * we don't care what the outcome or duration of that close operation         * is, as long as the file descriptor is released again. */        if (server.appendfd == -1) {            /* AOF disabled */             /* Don't care if this fails: oldfd will be -1 and we handle that.              * One notable case of -1 return is if the old file does              * not exist. */             oldfd = open(server.appendfilename,O_RDONLY|O_NONBLOCK);        } else {            /* AOF enabled */            oldfd = -1; /* We'll set this to the current AOF filedes later. */        }        /* Rename the temporary file. This will not unlink the target file if         * it exists, because we reference it with "oldfd". */        if (rename(tmpfile,server.appendfilename) == -1) {            redisLog(REDIS_WARNING,                "Error trying to rename the temporary AOF: %s", strerror(errno));            close(newfd);            if (oldfd != -1) close(oldfd);            goto cleanup;        }        if (server.appendfd == -1) {            /* AOF disabled, we don't need to set the AOF file descriptor             * to this new file, so we can close it. */            close(newfd);        } else {            /* AOF enabled, replace the old fd with the new one. */            oldfd = server.appendfd;            server.appendfd = newfd;            if (server.appendfsync == APPENDFSYNC_ALWAYS)                aof_fsync(newfd);            else if (server.appendfsync == APPENDFSYNC_EVERYSEC)                aof_background_fsync(newfd);            server.appendseldb = -1; /* Make sure SELECT is re-issued */            aofUpdateCurrentSize();            server.auto_aofrewrite_base_size = server.appendonly_current_size;            /* Clear regular AOF buffer since its contents was just written to             * the new AOF from the background rewrite buffer. */            sdsfree(server.aofbuf);            server.aofbuf = sdsempty();        }        redisLog(REDIS_NOTICE, "Background AOF rewrite successful");        /* Asynchronously close the overwritten AOF. */        if (oldfd != -1) bioCreateBackgroundJob(REDIS_BIO_CLOSE_FILE,(void*)(long)oldfd,NULL,NULL);        redisLog(REDIS_VERBOSE,            "Background AOF rewrite signal handler took %lldus", ustime()-now);    } else if (!bysignal && exitcode != 0) {        redisLog(REDIS_WARNING,            "Background AOF rewrite terminated with error");    } else {        redisLog(REDIS_WARNING,            "Background AOF rewrite terminated by signal %d",            WTERMSIG(statloc));    }cleanup:    sdsfree(server.bgrewritebuf);    server.bgrewritebuf = sdsempty();    aofRemoveTempFile(server.bgrewritechildpid);    server.bgrewritechildpid = -1;}
*/

void aof_background_fsync(int fd) {    bioCreateBackgroundJob(REDIS_BIO_AOF_FSYNC,(void*)(long)fd,NULL,NULL);}

// bio.cvoid bioCreateBackgroundJob(int type, void *arg1, void *arg2, void *arg3) {    struct bio_job *job = zmalloc(sizeof(*job));    job->time = time(NULL);    job->arg1 = arg1;    job->arg2 = arg2;    job->arg3 = arg3;    pthread_mutex_lock(&bio_mutex[type]);    listAddNodeTail(bio_jobs[type],job);    bio_pending[type]++;    pthread_cond_signal(&bio_condvar[type]);    pthread_mutex_unlock(&bio_mutex[type]);}

// bio.cvoid *bioProcessBackgroundJobs(void *arg) {    struct bio_job *job;    unsigned long type = (unsigned long) arg;    pthread_detach(pthread_self());    pthread_mutex_lock(&bio_mutex[type]);    while(1) {        listNode *ln;        /* The loop always starts with the lock hold. */        if (listLength(bio_jobs[type]) == 0) {            pthread_cond_wait(&bio_condvar[type],&bio_mutex[type]);            continue;        }        /* Pop the job from the queue. */        ln = listFirst(bio_jobs[type]);        job = ln->value;        /* It is now possible to unlock the background system as we know have         * a stand alone job structure to process.*/        pthread_mutex_unlock(&bio_mutex[type]);        /* Process the job accordingly to its type. */        if (type == REDIS_BIO_CLOSE_FILE) {            close((long)job->arg1);        } else if (type == REDIS_BIO_AOF_FSYNC) {            aof_fsync((long)job->arg1);        } else {            redisPanic("Wrong job type in bioProcessBackgroundJobs().");        }        zfree(job);        /* Lock again before reiterating the loop, if there are no longer         * jobs to process we'll block again in pthread_cond_wait(). */        pthread_mutex_lock(&bio_mutex[type]);        listDelNode(bio_jobs[type],ln);        bio_pending[type]--;    }}

我们继续看 aof_fsync 的函数定义，发现 aof_fsync 其实就是 fdatasync 和 fsync :

/* Define aof_fsync to fdatasync() in Linux and fsync() for all the rest */#ifdef __linux__#define aof_fsync fdatasync#else#define aof_fsync fsync#endif

// bio.h/* Background job opcodes */#define BIO_CLOSE_FILE    0 /* Deferred close(2) syscall. */#define BIO_AOF_FSYNC     1 /* Deferred AOF fsync. */#define BIO_LAZY_FREE     2 /* Deferred objects freeing. */#define BIO_NUM_OPS       3

// bio.cvoid *bioProcessBackgroundJobs(void *arg) {    struct bio_job *job;    unsigned long type = (unsigned long) arg;    sigset_t sigset;    /* Check that the type is within the right interval. */    if (type >= BIO_NUM_OPS) {        serverLog(LL_WARNING,            "Warning: bio thread started with wrong type %lu",type);        return NULL;    }    /* Make the thread killable at any time, so that bioKillThreads()     * can work reliably. */    pthread_setcancelstate(PTHREAD_CANCEL_ENABLE, NULL);    pthread_setcanceltype(PTHREAD_CANCEL_ASYNCHRONOUS, NULL);    pthread_mutex_lock(&bio_mutex[type]);    /* Block SIGALRM so we are sure that only the main thread will     * receive the watchdog signal. */    sigemptyset(&sigset);    sigaddset(&sigset, SIGALRM);    if (pthread_sigmask(SIG_BLOCK, &sigset, NULL))        serverLog(LL_WARNING,            "Warning: can't mask SIGALRM in bio.c thread: %s", strerror(errno));    while(1) {        listNode *ln;        /* The loop always starts with the lock hold. */        if (listLength(bio_jobs[type]) == 0) {            pthread_cond_wait(&bio_newjob_cond[type],&bio_mutex[type]);            continue;        }        /* Pop the job from the queue. */        ln = listFirst(bio_jobs[type]);        job = ln->value;        /* It is now possible to unlock the background system as we know have         * a stand alone job structure to process.*/        pthread_mutex_unlock(&bio_mutex[type]);        /* Process the job accordingly to its type. */        if (type == BIO_CLOSE_FILE) {            close((long)job->arg1);        } else if (type == BIO_AOF_FSYNC) {            aof_fsync((long)job->arg1);        } else if (type == BIO_LAZY_FREE) {            /* What we free changes depending on what arguments are set:             * arg1 -> free the object at pointer.             * arg2 & arg3 -> free two dictionaries (a Redis DB).             * only arg3 -> free the skiplist. */            if (job->arg1)                lazyfreeFreeObjectFromBioThread(job->arg1);            else if (job->arg2 && job->arg3)                lazyfreeFreeDatabaseFromBioThread(job->arg2,job->arg3);            else if (job->arg3)                lazyfreeFreeSlotsMapFromBioThread(job->arg3);        } else {            serverPanic("Wrong job type in bioProcessBackgroundJobs().");        }        zfree(job);        /* Unblock threads blocked on bioWaitStepOfType() if any. */        pthread_cond_broadcast(&bio_step_cond[type]);        /* Lock again before reiterating the loop, if there are no longer         * jobs to process we'll block again in pthread_cond_wait(). */        pthread_mutex_lock(&bio_mutex[type]);        listDelNode(bio_jobs[type],ln);        bio_pending[type]--;    }}

// lazyfree.c/* Release objects from the lazyfree thread. It's just decrRefCount() * updating the count of objects to release. */void lazyfreeFreeObjectFromBioThread(robj *o) {    decrRefCount(o);    atomicDecr(lazyfree_objects,1);}

// object.cvoid decrRefCount(robj *o) {    if (o->refcount == 1) {        switch(o->type) {        case OBJ_STRING: freeStringObject(o); break;        case OBJ_LIST: freeListObject(o); break;        case OBJ_SET: freeSetObject(o); break;        case OBJ_ZSET: freeZsetObject(o); break;        case OBJ_HASH: freeHashObject(o); break;        case OBJ_MODULE: freeModuleObject(o); break;        default: serverPanic("Unknown object type"); break;        }        zfree(o);    } else {        if (o->refcount <= 0) serverPanic("decrRefCount against refcount <= 0");        if (o->refcount != OBJ_SHARED_REFCOUNT) o->refcount--;    }}

// server.c#define OBJ_SHARED_REFCOUNT INT_MAXtypedef struct redisObject {    unsigned type:4;    unsigned encoding:4;    unsigned lru:LRU_BITS; /* LRU time (relative to global lru_clock) or                            * LFU data (least significant 8 bits frequency                            * and most significant 16 bits access time). */    int refcount;    void *ptr;} robj;// server.cvoid createSharedObjects(void) {......    for (j = 0; j < OBJ_SHARED_INTEGERS; j++) {        shared.integers[j] =            makeObjectShared(createObject(OBJ_STRING,(void*)(long)j));        shared.integers[j]->encoding = OBJ_ENCODING_INT;    }......}

int main(int argc, char **argv) {    struct timeval tv;    int j;    server.supervised = redisIsSupervised(server.supervised_mode);    int background = server.daemonize && !server.supervised;    if (background) daemonize();    ......some log......    readOOMScoreAdj();    initServer();    if (background || server.pidfile) createPidFile();    redisSetProcTitle(argv[0]);    redisAsciiArt();    checkTcpBacklogSettings();    if (!server.sentinel_mode) {        moduleLoadFromQueue();        ACLLoadUsersAtStartup();        InitServerLast();        loadDataFromDisk();        ......    } else {        InitServerLast();        sentinelIsRunning();        ......    }    ......    redisSetCpuAffinity(server.server_cpulist);    setOOMScoreAdj(-1);    aeMain(server.el);    aeDeleteEventLoop(server.el);    return 0;}

void initServer(void) {    ......    /* Initialization after setting defaults from the config system. */    server.aof_state = server.aof_enabled ? AOF_ON : AOF_OFF;    server.hz = server.config_hz;    server.pid = getpid();    server.in_fork_child = CHILD_TYPE_NONE;    server.main_thread_id = pthread_self();    server.current_client = NULL; // 当前正在执行命令的客户端    server.errors = raxNew();    server.fixed_time_expire = 0;    server.clients = listCreate(); // 活跃的客户端列表    server.clients_index = raxNew(); // 按照 client_id 索引的活跃的客户端字典    server.clients_to_close = listCreate(); // 需要异步关闭的客户端列表    server.slaves = listCreate();    server.monitors = listCreate();    server.clients_pending_write = listCreate(); // 等待写或者安装handler的客户端列表    server.clients_pending_read = listCreate(); // 等待读socket缓冲区的客户端列表    server.clients_timeout_table = raxNew();    server.replication_allowed = 1;    server.slaveseldb = -1; /* Force to emit the first SELECT command. */    server.unblocked_clients = listCreate(); // 下一个循环之前，要取消阻塞的客户端列表    server.ready_keys = listCreate();    server.clients_waiting_acks = listCreate();    server.get_ack_from_slaves = 0;    server.client_pause_type = 0;    server.paused_clients = listCreate();    server.events_processed_while_blocked = 0;    server.system_memory_size = zmalloc_get_memory_size();    server.blocked_last_cron = 0;    server.blocking_op_nesting = 0;    ......}在 InitServerLast()中 ，除了 initThreadedIO (Redis网络IO线程)，我们还能看到bioInit(background I/O 初始化)，两个模块使用了不同的资源：/* Some steps in server initialization need to be done last (after modules * are loaded). * Specifically, creation of threads due to a race bug in ld.so, in which * Thread Local Storage initialization collides with dlopen call. * see: https://sourceware.org/bugzilla/show_bug.cgi?id=19329 */void InitServerLast() {    bioInit();    initThreadedIO();    set_jemalloc_bg_thread(server.jemalloc_bg_thread);    server.initial_memory_usage = zmalloc_used_memory();}

/* ========================================================================== * Threaded I/O * ========================================================================== */#define IO_THREADS_MAX_NUM 128#define IO_THREADS_OP_READ 0#define IO_THREADS_OP_WRITE 1pthread_t io_threads[IO_THREADS_MAX_NUM];pthread_mutex_t io_threads_mutex[IO_THREADS_MAX_NUM];redisAtomic unsigned long io_threads_pending[IO_THREADS_MAX_NUM];int io_threads_op;      /* IO_THREADS_OP_WRITE or IO_THREADS_OP_READ. *//* This is the list of clients each thread will serve when threaded I/O is * used. We spawn io_threads_num-1 threads, since one is the main thread * itself. */list *io_threads_list[IO_THREADS_MAX_NUM];

/* Initialize the data structures needed for threaded I/O. */void initThreadedIO(void) {    server.io_threads_active = 0; /* We start with threads not active. */    /* Don't spawn any thread if the user selected a single thread:     * we'll handle I/O directly from the main thread. */    if (server.io_threads_num == 1) return;    if (server.io_threads_num > IO_THREADS_MAX_NUM) {        serverLog(LL_WARNING,"Fatal: too many I/O threads configured. "                             "The maximum number is %d.", IO_THREADS_MAX_NUM);        exit(1);    }    /* Spawn and initialize the I/O threads. */    for (int i = 0; i < server.io_threads_num; i++) {        /* Things we do for all the threads including the main thread. */        io_threads_list[i] = listCreate();        if (i == 0) continue; /* Thread 0 is the main thread. */        /* Things we do only for the additional threads. */        pthread_t tid;        pthread_mutex_init(&io_threads_mutex[i],NULL);        io_threads_pending[i] = 0;        pthread_mutex_lock(&io_threads_mutex[i]); /* Thread will be stopped. */        if (pthread_create(&tid,NULL,IOThreadMain,(void*)(long)i) != 0) {            serverLog(LL_WARNING,"Fatal: Can't initialize IO thread.");            exit(1);        }        io_threads[i] = tid;    }}

void initServer(void) {......    server.el = aeCreateEventLoop(server.maxclients+CONFIG_FDSET_INCR);......    /* Register before and after sleep handlers (note this needs to be done     * before loading persistence since it is used by processEventsWhileBlocked. */    aeSetBeforeSleepProc(server.el,beforeSleep);    aeSetAfterSleepProc(server.el,afterSleep);......}

/* This function gets called every time Redis is entering the * main loop of the event driven library, that is, before to sleep * for ready file descriptors. * * Note: This function is (currently) called from two functions: * 1. aeMain - The main server loop * 2. processEventsWhileBlocked - Process clients during RDB/AOF load * * If it was called from processEventsWhileBlocked we don't want * to perform all actions (For example, we don't want to expire * keys), but we do need to perform some actions. * * The most important is freeClientsInAsyncFreeQueue but we also * call some other low-risk functions. */void beforeSleep(struct aeEventLoop *eventLoop) {......    /* Just call a subset of vital functions in case we are re-entering     * the event loop from processEventsWhileBlocked(). Note that in this     * case we keep track of the number of events we are processing, since     * processEventsWhileBlocked() wants to stop ASAP if there are no longer     * events to handle. */    if (ProcessingEventsWhileBlocked) {        uint64_t processed = 0;        processed += handleClientsWithPendingReadsUsingThreads();        processed += tlsProcessPendingData();        processed += handleClientsWithPendingWrites();        processed += freeClientsInAsyncFreeQueue();        server.events_processed_while_blocked += processed;        return;    }......    /* We should handle pending reads clients ASAP after event loop. */    handleClientsWithPendingReadsUsingThreads();......    /* Handle writes with pending output buffers. */    handleClientsWithPendingWritesUsingThreads();    /* Close clients that need to be closed asynchronous */    freeClientsInAsyncFreeQueue();......    /* Before we are going to sleep, let the threads access the dataset by     * releasing the GIL. Redis main thread will not touch anything at this     * time. */    if (moduleCount()) moduleReleaseGIL();    /* Do NOT add anything below moduleReleaseGIL !!! */}

/* When threaded I/O is also enabled for the reading + parsing side, the * readable handler will just put normal clients into a queue of clients to * process (instead of serving them synchronously). This function runs * the queue using the I/O threads, and process them in order to accumulate * the reads in the buffers, and also parse the first command available * rendering it in the client structures. */int handleClientsWithPendingReadsUsingThreads(void) {    // IO线程在初始化时未激活，Redis配置了用IO线程读取和解析数据(io_threads_do_reads)，才会继续执行    if (!server.io_threads_active || !server.io_threads_do_reads) return 0;    int processed = listLength(server.clients_pending_read);    if (processed == 0) return 0;    /* Distribute the clients across N different lists. */    // 读取待处理的客户端列表 clients_pending_read，    // 将任务按照取模平均分配到不同线程的任务队列io_threads_list[target_id]    listIter li;    listNode *ln;    listRewind(server.clients_pending_read,&li);    int item_id = 0;    while((ln = listNext(&li))) {        client *c = listNodeValue(ln);        int target_id = item_id % server.io_threads_num;        listAddNodeTail(io_threads_list[target_id],c);        item_id++;    }    /* Give the start condition to the waiting threads, by setting the     * start condition atomic var. */    // 通过setIOPendingCount给对应的IO线程设置条件变量，激活IO线程    io_threads_op = IO_THREADS_OP_READ;    for (int j = 1; j < server.io_threads_num; j++) {        int count = listLength(io_threads_list[j]);        setIOPendingCount(j, count);    }    /* Also use the main thread to process a slice of clients. */    // 依然在主线程处理一些客户端请求    listRewind(io_threads_list[0],&li);    while((ln = listNext(&li))) {        client *c = listNodeValue(ln);        readQueryFromClient(c->conn);    }    listEmpty(io_threads_list[0]);    /* Wait for all the other threads to end their work. */    while(1) {        unsigned long pending = 0;        for (int j = 1; j < server.io_threads_num; j++)            pending += getIOPendingCount(j);        if (pending == 0) break;    }    /* Run the list of clients again to process the new buffers. */    while(listLength(server.clients_pending_read)) {        ln = listFirst(server.clients_pending_read);        client *c = listNodeValue(ln);        c->flags &= ~CLIENT_PENDING_READ;        listDelNode(server.clients_pending_read,ln);        if (processPendingCommandsAndResetClient(c) == C_ERR) {            /* If the client is no longer valid, we avoid             * processing the client later. So we just go             * to the next. */            continue;        }        processInputBuffer(c);        /* We may have pending replies if a thread readQueryFromClient() produced         * replies and did not install a write handler (it can't).         */        // 如果客户端等待写入，        // 并且响应的buffer还有待写数据，或有待发送给客户端的响应对象，        // 则给客户端的连接安装写handler        if (!(c->flags & CLIENT_PENDING_WRITE) && clientHasPendingReplies(c))            clientInstallWriteHandler(c);    }    /* Update processed count on server */    server.stat_io_reads_processed += processed;    return processed;}

int handleClientsWithPendingWritesUsingThreads(void) {    int processed = listLength(server.clients_pending_write);    if (processed == 0) return 0; /* Return ASAP if there are no clients. */    /* If I/O threads are disabled or we have few clients to serve, don't     * use I/O threads, but the boring synchronous code. */    // 如果没有开启多线程，或者等待的客户端数量小于线程数的两倍，则执行同步代码    if (server.io_threads_num == 1 || stopThreadedIOIfNeeded()) {        return handleClientsWithPendingWrites();    }    /* Start threads if needed. */    // 如果 IO 线程没有激活，则激活（在initThreadedIO函数创建线程时处于未激活状态）    if (!server.io_threads_active) startThreadedIO();    /* Distribute the clients across N different lists. */    listIter li;    listNode *ln;    listRewind(server.clients_pending_write,&li);    int item_id = 0;    while((ln = listNext(&li))) {        client *c = listNodeValue(ln);        c->flags &= ~CLIENT_PENDING_WRITE;        /* Remove clients from the list of pending writes since         * they are going to be closed ASAP. */        // 如果遇到需要关闭的客户端(CLIENT_CLOSE_ASAP)，则将其从待处理的客户端列表里删除        if (c->flags & CLIENT_CLOSE_ASAP) {            listDelNode(server.clients_pending_write, ln);            continue;        }        int target_id = item_id % server.io_threads_num;        listAddNodeTail(io_threads_list[target_id],c);        item_id++;    }    /* Give the start condition to the waiting threads, by setting the     * start condition atomic var. */    // 通过setIOPendingCount给对应的IO线程设置条件变量，激活IO线程    io_threads_op = IO_THREADS_OP_WRITE;    for (int j = 1; j < server.io_threads_num; j++) {        int count = listLength(io_threads_list[j]);        setIOPendingCount(j, count);    }        /* Also use the main thread to process a slice of clients. */    // 依然在主线程处理一些客户端请求    listRewind(io_threads_list[0],&li);    while((ln = listNext(&li))) {        client *c = listNodeValue(ln);        writeToClient(c,0);    }    listEmpty(io_threads_list[0]);    /* Wait for all the other threads to end their work. */    while(1) {        unsigned long pending = 0;        for (int j = 1; j < server.io_threads_num; j++)            pending += getIOPendingCount(j);        if (pending == 0) break;    }    /* Run the list of clients again to install the write handler where     * needed. */    listRewind(server.clients_pending_write,&li);    while((ln = listNext(&li))) {        client *c = listNodeValue(ln);        /* Install the write handler if there are pending writes in some         * of the clients. */        // 如果响应的buffer还有待写数据，或者还有待发送给客户端的响应对象，        // 则给客户端的连接安装写handler        if (clientHasPendingReplies(c) &&                connSetWriteHandler(c->conn, sendReplyToClient) == AE_ERR)        {            // 将待释放的客户端放入clients_to_close队列，            // 等待beforeSleep执行freeClientsInAsyncFreeQueue时实现异步释放客户端            freeClientAsync(c);        }    }    listEmpty(server.clients_pending_write);    /* Update processed count on server */    server.stat_io_writes_processed += processed;    return processed;}

void *IOThreadMain(void *myid) {    /* The ID is the thread number (from 0 to server.iothreads_num-1), and is     * used by the thread to just manipulate a single sub-array of clients. */    long id = (unsigned long)myid;    char thdname[16];    snprintf(thdname, sizeof(thdname), "io_thd_%ld", id);    redis_set_thread_title(thdname);    redisSetCpuAffinity(server.server_cpulist);    makeThreadKillable();    while(1) {        /* Wait for start */        for (int j = 0; j < 1000000; j++) {            if (io_threads_pending[id] != 0) break;        }        /* Give the main thread a chance to stop this thread. */        if (io_threads_pending[id] == 0) {            pthread_mutex_lock(&io_threads_mutex[id]);            pthread_mutex_unlock(&io_threads_mutex[id]);            continue;        }        serverAssert(io_threads_pending[id] != 0);        if (tio_debug) printf("[%ld] %d to handle\n", id, (int)listLength(io_threads_list[id]));        /* Process: note that the main thread will never touch our list         * before we drop the pending count to 0. */        listIter li;        listNode *ln;        listRewind(io_threads_list[id],&li);        while((ln = listNext(&li))) {            client *c = listNodeValue(ln);            if (io_threads_op == IO_THREADS_OP_WRITE) {                writeToClient(c,0);            } else if (io_threads_op == IO_THREADS_OP_READ) {                readQueryFromClient(c->conn);            } else {                serverPanic("io_threads_op value is unknown");            }        }        listEmpty(io_threads_list[id]);        io_threads_pending[id] = 0;        if (tio_debug) printf("[%ld] Done\n", id);    }}