Function WorkerLoop

src/common/cloud_work_queue.cpp:259–473

}

static void WorkerLoop(int threadId) {
    LOG("[CloudStorage] Background worker %d started", threadId);
    int consecutiveFailures = 0;
    while (g_workerRunning) {
        WorkItem item;
        {
            std::unique_lock<std::mutex> lock(g_queueMutex);
            auto eligibleNow = [](const WorkItem& q) {
                return !g_activePaths.count(q.cloudPath)
                    && std::chrono::steady_clock::now() >= q.notBefore;
            };
            auto havePending = [&]() {
                if (!g_workerRunning) return true;
                for (const auto& queued : g_workQueue) {
                    if (eligibleNow(queued)) return true;
                }
                return false;
            };
            while (!havePending()) {
                const auto kNoDeferred =
                    (std::chrono::steady_clock::time_point::max)();
                auto soonest = kNoDeferred;
                for (const auto& queued : g_workQueue) {
                    if (g_activePaths.count(queued.cloudPath)) continue;
                    if (queued.notBefore < soonest) soonest = queued.notBefore;
                }
                if (soonest == kNoDeferred) {
                    g_queueCV.wait(lock);
                } else {
                    g_queueCV.wait_until(lock, soonest);
                }
                if (!g_workerRunning && g_workQueue.empty()) break;
            }
            if (!g_workerRunning && g_workQueue.empty()) break;

            auto workIt = std::find_if(g_workQueue.begin(), g_workQueue.end(),
                [&](const WorkItem& queued) { return eligibleNow(queued); });
            if (workIt == g_workQueue.end()) continue;

            item = std::move(*workIt);
            if (item.type == WorkItem::Upload) {
                g_uploadIndex.erase(item.cloudPath);
            }
            g_workQueue.erase(workIt);
            ++g_activeWorkers;
            ++g_activePaths[item.cloudPath];
            if (item.bestEffort) ++g_activeBestEffortPaths[item.cloudPath];
            if (item.type == WorkItem::Delete) ++g_activeDeletes[item.cloudPath];
        }

        if (!g_provider) {
            std::lock_guard<std::mutex> lk(g_queueMutex);
            auto it = g_activePaths.find(item.cloudPath);
            if (it != g_activePaths.end() && --it->second <= 0) g_activePaths.erase(it);
            if (item.bestEffort) ClearActiveBestEffortLocked(item.cloudPath);
            if (item.type == WorkItem::Delete) ClearActiveDeleteLocked(item.cloudPath);
            --g_activeWorkers;
            g_drainCV.notify_all();