用有界环形队列向 I/O 线程派发任务

3182void readQueryFromClient(connection *conn) {3183    client *c = connGetPrivateData(conn);3184    /* Check if we can send the client to be handled by the IO-thread */3185    if (postponeClientRead(c)) return;31863187    if (c->io_write_state != CLIENT_IDLE || c->io_read_state != CLIENT_IDLE) return;31883189    readToQueryBuf(c);31903191    if (handleReadResult(c) == C_OK) {3192        if (processInputBuffer(c) == C_ERR) return;3193    }3194    beforeNextClient(c);3195}

343    IOJobQueue *jq = &io_jobs[tid];344    if (IOJobQueue_isFull(jq)) return C_ERR;345346    c->cur_tid = tid;347    c->read_flags = canParseCommand(c) ? 0 : READ_FLAGS_DONT_PARSE;348    c->read_flags |= authRequired(c) ? READ_FLAGS_AUTH_REQUIRED : 0;349350    c->io_read_state = CLIENT_PENDING_IO;351    connSetPostponeUpdateState(c->conn, 1);352    IOJobQueue_push(jq, ioThreadReadQueryFromClient, c);353    c->flag.pending_read = 1;354    listLinkNodeTail(server.clients_pending_io_read, &c->pending_read_list_node);355    return C_OK;

44static int IOJobQueue_isFull(const IOJobQueue *jq) {45    debugServerAssertWithInfo(NULL, NULL, inMainThread());46    size_t current_head = atomic_load_explicit(&jq->head, memory_order_relaxed);47    /* We don't use memory_order_acquire for the tail due to performance reasons,48     * In the worst case we will just assume wrongly the buffer is full and the main thread will do the job by itself. */49    size_t current_tail = atomic_load_explicit(&jq->tail, memory_order_relaxed);50    size_t next_head = (current_head + 1) % jq->size;51    return next_head == current_tail;52}

56static void IOJobQueue_push(IOJobQueue *jq, job_handler handler, void *data) {57    debugServerAssertWithInfo(NULL, NULL, inMainThread());58    /* Assert the queue is not full - should not happen as the caller should check for it before. */59    serverAssert(!IOJobQueue_isFull(jq));6061    /* No need to use atomic acquire for the head, as the main thread is the only one that writes to the head index. */62    size_t current_head = atomic_load_explicit(&jq->head, memory_order_relaxed);63    size_t next_head = (current_head + 1) % jq->size;6465    /* We store directly the job's fields to avoid allocating a new iojob structure. */66    serverAssert(jq->ring_buffer[current_head].data == NULL);67    serverAssert(jq->ring_buffer[current_head].handler == NULL);68    jq->ring_buffer[current_head].data = data;69    jq->ring_buffer[current_head].handler = handler;7071    /* memory_order_release to make sure the data is visible to the consumer (the IO thread). */72    atomic_store_explicit(&jq->head, next_head, memory_order_release);73}

79static size_t IOJobQueue_availableJobs(const IOJobQueue *jq) {80    debugServerAssertWithInfo(NULL, NULL, !inMainThread());81    /* We use memory_order_acquire to make sure the head and the job's fields are visible to the consumer (IO thread). */82    size_t current_head = atomic_load_explicit(&jq->head, memory_order_acquire);83    size_t current_tail = atomic_load_explicit(&jq->tail, memory_order_relaxed);8485    if (current_head >= current_tail) {86        return current_head - current_tail;87    } else {88        return jq->size - (current_tail - current_head);89    }90}

243        for (size_t j = 0; j < jobs_to_process; j++) {244            job_handler handler;245            void *data;246            /* We keep the job in the queue until it's processed. This ensures that if the main thread checks247             * and finds the queue empty, it can be certain that the IO thread is not currently handling any job. */248            IOJobQueue_peek(jq, &handler, &data);249            handler(data);250            /* Remove the job after it was processed */251            IOJobQueue_removeJob(jq);252        }253        /* Memory barrier to make sure the main thread sees the updated tail index.254         * We do it once per loop and not per tail-update for optimization reasons.255         * As the main-thread main concern is to check if the queue is empty, it's enough to do it once at the end. */256        atomic_thread_fence(memory_order_release);