Function DrainOnNetThread

src/platform/linux/schema_fetch.cpp:422–472

Drain on net thread.

// Drain on net thread.
void DrainOnNetThread() {
    if (!MetadataSync::SchemaFetchEnabled()) return;
    if (t_draining) return;
    if (g_shuttingDown.load(std::memory_order_acquire)) return;

    // Only log/work when the queue actually has something, to avoid spamming
    // every BYieldingSend tick.
    {
        std::lock_guard<std::mutex> lock(g_sendMutex);
        if (g_sendQueue.empty()) return;
    }

    if (!g_sessionCaptured.load(std::memory_order_relaxed)) {
        LOG("[SchemaFetch] Drain: queue non-empty but session not captured, deferring");
        return;
    }

    uint32_t conn = g_statsConnHandle.load(std::memory_order_relaxed);
    if (conn == 0) conn = g_connHandle.load(std::memory_order_relaxed);
    if (conn == 0) {
        LOG("[SchemaFetch] Drain: queue non-empty but conn==0, deferring");
        return;
    }

    void* cmInterface = g_cmInterface.load(std::memory_order_relaxed);
    if (!cmInterface) {
        LOG("[SchemaFetch] Drain: queue non-empty but cmInterface==null, deferring");
        return;
    }

    size_t qsize;
    {
        std::lock_guard<std::mutex> lock(g_sendMutex);
        qsize = g_sendQueue.size();
    }
    LOG("[SchemaFetch] Drain: on net thread, conn=%u, %zu queued", conn, qsize);

    t_draining = true;
    constexpr int kMaxPerTick = 2;
    for (int i = 0; i < kMaxPerTick; ++i) {
        SchemaSendItem item;
        {
            std::lock_guard<std::mutex> lock(g_sendMutex);
            if (g_sendQueue.empty()) break;
            item = g_sendQueue.front();
            g_sendQueue.pop();
        }
        SendSchemaRequest(item.appId, item.owner, cmInterface, conn);
    }
    t_draining = false;
}

// Inbound 819 capture (mirror of Windows TryHandleSchemaResponse).