Method DispatchBatchTask

phxqueue/producer/batchhelper.cpp:260–294

}

void BatchHelper::DispatchBatchTask(int vtid) {
    auto &batch_infos = impl_->vtid2batch_infos[vtid];
    auto rwlock = &impl_->vtid2rwlock[vtid];
    auto &idle_process_ctxs = impl_->vtid2idle_process_ctxs[vtid];

    static __thread uint64_t next_dispatch_timestampe_ms = 0;
    uint64_t ready_timestamp_ms;

    if (next_dispatch_timestampe_ms && next_dispatch_timestampe_ms > comm::utils::Time::GetSteadyClockMS()) return;
    next_dispatch_timestampe_ms = 0;

    comm::utils::RWLock rwlock_read(rwlock, comm::utils::RWLock::LockMode::READ);

    for (auto &&kv : batch_infos) {
        auto batch_info = &kv.second;

        lock_guard<mutex> lock_guard(batch_info->first);
        auto &batch_task = batch_info->second;

        if (nullptr != batch_task) {
            if (batch_task->Ready(&ready_timestamp_ms)) {
                if (!idle_process_ctxs.empty()) {
                    auto ctx = idle_process_ctxs.top();
                    idle_process_ctxs.pop();
                    ctx->ready_batch_task = move(batch_task);
                    co_resume(ctx->co);
                }
            } else {
                if (ready_timestamp_ms > 0 && (0 == next_dispatch_timestampe_ms || ready_timestamp_ms < next_dispatch_timestampe_ms)) {
                    next_dispatch_timestampe_ms = ready_timestamp_ms;
                }
            }
        }
    }
}

static void *ProcessRoutineRun(void *arg) {
    co_enable_hook_sys();