| 1058 | } |
| 1059 | |
| 1060 | void Controller::IssueRPC(int64_t start_realtime_us) { |
| 1061 | _current_call.begin_time_us = start_realtime_us; |
| 1062 | |
| 1063 | // If has retry/backup request，we will recalculate the timeout, |
| 1064 | if (_real_timeout_ms > 0) { |
| 1065 | _real_timeout_ms -= (start_realtime_us - _begin_time_us) / 1000; |
| 1066 | } |
| 1067 | |
| 1068 | // Clear last error, Don't clear _error_text because we append to it. |
| 1069 | _error_code = 0; |
| 1070 | |
| 1071 | // Make versioned correlation_id. |
| 1072 | // call_id : unversioned, mainly for ECANCELED and ERPCTIMEDOUT |
| 1073 | // call_id + 1 : first try. |
| 1074 | // call_id + 2 : retry 1 |
| 1075 | // ... |
| 1076 | // call_id + N + 1 : retry N |
| 1077 | // All ids except call_id are versioned. Say if we've sent retry 1 and |
| 1078 | // a failed response of first try comes back, it will be ignored. |
| 1079 | const CallId cid = current_id(); |
| 1080 | |
| 1081 | // Intercept IssueRPC when _sender is set. Currently _sender is only set |
| 1082 | // by SelectiveChannel. |
| 1083 | if (_sender) { |
| 1084 | if (_sender->IssueRPC(start_realtime_us) != 0) { |
| 1085 | return HandleSendFailed(); |
| 1086 | } |
| 1087 | CHECK_EQ(0, bthread_id_unlock(cid)); |
| 1088 | return; |
| 1089 | } |
| 1090 | |
| 1091 | // Pick a target server for sending RPC |
| 1092 | _current_call.need_feedback = false; |
| 1093 | _current_call.enable_circuit_breaker = has_enabled_circuit_breaker(); |
| 1094 | SocketUniquePtr tmp_sock; |
| 1095 | if (SingleServer()) { |
| 1096 | // Don't use _current_call.peer_id which is set to -1 after construction |
| 1097 | // of the backup call. |
| 1098 | const int rc = Socket::Address(_single_server_id, &tmp_sock); |
| 1099 | if (rc != 0 || (!is_health_check_call() && !tmp_sock->IsAvailable())) { |
| 1100 | SetFailed(EHOSTDOWN, "Not connected to %s yet, server_id=%" PRIu64, |
| 1101 | endpoint2str(_remote_side).c_str(), _single_server_id); |
| 1102 | tmp_sock.reset(); // Release ref ASAP |
| 1103 | return HandleSendFailed(); |
| 1104 | } |
| 1105 | _current_call.peer_id = _single_server_id; |
| 1106 | } else { |
| 1107 | LoadBalancer::SelectIn sel_in = |
| 1108 | { start_realtime_us, true, |
| 1109 | has_request_code(), _request_code, _accessed }; |
| 1110 | LoadBalancer::SelectOut sel_out(&tmp_sock); |
| 1111 | const int rc = _lb->SelectServer(sel_in, &sel_out); |
| 1112 | if (rc != 0) { |
| 1113 | std::ostringstream os; |
| 1114 | DescribeOptions opt; |
| 1115 | opt.verbose = false; |
| 1116 | _lb->Describe(os, opt); |
| 1117 | SetFailed(rc, "Fail to select server from %s", os.str().c_str()); |
no test coverage detected