| 116 | } |
| 117 | |
| 118 | void ProcessEspResponse(InputMessageBase* msg_base) { |
| 119 | const int64_t start_parse_us = butil::cpuwide_time_us(); |
| 120 | DestroyingPtr<MostCommonMessage> msg(static_cast<MostCommonMessage*>(msg_base)); |
| 121 | |
| 122 | // Fetch correlation id that we saved before in `PackEspRequest' |
| 123 | const CallId cid = { static_cast<uint64_t>(msg->socket()->correlation_id()) }; |
| 124 | Controller* cntl = NULL; |
| 125 | const int rc = bthread_id_lock(cid, (void**)&cntl); |
| 126 | if (rc != 0) { |
| 127 | LOG_IF(ERROR, rc != EINVAL && rc != EPERM) |
| 128 | << "Fail to lock correlation_id=" << cid << ", " << berror(rc); |
| 129 | return; |
| 130 | } |
| 131 | |
| 132 | ControllerPrivateAccessor accessor(cntl); |
| 133 | if (auto span = accessor.span()) { |
| 134 | span->set_base_real_us(msg->base_real_us()); |
| 135 | span->set_received_us(msg->received_us()); |
| 136 | span->set_response_size(msg->payload.length()); |
| 137 | span->set_start_parse_us(start_parse_us); |
| 138 | } |
| 139 | // MUST be EspMessage (checked in SerializeEspRequest) |
| 140 | EspMessage* response = (EspMessage*)cntl->response(); |
| 141 | const int saved_error = cntl->ErrorCode(); |
| 142 | |
| 143 | if (response != NULL) { |
| 144 | msg->meta.copy_to(&response->head, sizeof(EspHead)); |
| 145 | msg->payload.swap(response->body); |
| 146 | if (response->head.msg != 0) { |
| 147 | cntl->SetFailed(ENOENT, "esp response head msg != 0"); |
| 148 | LOG(WARNING) << "Server " << msg->socket()->remote_side() |
| 149 | << " doesn't contain the right data"; |
| 150 | } |
| 151 | } // else just ignore the response. |
| 152 | |
| 153 | // Unlocks correlation_id inside. Revert controller's |
| 154 | // error code if it version check of `cid' fails |
| 155 | msg.reset(); // optional, just release resource ASAP |
| 156 | accessor.OnResponse(cid, saved_error); |
| 157 | } |
| 158 | |
| 159 | } // namespace policy |
| 160 | } // namespace brpc |