Method TaskTimeout

src/sdk/table_impl.cc:1051–1112

}

void TableImpl::TaskTimeout(SdkTask* task) {
  perf_counter_.GetTimeoutCnt(task).Inc();
  CHECK_NOTNULL(task);

  task->ExcludeOtherRef();

  StatusCode err = task->GetInternalError();
  if (err == kKeyNotInRange || err == kConnectError) {
    ScheduleUpdateMeta(task->InternalRowKey(), task->GetMetaTimeStamp());
  } else if (err == kRPCTimeout && !task->GetServerAddr().empty()) {
    MutexLock lock(&rpc_timeout_duration_mutex_);
    if (rpc_timeout_duration_.find(task->GetServerAddr()) == rpc_timeout_duration_.end()) {
      rpc_timeout_duration_[task->GetServerAddr()] = get_millis();
    } else if (get_millis() - rpc_timeout_duration_[task->GetServerAddr()] >=
               FLAGS_tera_sdk_update_meta_rpc_timeout_max_ms) {
      LOG(WARNING) << "requests on server<" << task->GetServerAddr() << "> continuous "
                   << "over " << FLAGS_tera_sdk_update_meta_rpc_timeout_max_ms / 1000
                   << "s with the response of kRpcTimeout, last succ request time("
                   << rpc_timeout_duration_[task->GetServerAddr()] / 1000
                   << "), may be the tabletserver is zombie, try to update meta.";
      rpc_timeout_duration_[task->GetServerAddr()] = get_millis();
      rpc_timeout_duration_mutex_.Unlock();
      ScheduleUpdateMeta(task->InternalRowKey(), task->GetMetaTimeStamp());
      rpc_timeout_duration_mutex_.Lock();
    }
  }

  std::string err_reason;
  if (task->RetryTimes() == 0) {
    perf_counter_.GetQueueTimeoutCnt(task).Inc();
    err_reason = StringFormat("commit lld times, retry 0 times, in %u ms.", task->GetCommitTimes(),
                              task->Type() == SdkTask::READ ? read_timeout_ : write_timeout_);
  } else {
    err_reason = StringFormat("commit %lld times, retry %u times, in %u ms. last error: %s",
                              task->GetCommitTimes(), task->RetryTimes(),
                              task->Type() == SdkTask::READ ? read_timeout_ : write_timeout_,
                              StatusCodeToString(err).c_str());
  }
  switch (task->Type()) {
    case SdkTask::READ: {
      RowReaderImpl* row_reader = (RowReaderImpl*)task;
      row_reader->SetError(ErrorCode::kTimeout, err_reason);
      cur_reader_pending_counter_.Dec();
    } break;
    case SdkTask::BATCH_MUTATION: {
      BatchMutationImpl* batch_mutation = (BatchMutationImpl*)task;
      batch_mutation->SetError(ErrorCode::kTimeout, err_reason);
      cur_commit_pending_counter_.Dec();
    } break;
    case SdkTask::MUTATION: {
      RowMutationImpl* row_mutation = (RowMutationImpl*)task;
      row_mutation->SetError(ErrorCode::kTimeout, err_reason);
      cur_commit_pending_counter_.Dec();
    } break;
    default:
      abort();
  }
  int64_t perf_time = get_micros();