Function BatchHandleRoutineRun

phxqueue/consumer/consumer.cpp:375–388

}

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

    ConsumeCtx_t *ctx = static_cast<ConsumeCtx_t*>(arg);

    while (1) {
        if (ctx->consumer->HasBatchHandleTasks(ctx->cid)) {
            co_yield_ct();
            continue;
        }
        ctx->consumer->ProcessAllBatchHandleTasks(ctx->cid);
    }
    return nullptr;
}

bool Consumer::HasHandleTasks(const int cid) {
    return impl_->handle_buckets[cid].empty();