| 945 | } |
| 946 | |
| 947 | void TableImpl::MutateCallBack(std::vector<int64_t>* mu_id_list, WriteTabletRequest* request, |
| 948 | WriteTabletResponse* response, bool failed, int error_code) { |
| 949 | perf_counter_.rpc_w.Add(get_micros() - request->timestamp()); |
| 950 | perf_counter_.rpc_w_cnt.Inc(); |
| 951 | if (failed) { |
| 952 | if (error_code == sofa::pbrpc::RPC_ERROR_SERVER_SHUTDOWN || |
| 953 | error_code == sofa::pbrpc::RPC_ERROR_SERVER_UNREACHABLE || |
| 954 | error_code == sofa::pbrpc::RPC_ERROR_SERVER_UNAVAILABLE) { |
| 955 | response->set_status(kServerError); |
| 956 | } else if (error_code == sofa::pbrpc::RPC_ERROR_REQUEST_CANCELED || |
| 957 | error_code == sofa::pbrpc::RPC_ERROR_SEND_BUFFER_FULL) { |
| 958 | response->set_status(kClientError); |
| 959 | } else if (error_code == sofa::pbrpc::RPC_ERROR_CONNECTION_CLOSED || |
| 960 | error_code == sofa::pbrpc::RPC_ERROR_RESOLVE_ADDRESS) { |
| 961 | response->set_status(kConnectError); |
| 962 | } else if (error_code == sofa::pbrpc::RPC_ERROR_REQUEST_TIMEOUT) { |
| 963 | response->set_status(kRPCTimeout); |
| 964 | } else { |
| 965 | response->set_status(kRPCError); |
| 966 | } |
| 967 | } |
| 968 | |
| 969 | bool rpc_timeout_timer_reset = (kRPCTimeout != response->status()); |
| 970 | std::map<uint32_t, std::vector<int64_t>*> retry_times_list; |
| 971 | std::vector<SdkTask*> not_in_range_list; |
| 972 | for (uint32_t i = 0; i < mu_id_list->size(); ++i) { |
| 973 | const std::string& row = request->row_list(i).row_key(); |
| 974 | int64_t mu_id = (*mu_id_list)[i]; |
| 975 | if (rpc_timeout_timer_reset) { |
| 976 | SdkTask* task = task_pool_.GetTask(mu_id); |
| 977 | if (task == NULL) { |
| 978 | VLOG(10) << "mutation " << mu_id << " finish but timeout"; |
| 979 | } else if (!task->GetServerAddr().empty()) { |
| 980 | MutexLock lock(&rpc_timeout_duration_mutex_); |
| 981 | rpc_timeout_duration_[task->GetServerAddr()] = get_millis(); |
| 982 | rpc_timeout_timer_reset = false; |
| 983 | } else { |
| 984 | VLOG(20) << "task ServerAddr is not assigned"; |
| 985 | } |
| 986 | if (task != NULL) { |
| 987 | task->DecRef(); |
| 988 | } |
| 989 | } |
| 990 | StatusCode err = response->status(); |
| 991 | if (err == kTabletNodeOk) { |
| 992 | err = response->row_status_list(i); |
| 993 | } |
| 994 | if (err == kTabletNodeOk || err == kTxnFail || err == kTableInvalidArg || |
| 995 | err == kNotPermission) { |
| 996 | perf_counter_.mutate_ok_cnt.Inc(); |
| 997 | SdkTask* task = task_pool_.PopTask(mu_id); |
| 998 | if (task == NULL) { |
| 999 | VLOG(10) << "mutation " << mu_id << " finish but timeout: " << DebugString(row); |
| 1000 | continue; |
| 1001 | } |
| 1002 | CHECK_EQ(task->Type(), SdkTask::MUTATION); |
| 1003 | CHECK_EQ(task->GetRef(), 1); |
| 1004 | RowMutationImpl* row_mutation = (RowMutationImpl*)task; |
no test coverage detected