Method TaskDispatch

phxqueue/consumer/consumer.cpp:235–284

void Consumer::TaskDispatch() {

    // set nonblock for CoRead/CoWrite
    {
         auto flags = fcntl(impl_->consume_fds[1], F_GETFL, 0);
         fcntl(impl_->consume_fds[1], F_SETFL, flags | O_NONBLOCK);
    }

    comm::RetCode ret;

    char ch;
    while (1) {
        comm::ConsumerConsumeBP::GetThreadInstance()->OnTaskDispatch(impl_->cc);

        impl_->nhandle_task_finished = 0;
        impl_->nbatch_handle_task_finished = 0;
        
        while (!comm::utils::CoRead(impl_->consume_fds[1], &ch, sizeof(char))) {
            QLErr("CoRead fail");
            poll(nullptr, 0, 100);
        }

        if (comm::RetCode::RET_OK != (ret = MakeHandleBuckets())) {
            QLErr("MakeHandleBuckets ret %d", comm::as_integer(ret));
        }

        for (int i{0}; i < impl_->opt.nbatch_handler; ++i) {
            impl_->batch_handle_finish[i] = false;
            co_resume(impl_->batch_handle_ctxs[i].co);
        }

        for (int i{0}; i < impl_->opt.nhandler; ++i) {
            if (!impl_->handle_buckets[i].empty()) {
                co_resume(impl_->handle_ctxs[i].co);
            }
        }

        while (1) {
            if (IsAllTaskFinish()) break;
        }

        comm::ConsumerConsumeBP::GetThreadInstance()->OnTaskDispatchFinish(impl_->cc);


        while (!comm::utils::CoWrite(impl_->consume_fds[1], "a", sizeof(char))) {
            QLErr("CoWrite fail");
            poll(nullptr, 0, 100);
        }
    }
}

void Consumer::HandleTaskFinish() {
    ++impl_->nhandle_task_finished;