Why multithreading causes issues (summary)

Multithread RDMA crash Multithread RDMA crash (lock-free IO queue model) Both issues relate to the RDMA notification model. For socket/unix, ConnectionType.update_state is NULL, so connUpdateState is essentially a no-op. For RDMA, updateRdmaState calls straight into the data path:

static void updateRdmaState(struct connection *conn) {
	rdma_connection *rdma_conn = (rdma_connection *)conn;
	connRdmaSetRwHandler(conn);
	connRdmaEventHandler(NULL, -1, rdma_conn, 0);
}

And connRdmaEventHandler polls the CQ, and when not postponed loops calling read_handler (handing data already DMA’d into rx.addr up to the protocol layer):

static void connRdmaEventHandler(struct aeEventLoop * el, int fd, void * clientData, int mask) {
        rdma_connection * rdma_conn = (rdma_connection * ) clientData;
        connection * conn = & rdma_conn - > c;
        struct rdma_cm_id * cm_id = rdma_conn - > cm_id;
        RdmaContext * ctx = cm_id - > context;
        int ret = 0;

        UNUSED(el);
        UNUSED(fd);
        UNUSED(mask);

        ret = connRdmaHandleCq(rdma_conn);
        if (ret == C_ERR) {
            conn - > state = CONN_STATE_ERROR;
            return;
        }

        /* uplayer should read all */
        // 认为上层应该读取所有的数据,每次更新状态都可能重入readQueryFromClient.
        while (!(rdma_conn - > flags & RDMA_CONN_FLAG_POSTPONE_UPDATE_STATE) && ctx - > rx.pos < ctx - > rx.offset) {
            if (conn - > read_handler && (callHandler(conn, conn - > read_handler) == C_ERR)) {
                return;
            }
        }

How RDMA + IO multithreading introduces the bug — as a causal chain:

1. After an IO thread finishes a read, the main thread in processClientIOReadsDone does connSetPostponeUpdateState(c->conn, 0) then connUpdateState(c->conn) (still true on current unstable; see citations below).
2. For RDMA, connUpdateState → updateRdmaState → connRdmaEventHandler → connRdmaHandleCq + possibly multiple read_handler calls. So while the main thread finishes “this IO thread just completed a read”, it immediately runs another round of transport-layer read logic.
3. That overlaps with “IO thread just marked this client’s read COMPLETED; main thread is parsing / batching / running the state machine”, which easily yields:
   • Reentrancy: trigger the read path again, try again to hand the same client to an IO thread (trySendReadToIOThreads), or
   • Broken state assumptions: e.g. serverAssert(c->io_read_state == ...) expects IDLE/COMPLETED only, but RDMA already mutated read state again in the same stack frame.
4. TCP rarely does this: update_state is NULL, so processClientIOReadsDone does not pump a new read mid-path; RDMA binds “pump CQ + call read_handler” to connUpdateState, so the same networking path is heavier on RDMA and assertions/reentrancy show up.

# [BUG] RDMA: Assertion ‘c->cmd_queue.len == 0’ failed with high pipelining under new I/O queue model

#

Why does this assertion keep failing?

/* Parse one or more commands from the query buf.
 *
 * This function may be called from the main thread or from the I/O thread. 主线程或者IO线程都可以进行调用.
 *
 * Sets the client's read_flags to indicate the parsing outcome. If multiple
 * commands could be parsed, additional parsed commands are stored in the
 * client's command queue. */
// 这里的意思应该是解析buffer.
void parseInputBuffer(client *c) {
    /* The command queue must be emptied before parsing. */
    serverAssert(c->cmd_queue.len == 0);

Before each parse, the command queue must be empty. You must not start another parseInputBuffer while commands from a previous parse still sit unconsumed by processInputBuffer — otherwise two parse states stack and the queue semantics blur. Main-thread consumption order in processInputBuffer: first consumeCommandQueue as much as possible; only when the queue is empty call parseInputBuffer on remaining querybuf bytes. Main thread keeps parsing inside the loop:

int processInputBuffer(client *c) {
    /* Parse the query buffer and/or execute already parsed commands. */
    while ((c->querybuf && c->qb_pos < sdslen(c->querybuf)) ||
           c->cmd_queue.off < c->cmd_queue.len) {
        if (!canParseCommand(c)) {
            break; 
        }
        // ...
        /* If commands are queued up, pop from the queue first */
        // 也就是当队列为空的时候才会进行调用.
        if (!consumeCommandQueue(c)) {
            parseInputBuffer(c);
            prepareCommandQueue(c);
        }
        // ...

See also:

/* Pops a command from the command queue and sets it as the client's current
* command. Returns true on success and false if the queue was empty. */
static bool consumeCommandQueue(client * c) {
    cmdQueue * queue = & c - > cmd_queue;
    // 这里就是当队列为空的时候才会返回false.
    if (queue - > off >= queue - > len) return false;
    parsedCommand * p = & queue - > cmds[queue - > off++];
    
    /* Combine the command's read flags with the client's read flags. Some read
    * flags describe the client state (AUTH_REQUIRED) while others describe the
    * command parsing outcome (PARSING_COMPLETED). */
    c - > read_flags |= p - > read_flags;
    c - > argc = p - > argc;
    c - > argv = p - > argv;
    c - > argv_len = p - > argv_len;
    c - > argv_len_sum = p - > argv_len_sum;
    c - > net_input_bytes_curr_cmd = p - > input_bytes;
    c - > parsed_cmd = p - > cmd;
    c - > slot = p - > slot;
    if (queue - > off == queue - > len) {
        /* The queue is empty. Don't free it here, because if parsing is done in
        * I/O threads, we want to free it in I/O threads too, to avoid
        * fragmentation. */
        queue - > off = queue - > len = 0;
    }
    return true;
}

The I/O-thread path does not use this loop — it parses directly (next section), so it still owes parseInputBuffer the len == 0 invariant. ioThreadReadQueryFromClient reads into querybuf on the I/O thread, then calls parseInputBuffer directly (no “consume queue first” logic from processInputBuffer):

void ioThreadReadQueryFromClient(client *c) {
    serverAssert(c->io_read_state == CLIENT_PENDING_IO);

    /* Read */
    readToQueryBuf(c);
    // 这里就是直接进行解析的操作.
    parseInputBuffer(c);
    trimCommandQueue(c);
    prepareCommandQueue(c);

Under high pipelining (e.g. -P 256), one parseInputBuffer often fills cmd_queue with many parsed commands (len large) while querybuf may still have an unparsed tail (or later RDMA writes more). The main thread later drains those queue entries via addCommandToBatchAndProcessIfFull / processClientsCommandsBatch.

1. First I/O thread: parseInputBuffer succeeds, cmd_queue.len is e.g. 256; main thread has not started consuming.
2. Main thread processClientIOReadsDone prematurely connUpdateState → nested readQueryFromClient → postponeClientRead succeeds again → schedules a second read job.
3. Second I/O thread enters ioThreadReadQueryFromClient → parseInputBuffer(c) again → entry serverAssert(c->cmd_queue.len == 0), but len is still 256 → assertion fails. So it is not “the parser algorithm is wrong”; it is “another I/O-thread parse was scheduled while cmd_queue was still non-empty”. High pipeline keeps cmd_queue.len != 0 after the first parse, so it reproduces reliably. In short: c->cmd_queue.len == 0 fails because the first I/O-thread parseInputBuffer already queued a large pipeline into cmd_queue, and the main thread in processClientIOReadsDone ran RDMA’s connUpdateState too early, nested readQueryFromClient → trySendReadToIOThreads again, and the second I/O thread called parseInputBuffer with a non-empty queue — violating “queue must be empty before parse”.