Function acceptOnThread

src/networking.cpp:1339–1387

}

void acceptOnThread(connection *conn, int flags, char *cip)
{
    int ielCur = ielFromEventLoop(serverTL->el);
    bool fBootLoad = (g_pserver->loading == LOADING_BOOT);

    int ielTarget = ielCur;
    if (fBootLoad)
    {
        ielTarget = IDX_EVENT_LOOP_MAIN;    // During load only the main thread is active
    }
    else if (g_fTestMode)
    {
        // On test mode we don't want any bunching of clients
        while (cserver.cthreads > 1 && ielTarget == IDX_EVENT_LOOP_MAIN)
            ielTarget = rand() % cserver.cthreads;
    }
    else if (g_pserver->active_client_balancing)
    {
        // Cluster connections are more transient, so its not worth the cost to balance
        //  we can trust that SO_REUSEPORT is doing its job of distributing connections
        ielTarget = g_pserver->cluster_enabled ? ielCur : chooseBestThreadForAccept();
    }

    rgacceptsInFlight[ielTarget].fetch_add(1, std::memory_order_relaxed);
    if (ielTarget != ielCur)
    {
        char *szT = nullptr;
        if (cip != nullptr)
        {
            szT = (char*)zmalloc(NET_IP_STR_LEN, MALLOC_LOCAL);
            memcpy(szT, cip, NET_IP_STR_LEN);
        }
        int res = aePostFunction(g_pserver->rgthreadvar[ielTarget].el, [conn, flags, ielTarget, szT] {
            connMarshalThread(conn);
            acceptCommonHandler(conn,flags,szT,ielTarget);
            rgacceptsInFlight[ielTarget].fetch_sub(1, std::memory_order_relaxed);
            zfree(szT);
        });

        if (res == AE_OK)
            return;
        // If res != AE_OK we can still try to accept on the local thread
    }
    rgacceptsInFlight[ielTarget].fetch_sub(1, std::memory_order_relaxed);

    aeAcquireLock();
    acceptCommonHandler(conn,flags,cip,ielCur);
    aeReleaseLock();
}

void acceptTcpHandler(aeEventLoop *el, int fd, void *privdata, int mask) {
    int cport, cfd, max = MAX_ACCEPTS_PER_CALL;