| 455 | } |
| 456 | |
| 457 | void RemoteTabletNode::ScanTablet(google::protobuf::RpcController* controller, |
| 458 | const ScanTabletRequest* request, ScanTabletResponse* response, |
| 459 | google::protobuf::Closure* done) { |
| 460 | done = ScanDoneWrapper::NewInstance(get_micros(), request, response, done, quota_entry_); |
| 461 | VLOG(8) << "accept RPC (ScanTablet): [" << request->table_name() << "] " |
| 462 | << tera::utils::GetRemoteAddress(controller); |
| 463 | scan_request_counter.Inc(); |
| 464 | if (scan_pending_counter.Get() > FLAGS_tera_scan_request_pending_limit) { |
| 465 | response->set_sequence_id(request->sequence_id()); |
| 466 | response->set_status(kTabletNodeIsBusy); |
| 467 | scan_reject_counter.Inc(); |
| 468 | done->Run(); |
| 469 | VLOG(8) << "finish RPC (ScanTablet)"; |
| 470 | } else { |
| 471 | // check user identification & access |
| 472 | if (!access_entry_->VerifyAndAuthorize(request, response)) { |
| 473 | response->set_sequence_id(request->sequence_id()); |
| 474 | VLOG(20) << "Access VerifyAndAuthorize failed for ScanTablet"; |
| 475 | done->Run(); |
| 476 | return; |
| 477 | } |
| 478 | ScanRpc* rpc = new ScanRpc(controller, request, response, done); |
| 479 | if (scan_pending_counter.Get() >= |
| 480 | FLAGS_tera_request_pending_limit * FLAGS_tera_quota_unlimited_pending_ratio) { |
| 481 | if (!DoQuotaScanRpcRetry(rpc)) { |
| 482 | VLOG(8) << "ScanTablet Rpc push to QuotaRetry queue"; |
| 483 | return; |
| 484 | } |
| 485 | } |
| 486 | scan_pending_counter.Inc(); |
| 487 | scan_rpc_schedule_->EnqueueRpc(request->table_name(), rpc); |
| 488 | scan_thread_pool_->AddTask( |
| 489 | std::bind(&RemoteTabletNode::DoScheduleRpc, this, scan_rpc_schedule_.get())); |
| 490 | } |
| 491 | } |
| 492 | |
| 493 | void RemoteTabletNode::Query(google::protobuf::RpcController* controller, |
| 494 | const QueryRequest* request, QueryResponse* response, |
no test coverage detected