IMPALA-6171: Revert "IMPALA-1575: part 2: yield admission control resources"
This reverts commit fe90867d890c71bfdcf8ff941f8ec51e36083f25. Change-Id: I3eec4b5a6ff350933ffda0bb80949c5960ecdf25 Reviewed-on: http://gerrit.cloudera.org:8080/8499 Reviewed-by: Thomas Tauber-Marshall <[email protected]> Tested-by: Impala Public Jenkins Project: http://git-wip-us.apache.org/repos/asf/incubator-impala/repo Commit: http://git-wip-us.apache.org/repos/asf/incubator-impala/commit/a772f845 Tree: http://git-wip-us.apache.org/repos/asf/incubator-impala/tree/a772f845 Diff: http://git-wip-us.apache.org/repos/asf/incubator-impala/diff/a772f845 Branch: refs/heads/master Commit: a772f845628f19b816935cd4770714c980c553a6 Parents: 19c17e6 Author: Tim Armstrong <[email protected]> Authored: Wed Nov 8 10:33:52 2017 -0800 Committer: Impala Public Jenkins <[email protected]> Committed: Wed Nov 8 22:03:59 2017 +0000 ---------------------------------------------------------------------- be/src/runtime/coordinator.cc | 29 +--- be/src/runtime/coordinator.h | 38 +---- be/src/runtime/mem-tracker.cc | 6 +- be/src/runtime/mem-tracker.h | 16 +- be/src/runtime/query-state.cc | 3 - be/src/scheduling/admission-controller.cc | 13 +- be/src/scheduling/admission-controller.h | 15 +- be/src/service/client-request-state.cc | 11 ++ .../custom_cluster/test_admission_controller.py | 155 +++++++------------ 9 files changed, 94 insertions(+), 192 deletions(-) ---------------------------------------------------------------------- http://git-wip-us.apache.org/repos/asf/incubator-impala/blob/a772f845/be/src/runtime/coordinator.cc ---------------------------------------------------------------------- diff --git a/be/src/runtime/coordinator.cc b/be/src/runtime/coordinator.cc index ae0c9ad..02c0ca1 100644 --- a/be/src/runtime/coordinator.cc +++ b/be/src/runtime/coordinator.cc @@ -37,7 +37,6 @@ #include "runtime/coordinator-backend-state.h" #include "runtime/debug-options.h" #include "runtime/query-state.h" -#include "scheduling/admission-controller.h" #include "scheduling/scheduler.h" #include "util/bloom-filter.h" #include "util/counting-barrier.h" @@ -80,9 +79,6 @@ Coordinator::Coordinator( Coordinator::~Coordinator() { DCHECK(released_exec_resources_) << "ReleaseExecResources() must be called before Coordinator is destroyed"; - DCHECK(released_admission_control_resources_) - << "ReleaseAdmissionControlResources() must be called before Coordinator is " - << "destroyed"; if (query_state_ != nullptr) { ExecEnv::GetInstance()->query_exec_mgr()->ReleaseQueryState(query_state_); } @@ -833,16 +829,15 @@ Status Coordinator::Wait() { // Execution of query fragments has finished. We don't need to hold onto query execution // resources while we finalize the query. ReleaseExecResources(); + // Query finalization is required only for HDFS table sinks if (needs_finalization_) RETURN_IF_ERROR(FinalizeQuery()); - // Release admission control resources after we'd done the potentially heavyweight - // finalization. - ReleaseAdmissionControlResources(); query_profile_->AddInfoString( "DML Stats", DataSink::OutputDmlStats(per_partition_status_, "\n")); // For DML queries, when Wait is done, the query is complete. ComputeQuerySummary(); + return status; } @@ -873,11 +868,10 @@ Status Coordinator::GetNext(QueryResultSet* results, int max_rows, bool* eos) { returned_all_results_ = true; // release query execution resources here, since we won't be fetching more result rows ReleaseExecResources(); + // wait for all backends to complete before computing the summary // TODO: relocate this so GetNext() won't have to wait for backends to complete? RETURN_IF_ERROR(WaitForBackendCompletion()); - // Release admission control resources after backends are finished. - ReleaseAdmissionControlResources(); // if the query completed successfully, compute the summary if (query_status_.ok()) ComputeQuerySummary(); } @@ -916,7 +910,7 @@ void Coordinator::CancelInternal() { backend_completion_cv_.NotifyAll(); ReleaseExecResourcesLocked(); - ReleaseAdmissionControlResourcesLocked(); + // Report the summary with whatever progress the query made before being cancelled. ComputeQuerySummary(); } @@ -1088,21 +1082,6 @@ void Coordinator::ReleaseExecResourcesLocked() { // caching. The query MemTracker will be cleaned up later. } -void Coordinator::ReleaseAdmissionControlResources() { - lock_guard<mutex> l(lock_); - ReleaseAdmissionControlResourcesLocked(); -} - -void Coordinator::ReleaseAdmissionControlResourcesLocked() { - if (released_admission_control_resources_) return; - LOG(INFO) << "Release admssion control resources for query " - << PrintId(query_ctx_.query_id); - AdmissionController* admission_controller = - ExecEnv::GetInstance()->admission_controller(); - if (admission_controller != nullptr) admission_controller->ReleaseQuery(schedule_); - released_admission_control_resources_ = true; -} - void Coordinator::UpdateFilter(const TUpdateFilterParams& params) { DCHECK_NE(filter_mode_, TRuntimeFilterMode::OFF) << "UpdateFilter() called although runtime filters are disabled"; http://git-wip-us.apache.org/repos/asf/incubator-impala/blob/a772f845/be/src/runtime/coordinator.h ---------------------------------------------------------------------- diff --git a/be/src/runtime/coordinator.h b/be/src/runtime/coordinator.h index e7ddee9..65ef678 100644 --- a/be/src/runtime/coordinator.h +++ b/be/src/runtime/coordinator.h @@ -189,8 +189,7 @@ class Coordinator { // NOLINT: The member variables could be re-ordered to save class FilterState; class FragmentStats; - /// owned by the ClientRequestState that owns this coordinator - const QuerySchedule& schedule_; + const QuerySchedule schedule_; /// copied from TQueryExecRequest; constant across all fragments TQueryCtx query_ctx_; @@ -347,9 +346,6 @@ class Coordinator { // NOLINT: The member variables could be re-ordered to save /// True if and only if ReleaseExecResources() has been called. bool released_exec_resources_ = false; - /// True if and only if ReleaseAdmissionControlResources() has been called. - bool released_admission_control_resources_ = false; - /// Returns a local object pool. ObjectPool* obj_pool() { return obj_pool_.get(); } @@ -441,38 +437,6 @@ class Coordinator { // NOLINT: The member variables could be re-ordered to save /// Same as ReleaseExecResources() except the lock must be held by the caller. void ReleaseExecResourcesLocked(); - - /// Releases admission control resources for use by other queries. - /// This should only be called if one of following preconditions is satisfied for each - /// backend on which the query is executing: - /// * The backend finished execution. - /// Rationale: the backend isn't consuming resources. - // - /// * A cancellation RPC was delivered to the backend. - /// Rationale: the backend will be cancelled and release resources soon. By the - /// time a newly admitted query fragment starts up on the backend and starts consuming - /// resources, the resources from this query will probably have been released. - // - /// * Sending the cancellation RPC to the backend failed - /// Rationale: the backend is either down or will tear itself down when it next tries - /// to send a status RPC to the coordinator. It's possible that the fragment will be - /// slow to tear down and we could overadmit and cause query failures. However, given - /// the communication errors, we need to proceed based on incomplete information about - /// the state of the cluster. We choose to optimistically assume that the backend will - /// tear itself down in a timely manner and admit more queries instead of - /// pessimistically queueing queries while we wait for a response from a backend that - /// may never come. - /// - /// Calling WaitForBackendCompletion() or CancelInternal() before this function is - /// sufficient to satisfy the above preconditions. If the query has an expensive - /// finalization step post query execution (e.g. a DML statement), then this should - /// be called after that completes to avoid over-admitting queries. - /// - /// Acquires lock_. Idempotent. - void ReleaseAdmissionControlResources(); - - /// Same as ReleaseAdmissionControlResources() except lock must be held by caller. - void ReleaseAdmissionControlResourcesLocked(); }; } http://git-wip-us.apache.org/repos/asf/incubator-impala/blob/a772f845/be/src/runtime/mem-tracker.cc ---------------------------------------------------------------------- diff --git a/be/src/runtime/mem-tracker.cc b/be/src/runtime/mem-tracker.cc index 4162e8a..a8cb37e 100644 --- a/be/src/runtime/mem-tracker.cc +++ b/be/src/runtime/mem-tracker.cc @@ -141,13 +141,11 @@ int64_t MemTracker::GetPoolMemReserved() { lock_guard<SpinLock> l(child_trackers_lock_); for (MemTracker* child : child_trackers_) { int64_t child_limit = child->limit(); - bool query_exec_finished = child->query_exec_finished_.Load() != 0; - if (child_limit > 0 && !query_exec_finished) { + if (child_limit > 0) { // Make sure we don't overflow if the query limits are set to ridiculous values. mem_reserved += std::min(child_limit, MemInfo::physical_mem()); } else { - DCHECK(query_exec_finished || child_limit == -1) - << child->LogUsage(UNLIMITED_DEPTH); + DCHECK_EQ(child_limit, -1) << child->LogUsage(UNLIMITED_DEPTH); mem_reserved += child->consumption(); } } http://git-wip-us.apache.org/repos/asf/incubator-impala/blob/a772f845/be/src/runtime/mem-tracker.h ---------------------------------------------------------------------- diff --git a/be/src/runtime/mem-tracker.h b/be/src/runtime/mem-tracker.h index 539f973..1260351 100644 --- a/be/src/runtime/mem-tracker.h +++ b/be/src/runtime/mem-tracker.h @@ -294,9 +294,8 @@ class MemTracker { /// Returns the memory 'reserved' by this resource pool mem tracker, which is the sum /// of the memory reserved by the queries in it (i.e. its child trackers). The mem - /// reserved for a query that is currently executing is its limit_, if set (which - /// should be the common case with admission control). Otherwise, if the query has - /// no limit or the query is finished executing, the current consumption is used. + /// reserved for a query is its limit_, if set (which should be the common case with + /// admission control). Otherwise the current consumption is used. int64_t GetPoolMemReserved(); /// Returns the memory consumed in bytes. @@ -352,11 +351,6 @@ class MemTracker { Status MemLimitExceeded(RuntimeState* state, const std::string& details, int64_t failed_allocation = 0) WARN_UNUSED_RESULT; - void set_query_exec_finished() { - DCHECK(is_query_mem_tracker_); - query_exec_finished_.Store(1); - } - static const std::string COUNTER_NAME; private: @@ -392,12 +386,6 @@ class MemTracker { /// True if this is a Query MemTracker returned from CreateQueryMemTracker(). bool is_query_mem_tracker_ = false; - /// Only used if 'is_query_mem_tracker_' is true. - /// 0 if the query is still executing or 1 if it has finished executing. Before - /// it has finished executing, the tracker limit is treated as "reserved memory" - /// for the purpose of admission control - see GetPoolMemReserved(). - AtomicInt32 query_exec_finished_{0}; - /// Only valid for MemTrackers returned from CreateQueryMemTracker() TUniqueId query_id_; http://git-wip-us.apache.org/repos/asf/incubator-impala/blob/a772f845/be/src/runtime/query-state.cc ---------------------------------------------------------------------- diff --git a/be/src/runtime/query-state.cc b/be/src/runtime/query-state.cc index fd72f6a..6796c82 100644 --- a/be/src/runtime/query-state.cc +++ b/be/src/runtime/query-state.cc @@ -82,9 +82,6 @@ void QueryState::ReleaseExecResources() { if (initial_reservations_ != nullptr) initial_reservations_->ReleaseResources(); if (buffer_reservation_ != nullptr) buffer_reservation_->Close(); if (desc_tbl_ != nullptr) desc_tbl_->ReleaseResources(); - // Mark the query as finished on the query MemTracker so that admission control will - // not consider the whole query memory limit to be "reserved". - query_mem_tracker_->set_query_exec_finished(); // At this point query execution should not be consuming any resources but some tracked // memory may still be used by the ClientRequestState for result caching. The query // MemTracker will be closed later when this QueryState is torn down. http://git-wip-us.apache.org/repos/asf/incubator-impala/blob/a772f845/be/src/scheduling/admission-controller.cc ---------------------------------------------------------------------- diff --git a/be/src/scheduling/admission-controller.cc b/be/src/scheduling/admission-controller.cc index 99f659a..a5d62b4 100644 --- a/be/src/scheduling/admission-controller.cc +++ b/be/src/scheduling/admission-controller.cc @@ -608,19 +608,20 @@ Status AdmissionController::AdmitQuery(QuerySchedule* schedule) { } } -void AdmissionController::ReleaseQuery(const QuerySchedule& schedule) { - if (!schedule.is_admitted()) return; // No-op if query was not admitted - const string& pool_name = schedule.request_pool(); +Status AdmissionController::ReleaseQuery(QuerySchedule* schedule) { + if (!schedule->is_admitted()) return Status::OK(); // No-op if query was not admitted + const string& pool_name = schedule->request_pool(); { lock_guard<mutex> lock(admission_ctrl_lock_); PoolStats* stats = GetPoolStats(pool_name); - stats->Release(schedule); - UpdateHostMemAdmitted(schedule, -schedule.GetPerHostMemoryEstimate()); + stats->Release(*schedule); + UpdateHostMemAdmitted(*schedule, -schedule->GetPerHostMemoryEstimate()); pools_for_updates_.insert(pool_name); - VLOG_RPC << "Released query id=" << schedule.query_id() << " " + VLOG_RPC << "Released query id=" << schedule->query_id() << " " << stats->DebugString(); } dequeue_cv_.NotifyOne(); + return Status::OK(); } // Statestore subscriber callback for IMPALA_REQUEST_QUEUE_TOPIC. http://git-wip-us.apache.org/repos/asf/incubator-impala/blob/a772f845/be/src/scheduling/admission-controller.h ---------------------------------------------------------------------- diff --git a/be/src/scheduling/admission-controller.h b/be/src/scheduling/admission-controller.h index 2830bee..86b8338 100644 --- a/be/src/scheduling/admission-controller.h +++ b/be/src/scheduling/admission-controller.h @@ -96,13 +96,12 @@ class ExecEnv; /// a) Mem Reserved: the amount of memory that has been reported as reserved by all /// backends, which come from the statestore topic updates. The values that are sent /// come from the pool mem trackers in UpdateMemTrackerStats(), which reflects the -/// memory reserved by fragments that have begun execution. For queries that are -/// executing and have mem limits, the limit is considered to be its reserved memory -/// because it may consume up to that limit. Otherwise the query's current consumption -/// is used (see MemTracker::GetPoolMemReserved()). The per-pool and per-host -/// aggregates are computed in UpdateClusterAggregates(). This state, once all updates -/// are fully distributed and aggregated, provides enough information to make -/// admission decisions by any impalad. However, this requires waiting for both +/// memory reserved by fragments that have begun execution. For queries that have mem +/// limits, the limit is considered to be its reserved memory, otherwise the current +/// consumption is used (see MemTracker::GetPoolMemReserved()). The per-pool and +/// per-host aggregates are computed in UpdateClusterAggregates(). This state, once +/// all updates are fully distributed and aggregated, provides enough information to +/// make admission decisions by any impalad. However, this requires waiting for both /// admitted requests to start all remote fragments and then for the updated state to /// be distributed via the statestore. /// b) Mem Admitted: the amount of memory required (i.e. the value used in admission, @@ -197,7 +196,7 @@ class AdmissionController { /// been submitted via AdmitQuery(). (If the request was not admitted, this is /// a no-op.) /// This does not block. - void ReleaseQuery(const QuerySchedule& schedule); + Status ReleaseQuery(QuerySchedule* schedule); /// Registers the request queue topic with the statestore. Status Init(); http://git-wip-us.apache.org/repos/asf/incubator-impala/blob/a772f845/be/src/service/client-request-state.cc ---------------------------------------------------------------------- diff --git a/be/src/service/client-request-state.cc b/be/src/service/client-request-state.cc index 5d00d24..38c556d 100644 --- a/be/src/service/client-request-state.cc +++ b/be/src/service/client-request-state.cc @@ -572,6 +572,17 @@ void ClientRequestState::Done() { // Update result set cache metrics, and update mem limit accounting before tearing // down the coordinator. ClearResultCache(); + + if (coord_.get() != NULL) { + // Release any reserved resources. + if (exec_env_->admission_controller() != nullptr) { + Status status = exec_env_->admission_controller()->ReleaseQuery(schedule_.get()); + if (!status.ok()) { + LOG(WARNING) << "Failed to release resources of query " << schedule_->query_id() + << " because of error: " << status.GetDetail(); + } + } + } } Status ClientRequestState::Exec(const TMetadataOpRequest& exec_request) { http://git-wip-us.apache.org/repos/asf/incubator-impala/blob/a772f845/tests/custom_cluster/test_admission_controller.py ---------------------------------------------------------------------- diff --git a/tests/custom_cluster/test_admission_controller.py b/tests/custom_cluster/test_admission_controller.py index 02fd6db..eb1e942 100644 --- a/tests/custom_cluster/test_admission_controller.py +++ b/tests/custom_cluster/test_admission_controller.py @@ -40,25 +40,24 @@ from TCLIService import TCLIService LOG = logging.getLogger('admission_test') -# The query used for testing. It is important that this query be able to fetch many -# rows. This allows a thread to stay active by fetching one row at a time. The -# where clause is for debugging purposes; each thread will insert its id so +# We set a WAIT debug action so it doesn't complete the execution of this query. The +# limit is a parameter for debugging purposes; each thread will insert its id so # that running queries can be correlated with the thread that submitted them. -QUERY = "select * from alltypes where id != %s" +QUERY = "select * from alltypes where id != %s"# limit %s" # Time to sleep (in milliseconds) between issuing queries. The default statestore # heartbeat is 500ms, so the lower the delay the more we can submit before the global # state is updated. When the delay is at least the statestore heartbeat frequency, all # state should be visible by every impalad by the time the next query is submitted. -SUBMISSION_DELAY_MS = [0, 50, 100, 600] +SUBMISSION_DELAY_MS = [50] # The number of queries to submit. The test does not support fewer queries than # MAX_NUM_CONCURRENT_QUERIES + MAX_NUM_QUEUED_QUERIES to keep some validation logic # simple. -NUM_QUERIES = [15, 30, 50] +NUM_QUERIES = [50] # Whether we will submit queries to all available impalads (in a round-robin fashion) -ROUND_ROBIN_SUBMISSION = [True, False] +ROUND_ROBIN_SUBMISSION = [True] # The query pool to use. The impalads should be configured to recognize this # pool with the parameters below. @@ -87,9 +86,6 @@ _STATESTORED_ARGS = "-statestore_heartbeat_frequency_ms=%s "\ # Key in the query profile for the query options. PROFILE_QUERY_OPTIONS_KEY = "Query Options (set by configuration): " -# The different ways that a query thread can end its query. -QUERY_END_BEHAVIORS = ['EOS', 'CLIENT_CANCEL', 'QUERY_TIMEOUT', 'CLIENT_CLOSE'] - def impalad_admission_ctrl_flags(max_requests, max_queued, pool_max_mem, proc_mem_limit = None): if proc_mem_limit is not None: @@ -369,13 +365,12 @@ class TestAdmissionController(TestAdmissionControllerBase, HS2TestSuite): class TestAdmissionControllerStress(TestAdmissionControllerBase): """Submits a number of queries (parameterized) with some delay between submissions (parameterized) and the ability to submit to one impalad or many in a round-robin - fashion. Each query is submitted on a separate thread. After admission, the query - thread will block with the query open and wait for the main thread to notify it to - end its query. The query thread can end its query by fetching to the end, cancelling - itself, closing itself, or waiting for the query timeout to take effect. Depending - on the test parameters a varying number of queries will be admitted, queued, and - rejected. After the queries are admitted, the main thread will request each admitted - query thread to end its query and allow queued queries to be admitted. + fashion. The queries are set with the WAIT debug action so that we have more control + over the state that the admission controller uses to make decisions. Each query is + submitted on a separate thread. Depending on the test parameters a varying number of + queries will be admitted, queued, and rejected. Once queries are admitted, the query + execution blocks and we can cancel the query in order to allow another queued query to + be admitted. The test tracks the state of the admission controller using the metrics from each impalad to do the following: @@ -383,15 +378,13 @@ class TestAdmissionControllerStress(TestAdmissionControllerBase): queued, and rejected requests should sum to the number of queries and that the values are reasonable given the test parameters. (2) While there are running queries: - * Request the currently running queries to end and wait for the queries to end. - Verify the metric for the number of completed queries. The threads that - submitted those queries will keep their connections open until the entire test - completes. This verifies that admission control is tied to the end of the query - and does not depend on closing the connection. + * Cancel the currently running queries (they are blocked with the WAIT debug action) + and verify the metric for the number of completed queries. The threads that + submitted those queries should complete. * Check that queued requests are then dequeued and verify using the metric for the number of dequeued requests. The threads that were waiting to submit the query should then insert themselves into a list of currently running queries and then - wait for a notification from the main thread. + fetch() the results (which will block). (3) After all queries have completed, check that the final number of admitted, queued, and rejected requests are reasonable given the test parameters. When submitting to a single impalad, we know exactly what the values should be, @@ -435,7 +428,6 @@ class TestAdmissionControllerStress(TestAdmissionControllerBase): self.executing_threads = list() def teardown(self): - # Set shutdown for all threads (cancel if needed) for thread in self.all_threads: try: thread.lock.acquire() @@ -450,9 +442,6 @@ class TestAdmissionControllerStress(TestAdmissionControllerBase): client.close() finally: thread.lock.release() - - # Wait for all threads to exit - for thread in self.all_threads: thread.join(5) LOG.debug("Join thread for query num %s %s", thread.query_num, "TIMED OUT" if thread.isAlive() else "") @@ -548,39 +537,36 @@ class TestAdmissionControllerStress(TestAdmissionControllerBase): LOG.debug("Found all %s admitted threads after %s seconds", num_threads, round(time() - start_time, 1)) - def end_admitted_queries(self, num_queries): + def cancel_admitted_queries(self, num_queries): """ - Requests each admitted query to end its query. + Cancels queries on threads that are currently blocked on query execution. """ assert len(self.executing_threads) >= num_queries - LOG.debug("Requesting {0} clients to end queries".format(num_queries)) - - # Request admitted clients to end their queries - current_executing_queries = [] + LOG.debug("Cancelling %s queries", num_queries) for i in xrange(num_queries): # pop() is thread-safe, it's OK if another thread is appending concurrently. thread = self.executing_threads.pop(0) LOG.debug("Cancelling query %s", thread.query_num) + # The other thread sets the query_state before appending itself to the list, + # and will not change its state until it is cancelled by this thread. assert thread.query_state == 'ADMITTED' - current_executing_queries.append(thread) - thread.query_state = 'REQUEST_QUERY_END' - - # Wait for the queries to end - start_time = time() - while True: - all_done = True - for thread in self.all_threads: - if thread.query_state == 'REQUEST_QUERY_END': - all_done = False - if all_done: - break - assert (time() - start_time < STRESS_TIMEOUT),\ - "Timed out waiting %s seconds for query end" % (STRESS_TIMEOUT,) - sleep(1) + client = thread.impalad.service.create_beeswax_client() + try: + cancel_result = client.cancel(thread.query_handle) + assert cancel_result.status_code == 0,\ + 'Unexpected status code from cancel request: %s' % cancel_result + # Wait for the query to be cancelled and return + thread.join(20) + LOG.debug("Cancelled admitted query %s %s", + thread.query_num, "TIMED OUT" if thread.isAlive() else "") + assert not thread.isAlive() + assert thread.query_state == 'COMPLETED' + finally: + client.close() class SubmitQueryThread(threading.Thread): def __init__(self, impalad, additional_query_options, vector, query_num, - query_end_behavior, executing_threads): + executing_threads): """ executing_threads must be provided so that this thread can add itself when the query is admitted and begins execution. @@ -590,7 +576,6 @@ class TestAdmissionControllerStress(TestAdmissionControllerBase): self.vector = vector self.additional_query_options = additional_query_options self.query_num = query_num - self.query_end_behavior = query_end_behavior self.impalad = impalad self.error = None # query_state is defined and used only by the test code, not a property exposed by @@ -614,6 +599,7 @@ class TestAdmissionControllerStress(TestAdmissionControllerBase): return exec_options = self.vector.get_value('exec_option') + exec_options['debug_action'] = '0:GETNEXT:WAIT' exec_options.update(self.additional_query_options) query = QUERY % (self.query_num,) self.query_state = 'SUBMITTING' @@ -621,9 +607,6 @@ class TestAdmissionControllerStress(TestAdmissionControllerBase): ImpalaTestSuite.change_database(client, self.vector.get_value('table_format')) client.set_configuration(exec_options) - if self.query_end_behavior == 'QUERY_TIMEOUT': - client.execute("SET QUERY_TIMEOUT_S=5") - LOG.debug("Submitting query %s", self.query_num) self.query_handle = client.execute_async(query) except ImpalaBeeswaxException as e: @@ -644,22 +627,22 @@ class TestAdmissionControllerStress(TestAdmissionControllerBase): # The thread becomes visible to the main thread when it is added to the # shared list of executing_threads. append() is atomic and thread-safe. self.executing_threads.append(self) - - # Synchronize with the main thread. At this point, the thread is executing a - # query. It needs to wait until the main thread requests it to end its query. - while not self.shutdown: - # The QUERY_TIMEOUT needs to stay active until the main thread requests it - # to end. Otherwise, the query may get cancelled early. Fetch a row every - # second to avoid going idle. - if self.query_end_behavior == 'QUERY_TIMEOUT' and \ - self.query_state != 'COMPLETED': - client.fetch(query, self.query_handle, 1) - if self.query_state == 'REQUEST_QUERY_END': - self._end_query(client, query) - # The query has released admission control resources + try: + # fetch() will block until we cancel the query from the main thread + # (unless an unexpected error occurs). If an error occurs on the main therad, + # it is possible that teardown() cancels this query before we call fetch(). In + # that case a different exception is thrown and we handle it gracefully. + client.fetch(query, self.query_handle) + except ImpalaBeeswaxException as e: + if "Cancelled" in str(e): + LOG.debug("Query %s completed", self.query_num) self.query_state = 'COMPLETED' self.query_handle = None - sleep(1) + elif "Invalid or unknown query handle" in str(e): + # May happen if the test is being torn down early (i.e. an error occurred). + LOG.debug("Query %s already cancelled in test shutdown.") + else: + raise e except Exception as e: LOG.exception(e) # Unknown errors will be raised later @@ -670,27 +653,6 @@ class TestAdmissionControllerStress(TestAdmissionControllerBase): if client is not None: client.close() - def _end_query(self, client, query): - """Bring the query to the appropriate end state defined by self.query_end_behaviour. - Returns once the query has reached that state.""" - if self.query_end_behavior == 'QUERY_TIMEOUT': - # Sleep and wait for the query to be cancelled. The cancellation will - # set the state to EXCEPTION. - start_time = time() - while (client.get_state(self.query_handle) != \ - client.QUERY_STATES['EXCEPTION']): - assert (time() - start_time < STRESS_TIMEOUT),\ - "Timed out waiting %s seconds for query cancel" % (STRESS_TIMEOUT,) - sleep(1) - elif self.query_end_behavior == 'EOS': - # Fetch all rows so we hit eos. - client.fetch(query, self.query_handle) - elif self.query_end_behavior == 'CLIENT_CANCEL': - client.cancel(self.query_handle) - else: - assert self.query_end_behavior == 'CLIENT_CLOSE' - client.close_query(self.query_handle) - def _check_queries_page_resource_pools(self): """Checks that all queries in the '/queries' webpage json have the correct resource pool (this is called after all queries have been admitted, queued, or rejected, so @@ -731,11 +693,14 @@ class TestAdmissionControllerStress(TestAdmissionControllerBase): initial_metrics = self.get_admission_metrics(); log_metrics("Initial metrics: ", initial_metrics); - for query_num in xrange(num_queries): + # Want query_num to start at 1 because this gets used as the limit in the query to + # help debugging (we can associate a running query with a thread). If we start at 0, + # that query would be evaluated as a constant expression and never hit the WAIT debug + # action. + for query_num in xrange(1, num_queries + 1): impalad = self.impalads[query_num % len(self.impalads)] - query_end_behavior = QUERY_END_BEHAVIORS[query_num % len(QUERY_END_BEHAVIORS)] thread = self.SubmitQueryThread(impalad, additional_query_options, vector, - query_num, query_end_behavior, self.executing_threads) + query_num, self.executing_threads) thread.start() self.all_threads.append(thread) sleep(submission_delay_ms / 1000.0) @@ -770,10 +735,10 @@ class TestAdmissionControllerStress(TestAdmissionControllerBase): while len(self.executing_threads) > 0: curr_metrics = self.get_admission_metrics(); log_metrics("Main loop, curr_metrics: ", curr_metrics); - num_to_end = len(self.executing_threads) - LOG.debug("Main loop, will request %s queries to end", num_to_end) - self.end_admitted_queries(num_to_end) - self.wait_for_metric_changes(['released'], curr_metrics, num_to_end) + num_to_cancel = len(self.executing_threads) + LOG.debug("Main loop, will cancel %s queries", num_to_cancel) + self.cancel_admitted_queries(num_to_cancel) + self.wait_for_metric_changes(['released'], curr_metrics, num_to_cancel) num_queued_remaining =\ curr_metrics['queued'] - curr_metrics['dequeued'] - curr_metrics['timed-out']
