Method ConsumeThreadRun

phxqueue/consumer/consumer.cpp:416–477

}

void Consumer::ConsumeThreadRun(const int vpid) {
    OnConsumeThreadRun(vpid);
    comm::ConsumerConsumeBP::GetThreadInstance()->OnConsumeThreadRun(impl_->cc);

    //stShareStack_t *share_stack= co_alloc_sharestack(impl_->opt.nshare_stack, 1024 * impl_->opt.share_stack_size_kb);
    stCoRoutineAttr_t attr;
    //attr.stack_size = 0;
    attr.stack_size = 1024 * impl_->opt.share_stack_size_kb;
    //attr.share_stack = share_stack;

	impl_->cond = co_cond_alloc();

    if (0 != socketpair(PF_LOCAL, SOCK_STREAM, 0, impl_->consume_fds)) {
        QLErr("socketpair fail");
        return;
    }

    impl_->handle_buckets = unique_ptr<queue<int>[]>(new queue<int>[impl_->opt.nhandler]);

    impl_->handle_ctxs = unique_ptr<struct ConsumeCtx_t[]>(new ConsumeCtx_t[impl_->opt.nhandler]);
    for (int i{0}; i < impl_->opt.nhandler; ++i) {
        auto &&ctx = impl_->handle_ctxs[i];
        ctx.co = nullptr;
        ctx.cid = i;
        ctx.consumer = this;

        co_create(&(ctx.co), &attr, HandleRoutineRun, &ctx);
        co_resume(ctx.co);
    }

    impl_->batch_handle_finish = unique_ptr<bool[]>(new bool[impl_->opt.nbatch_handler]);

    impl_->batch_handle_ctxs = unique_ptr<struct ConsumeCtx_t[]>(new ConsumeCtx_t[impl_->opt.nbatch_handler]);
    for (int i{0}; i < impl_->opt.nbatch_handler; ++i) {
        impl_->batch_handle_finish[i] = true;

        auto &&ctx = impl_->batch_handle_ctxs[i];
        ctx.co = NULL;
        ctx.cid = i;
        ctx.consumer = this;

        co_create(&(ctx.co), &attr, BatchHandleRoutineRun, &ctx);
        co_resume(ctx.co);
    }

    // task dispatch
    {
        auto &&ctx = impl_->dispatch_ctx;
        ctx.co = nullptr;
        ctx.consumer = this;
        co_create(&(ctx.co), &attr, DispatchRoutineRun, &ctx);
        co_resume(ctx.co);
    }


    co_eventloop(co_get_epoll_ct(), nullptr, this);

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