Method loadWorkerThreadMain

src/rdb.cpp:2976–3039

    }

    static void loadWorkerThreadMain(rdbAsyncWorkThread *pqueue) {
        rdbAsyncWorkThread &queue = *pqueue;
        redisServerThreadVars vars = {};
        vars.clients_pending_asyncwrite = listCreate();
        serverTL = &vars;
        aeSetThreadOwnsLockOverride(true);

#ifdef __linux__
        // We will inheret the server thread's affinity mask, clear it as we want to run on a different core.
        cpu_set_t *cpuset = CPU_ALLOC(std::thread::hardware_concurrency());
        if (cpuset != nullptr) {
            size_t size = CPU_ALLOC_SIZE(std::thread::hardware_concurrency());
            CPU_ZERO_S(size, cpuset);
            for (unsigned i = 0; i < std::thread::hardware_concurrency(); ++i) {
                CPU_SET_S(i, size, cpuset);
            }
            pthread_setaffinity_np(pthread_self(), size, cpuset);
            CPU_FREE(cpuset);
        }
#endif

        for (;;) {
            if (queue.queueJobs.size_approx() == 0) {
                if (queue.fExit.load(std::memory_order_relaxed))
                    break;
            }

            if (queue.fExit.load(std::memory_order_seq_cst) && queue.queueJobs.size_approx() == 0)
                break;

            vars.gcEpoch = g_pserver->garbageCollector.startEpoch();
            JobBase *rgjob[64];
            int cjobs = 0;
            while ((cjobs = pqueue->queueJobs.wait_dequeue_bulk_timed(rgjob, 64, std::chrono::milliseconds(5))) > 0) {
                std::unique_lock<fastlock> ulPause(pqueue->m_lockPause);

                for (int ijob = 0; ijob < cjobs; ++ijob) {
                    JobBase *pjob = rgjob[ijob];
                    switch (pjob->type)
                    {
                    case JobBase::JobType::Insert:
                        pqueue->processJob(*static_cast<rdbInsertJob*>(pjob));
                        break;

                    case JobBase::JobType::Function:
                        static_cast<rdbFunctionJob*>(pjob)->m_fn();
                        break;
                    }
                    delete pjob;
                }
            }
            g_pserver->garbageCollector.endEpoch(vars.gcEpoch);
        }

        if (g_pserver->m_pstorageFactory) {
            for (int idb = 0; idb < cserver.dbnum; ++idb)
                g_pserver->db[idb]->processChangesAsync(queue.cstorageWritesInFlight);
        }