Method call_task

src/client/partition_resolver.cpp:57–130

}

void partition_resolver::call_task(const rpc_response_task_ptr &t)
{
    auto &hdr = *(t->get_request()->header);
    uint64_t deadline_ms = dsn_now_ms() + hdr.client.timeout_ms;

    rpc_response_handler old_callback;
    t->fetch_current_handler(old_callback);
    auto new_callback = [ this, deadline_ms, oc = std::move(old_callback) ](
        dsn::error_code err, dsn::message_ex * req, dsn::message_ex * resp)
    {
        if (req->header->gpid.value() != 0 && err != ERR_OK && error_retry(err)) {
            on_access_failure(req->header->gpid.get_partition_index(), err);
            // still got time, retry
            uint64_t nms = dsn_now_ms();
            uint64_t gap = 8 << req->send_retry_count;
            if (gap > 1000)
                gap = 1000;
            if (nms + gap < deadline_ms) {
                req->send_retry_count++;
                req->header->client.timeout_ms = static_cast<int>(deadline_ms - nms - gap);

                rpc_response_task_ptr ctask =
                    dynamic_cast<rpc_response_task *>(task::get_current_task());
                partition_resolver_ptr r(this);

                CHECK_NOTNULL(ctask, "current task must be rpc_response_task");
                ctask->replace_callback(std::move(oc));
                CHECK(ctask->set_retry(false),
                      "rpc_response_task set retry failed, state = {}",
                      enum_to_string(ctask->state()));

                // sleep gap milliseconds before retry
                tasking::enqueue(LPC_RPC_DELAY_CALL,
                                 nullptr,
                                 [r, ctask]() { r->call_task(ctask); },
                                 0,
                                 std::chrono::milliseconds(gap));
                return;
            } else {
                LOG_ERROR("service access failed ({}), no more time for further tries, set error "
                          "= ERR_TIMEOUT, trace_id = {:#018x}",
                          err,
                          req->header->trace_id);
                err = ERR_TIMEOUT;
            }
        }

        if (oc)
            oc(err, req, resp);
    };
    t->replace_callback(std::move(new_callback));

    resolve(hdr.client.partition_hash,
            [t](resolve_result &&result) mutable {
                if (result.err != ERR_OK) {
                    t->enqueue(result.err, nullptr);
                    return;
                }