| 1334 | } |
| 1335 | |
| 1336 | void ReadTabletTask::DoRead(std::shared_ptr<ShardRequest> shard_req) { |
| 1337 | bool is_timeout{false}; |
| 1338 | |
| 1339 | auto& row_results = *shard_req->row_results; |
| 1340 | int64_t index = shard_req->offset; |
| 1341 | int64_t end_index = index + shard_req->row_num; |
| 1342 | |
| 1343 | while (index < end_index) { |
| 1344 | int64_t time_remain_ms = end_time_ms_ - GetTimeStampInMs(); |
| 1345 | StatusCode row_status = kTabletNodeOk; |
| 1346 | |
| 1347 | io::TabletIO* tablet_io = tablet_manager_->GetTablet( |
| 1348 | request_->tablet_name(), request_->row_info_list(index).key(), &row_status); |
| 1349 | if (tablet_io == NULL) { |
| 1350 | response_->mutable_detail()->mutable_status()->Set(index, kKeyNotInRange); |
| 1351 | read_range_error_counter.Inc(); |
| 1352 | } else { |
| 1353 | row_results.emplace_back(new RowResult{}); |
| 1354 | VLOG(20) << "time_remain_ms: " << time_remain_ms; |
| 1355 | if (tablet_io->ReadCells(request_->row_info_list(index), row_results.back().get(), |
| 1356 | snapshot_id_, &row_status, time_remain_ms)) { |
| 1357 | read_success_num_.Inc(); |
| 1358 | } else { |
| 1359 | if (row_status != kKeyNotExist && row_status != kRPCTimeout) { |
| 1360 | if (row_status == kTabletNodeIsBusy) { |
| 1361 | read_reject_counter.Inc(); |
| 1362 | } else { |
| 1363 | read_error_counter.Inc(); |
| 1364 | } |
| 1365 | } |
| 1366 | row_results.pop_back(); |
| 1367 | } |
| 1368 | tablet_io->DecRef(); |
| 1369 | response_->mutable_detail()->mutable_status()->Set(index, row_status); |
| 1370 | } |
| 1371 | if (row_status == kRPCTimeout || has_timeout_.load()) { |
| 1372 | is_timeout = true; |
| 1373 | LOG(WARNING) << "seq_id: " << request_->sequence_id() << " timeout," |
| 1374 | << " clinet_timeout_ms: " << request_->client_timeout_ms(); |
| 1375 | break; |
| 1376 | } |
| 1377 | ++index; |
| 1378 | } |
| 1379 | |
| 1380 | if (is_timeout) { |
| 1381 | has_timeout_.store(true); |
| 1382 | } |
| 1383 | |
| 1384 | FinishShardRequest(shard_req); |
| 1385 | } |
| 1386 | |
| 1387 | void ReadTabletTask::FinishShardRequest(const std::shared_ptr<ShardRequest>& shard_req) { |
| 1388 | if (finished_.Add(shard_req->row_num) == total_row_num_) { |