| 801 | } |
| 802 | |
| 803 | void TableImpl::BatchMutateCallBack(std::vector<int64_t>* mu_id_list, WriteTabletRequest* request, |
| 804 | WriteTabletResponse* response, bool failed, int error_code) { |
| 805 | perf_counter_.rpc_w.Add(get_micros() - request->timestamp()); |
| 806 | perf_counter_.rpc_w_cnt.Inc(); |
| 807 | if (failed) { |
| 808 | if (error_code == sofa::pbrpc::RPC_ERROR_SERVER_SHUTDOWN || |
| 809 | error_code == sofa::pbrpc::RPC_ERROR_SERVER_UNREACHABLE || |
| 810 | error_code == sofa::pbrpc::RPC_ERROR_SERVER_UNAVAILABLE) { |
| 811 | response->set_status(kServerError); |
| 812 | } else if (error_code == sofa::pbrpc::RPC_ERROR_REQUEST_CANCELED || |
| 813 | error_code == sofa::pbrpc::RPC_ERROR_SEND_BUFFER_FULL) { |
| 814 | response->set_status(kClientError); |
| 815 | } else if (error_code == sofa::pbrpc::RPC_ERROR_CONNECTION_CLOSED || |
| 816 | error_code == sofa::pbrpc::RPC_ERROR_RESOLVE_ADDRESS) { |
| 817 | response->set_status(kConnectError); |
| 818 | } else if (error_code == sofa::pbrpc::RPC_ERROR_REQUEST_TIMEOUT) { |
| 819 | response->set_status(kRPCTimeout); |
| 820 | } else { |
| 821 | response->set_status(kRPCError); |
| 822 | } |
| 823 | } |
| 824 | |
| 825 | bool rpc_timeout_timer_reset = (kRPCTimeout != response->status()); |
| 826 | std::map<uint32_t, std::vector<int64_t>*> retry_times_list; |
| 827 | std::vector<SdkTask*> not_in_range_list; |
| 828 | for (uint32_t i = 0; i < mu_id_list->size(); ++i) { |
| 829 | const std::string& row = request->row_list(i).row_key(); |
| 830 | int64_t mu_id = (*mu_id_list)[i]; |
| 831 | if (rpc_timeout_timer_reset) { |
| 832 | SdkTask* task = task_pool_.GetTask(mu_id); |
| 833 | if (task == NULL) { |
| 834 | VLOG(10) << "mutation " << mu_id << " finish but timeout"; |
| 835 | } else if (!task->GetServerAddr().empty()) { |
| 836 | MutexLock lock(&rpc_timeout_duration_mutex_); |
| 837 | rpc_timeout_duration_[task->GetServerAddr()] = get_millis(); |
| 838 | rpc_timeout_timer_reset = false; |
| 839 | } else { |
| 840 | VLOG(20) << "task ServerAddr is not assigned"; |
| 841 | } |
| 842 | if (task != NULL) { |
| 843 | task->DecRef(); |
| 844 | } |
| 845 | } |
| 846 | StatusCode err = response->status(); |
| 847 | if (err == kTabletNodeOk) { |
| 848 | err = response->row_status_list(i); |
| 849 | } |
| 850 | |
| 851 | if (err == kTabletNodeOk || err == kTxnFail || err == kTableInvalidArg) { |
| 852 | perf_counter_.mutate_ok_cnt.Inc(); |
| 853 | SdkTask* task = task_pool_.PopTask(mu_id); |
| 854 | if (task == NULL) { |
| 855 | VLOG(10) << "mutation " << mu_id << " finish but timeout: " << DebugString(row); |
| 856 | continue; |
| 857 | } |
| 858 | CHECK_EQ(task->Type(), SdkTask::BATCH_MUTATION); |
| 859 | CHECK_EQ(task->GetRef(), 1); |
| 860 | BatchMutationImpl* batch_mutation = (BatchMutationImpl*)task; |
no test coverage detected