Method WorkerThreadMain

src/AsyncWorkQueue.cpp:14–62

}

void AsyncWorkQueue::WorkerThreadMain()
{
    redisServerThreadVars vars;
    serverTL = &vars;

    vars.clients_pending_asyncwrite = listCreate();

    m_mutex.lock();
    m_vecpthreadVars.push_back(&vars);
    m_mutex.unlock();

    while (!m_fQuitting)
    {
        std::unique_lock<std::mutex> lock(m_mutex);
        if (m_workqueue.empty())
            m_cvWakeup.wait(lock);

        aeThreadOnline();
        while (!m_workqueue.empty())
        {
            WorkItem task = std::move(m_workqueue.front());
            m_workqueue.pop_front();

            lock.unlock();
            serverTL->gcEpoch = g_pserver->garbageCollector.startEpoch();
            task.fnAsync();
            g_pserver->garbageCollector.endEpoch(serverTL->gcEpoch);
            lock.lock();
        }

        lock.unlock();
        serverTL->gcEpoch = g_pserver->garbageCollector.startEpoch();
        if (listLength(serverTL->clients_pending_asyncwrite)) {
            aeAcquireLock();
            ProcessPendingAsyncWrites();
            aeReleaseLock();
        }
        g_pserver->garbageCollector.endEpoch(serverTL->gcEpoch);
        serverTL->gcEpoch.reset();
        aeThreadOffline();
    }

    listRelease(vars.clients_pending_asyncwrite);

    std::unique_lock<fastlock> lockf(serverTL->lockPendingWrite);
    serverTL->vecclientsProcess.clear();
    serverTL->clients_pending_write.clear();
    std::atomic_thread_fence(std::memory_order_seq_cst);
}

bool AsyncWorkQueue::removeClientAsyncWrites(client *c)
{