| 487 | } |
| 488 | |
| 489 | void ImpalaServer::ShutdownWorkloadManagement() { |
| 490 | unique_lock<mutex> l(workload_mgmt_state_mu_); |
| 491 | |
| 492 | // Handle the situation where this function runs before the workload management process |
| 493 | // has been started and thus workload_management_thread_ holds a nullptr. |
| 494 | if (workload_mgmt_state_ == WorkloadManagementState::NOT_STARTED) { |
| 495 | workload_mgmt_state_ = WorkloadManagementState::SHUTDOWN; |
| 496 | return; |
| 497 | } |
| 498 | |
| 499 | DCHECK_NE(nullptr, workload_management_thread_.get()); |
| 500 | |
| 501 | // If the completed queries thread is not yet running, then we don't need to give it a |
| 502 | // chance to flush the in-memory queue to the completed queries table. |
| 503 | if (workload_mgmt_state_ == WorkloadManagementState::RUNNING) { |
| 504 | workload_mgmt_state_ = WorkloadManagementState::SHUTTING_DOWN; |
| 505 | LOG(INFO) << "Workload management is shutting down"; |
| 506 | |
| 507 | // Wake up the completed queries processing thread so it can drain the in-memory |
| 508 | // completed queries queue. |
| 509 | _completed_queries_cv.notify_all(); |
| 510 | |
| 511 | // Wait for the completed queries processing thread to drain the in-memory queue. If |
| 512 | // the timeout expires before the queue is drained, the thread will be detached |
| 513 | // and shutdown will continue. |
| 514 | _completed_queries_shutdown_cv.wait_for(l, |
| 515 | chrono::seconds(FLAGS_query_log_shutdown_timeout_s), |
| 516 | [this] { return workload_mgmt_state_ == WorkloadManagementState::SHUTDOWN; }); |
| 517 | } |
| 518 | |
| 519 | switch (workload_mgmt_state_) { |
| 520 | case WorkloadManagementState::SHUTDOWN: |
| 521 | // Safe to join the thread here because the workload managmenent processing loop |
| 522 | // sets the thread state to ThreadState::SHUTDOWN immediately before it returns. |
| 523 | LOG(INFO) << "Workload management shutdown successful"; |
| 524 | workload_management_thread_->Join(); |
| 525 | break; |
| 526 | default: |
| 527 | // The shutdown timeout expired without the completed queries queue draining. |
| 528 | LOG(INFO) << "Workload management shutdown timed out. Up to '" |
| 529 | << ImpaladMetrics::COMPLETED_QUERIES_QUEUED->GetValue() |
| 530 | << "' queries may have " |
| 531 | << "been lost"; |
| 532 | workload_management_thread_->Detach(); |
| 533 | break; |
| 534 | } |
| 535 | } // function ImpalaServer::ShutdownWorkloadManagement |
| 536 | |
| 537 | void ImpalaServer::EnqueueCompletedQuery( |
| 538 | const QueryHandle& query_handle, shared_ptr<QueryStateRecord> qs_rec) { |