| 294 | } |
| 295 | |
| 296 | void RemoteTabletNode::ReadTablet(google::protobuf::RpcController* controller, |
| 297 | const ReadTabletRequest* request, ReadTabletResponse* response, |
| 298 | google::protobuf::Closure* done) { |
| 299 | int64_t start_micros = get_micros(); |
| 300 | done = ReadDoneWrapper::NewInstance(start_micros, request, response, done, quota_entry_); |
| 301 | VLOG(8) << "accept RPC (ReadTablet): [" << request->tablet_name() << "] " |
| 302 | << tera::utils::GetRemoteAddress(controller); |
| 303 | static uint32_t last_print = time(NULL); |
| 304 | int32_t row_num = request->row_info_list_size(); |
| 305 | read_request_counter.Add(row_num); |
| 306 | if (read_pending_counter.Get() > FLAGS_tera_request_pending_limit) { |
| 307 | response->set_sequence_id(request->sequence_id()); |
| 308 | response->set_status(kTabletNodeIsBusy); |
| 309 | read_reject_counter.Add(row_num); |
| 310 | done->Run(); |
| 311 | uint32_t now_time = time(NULL); |
| 312 | if (now_time > last_print) { |
| 313 | LOG(WARNING) << "Too many pending read requests, return TabletNode Is Busy!"; |
| 314 | last_print = now_time; |
| 315 | } |
| 316 | VLOG(8) << "finish RPC (ReadTablet)"; |
| 317 | } else { |
| 318 | // check user identification & access |
| 319 | if (!access_entry_->VerifyAndAuthorize(request, response)) { |
| 320 | response->set_sequence_id(request->sequence_id()); |
| 321 | VLOG(20) << "Access VerifyAndAuthorize failed for ReadTablet"; |
| 322 | done->Run(); |
| 323 | return; |
| 324 | } |
| 325 | if (read_pending_counter.Get() >= |
| 326 | FLAGS_tera_request_pending_limit * FLAGS_tera_quota_unlimited_pending_ratio) { |
| 327 | if (!quota_entry_->CheckAndConsume( |
| 328 | request->tablet_name(), |
| 329 | quota::OpTypeAmountList{std::make_pair(kQuotaReadReqs, row_num)})) { |
| 330 | response->set_sequence_id(request->sequence_id()); |
| 331 | response->set_status(kQuotaLimited); |
| 332 | read_quota_rejest_counter.Add(row_num); |
| 333 | VLOG(20) << "quota_entry check failed for ReadTablet"; |
| 334 | done->Run(); |
| 335 | return; |
| 336 | } |
| 337 | } |
| 338 | read_pending_counter.Add(row_num); |
| 339 | ReadRpcTimer* timer = new ReadRpcTimer(request, response, done, start_micros); |
| 340 | RpcTimerList::Instance()->Push(timer); |
| 341 | |
| 342 | ReadRpc* rpc = new ReadRpc(controller, request, response, done, timer, start_micros); |
| 343 | read_rpc_schedule_->EnqueueRpc(request->tablet_name(), rpc); |
| 344 | read_thread_pool_->AddTask( |
| 345 | std::bind(&RemoteTabletNode::DoScheduleRpc, this, read_rpc_schedule_.get())); |
| 346 | } |
| 347 | } |
| 348 | |
| 349 | void RemoteTabletNode::WriteTablet(google::protobuf::RpcController* controller, |
| 350 | const WriteTabletRequest* request, WriteTabletResponse* response, |
no test coverage detected