redis源码之线程模型

#io-threads 1 
# 
# Setting io-threads to 1 will just use the main thread as usually. 
# When I/O threads are enabled, we only use threads for writes, that is 
# to thread the write(2) syscall and transfer the client buffers to the 
# socket. However it is also possible to enable threading of reads and 
# protocol parsing using the following configuration directive, by setting 
# it to yes: 
# 
# io-threads-do-reads no 
# 
# Usually threading reads doesn't help much. 
# 
# NOTE 1: This configuration directive cannot be changed at runtime via 
# CONFIG SET. Aso this feature currently does not work when SSL is 
# enabled. 

/* Initialize the data structures needed for threaded I/O. */ 
void initThreadedIO(void) { 
    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 *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); 

    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); 
    } 
}