ExecutorResourceMgr is the central manager for resources available to all executors in the system. It manages an ExecutorResourcePool to keep track of available and allocated resources (currently CPU slots/threads, GPUs, CPU result memory, and CPU and GPU buffer pool memory). It also manages a thread queue which keeps requesting threads (from Executor::launchKernelsViaResourceMgr) waiting until there it can schedule them. At that point, it gives the calling executor thread a ResourceHandle detailing the resources granted to the query, which once it goes out of scope will return the granted resources to the ExecutorResourcePool. More...

#include <ExecutorResourceMgr.h>

Inheritance diagram for ExecutorResourceMgr_Namespace::ExecutorResourceMgr:

Collaboration diagram for ExecutorResourceMgr_Namespace::ExecutorResourceMgr:

Public Member Functions
	ExecutorResourceMgr (const std::vector< std::pair< ResourceType, size_t >> &total_resources, const std::vector< ConcurrentResourceGrantPolicy > &concurrent_resource_grant_policies, const std::vector< ResourceGrantPolicy > &max_per_request_resource_grant_policies, const double max_available_resource_use_ratio)
	The constructor instantiates an `ExecutorResourcePool` with the provided parameters, and starts the process queue by launching a thread to invoke process_queue_loop. More...

	~ExecutorResourceMgr ()
	The destructor ensures that the process queue thread (`process_queue_thread`) is stopped and that any threads waiting for resources are joined. Currently only called on database shutdown. More...

std::unique_ptr < ExecutorResourceHandle >	request_resources_with_timeout (const RequestInfo &request_info, const size_t timeout_in_ms)
	Requests resources from `ExecutorResourceMgr`, will throw if request takes longer than time specified by timeout_in_ms. More...

std::unique_ptr < ExecutorResourceHandle >	request_resources (const RequestInfo &request_info)
	Requests resources from `ExecutorResourceMgr`, with no timeout (unlike `request_resources_with_timeout`) More...

void	release_resources (const RequestId request_id, const ResourceGrant &resource_grant)
	Instructs `ExecutorResourceMgr` that the resources held by the requestor with the given `request_id` can be freed/returned to the `ExecutorResourcePool`. More...

ExecutorStats	get_executor_stats () const
	Returns a copy of the `ExecutorStats` struct held by `ExecutorResourceMgr`. Used for testing currently. More...

void	print_executor_stats () const
	Prints the `ExecutorStats` struct. Use for debugging. More...

std::pair< size_t, size_t >	get_resource_info (const ResourceType resource_type) const
	Returns the allocated and total available amount of the resource specified. More...

ResourcePoolInfo	get_resource_info () const
	Returns a struct containing the total and allocated amounts of all resources tracked by `ExecutorResourceMgr`/`ExecutorResourcePool`, as well as the number of outstanding requests (in total and by resource type) More...

void	set_resource (const ResourceType resource_type, const size_t resoure_quantity)
	Used to change the total amount available of a specified resource after construction of `ExecutorResourceMgr` More...

ConcurrentResourceGrantPolicy	get_concurrent_resource_grant_policy (const ResourceType resource_type) const
	Get the concurrent resource grant policy for a given resource type. More...

void	set_concurrent_resource_grant_policy (const ConcurrentResourceGrantPolicy &concurrent_resource_grant_policy)
	Set the concurrent resource grant policy for a given resource type (stored in `ConcurrentResourceGrantPolicy`) More...

void	pause_process_queue ()
	Pauses the process queue in a thread-safe manner, waiting for all queries in the executing stage to finish before yielding to the caller (ensuring that the ExecutorResourcePool has no outstanding allocations). If the process queue is already paused, the call is a no-op. This method is used to live-change parameters associated with ExecutorResourcePool. More...

void	resume_process_queue ()
	Resumes the process queue in a thread-safe manner. If the process queue is already paused, the call is a no-op. More...

Private Member Functions
void	process_queue_loop ()
	Internal method: A thread is assigned to run this function in the constructor of `ExecutorResourceMgr`, where it loops continuously waiting for changes to the process queue (i.e. the introduction of a new resource request or the finish of an existing request). More...

RequestStats	get_request_for_id (const RequestId request_id) const
	Internal method: Returns the `RequestStats` for a request specified by `request_id`. More...

void	mark_request_error (const RequestId request_id, std::string error_msg)

RequestId	choose_next_request ()
	Internal method: Invoked from `process_queue_loop`, chooses the next resource request to grant. More...

RequestId	enqueue_request (const RequestInfo &request_info, const size_t timeout_in_ms, const ResourceGrant &min_resource_grant, const ResourceGrant &max_resource_grant)
	Internal method: Invoked from request_resource/request_resource_with_timeout, places request in the request queue where it will be wait to be granted the resources requested. More...

void	mark_request_dequed (const RequestId request_id)
	Internal method: Moves the request from the `QUEUED` stage to `EXECUTING` stage and performs other bookkeeping. More...

void	mark_request_timed_out (const RequestId request_id)
	Internal method: Called if the request times out (i.e. request was made via `request_resources_with_timeout`), moves request out of `QUEUED` stage and does other bookkeeping on the request's `RequestStats` and on `ExecutorStats`. More...

void	mark_request_finished (const RequestId request_id)
	Internal method: Invoked on successful completion of a query step from release_resources method, removes request from `EXECUTING` stage and performs various bookkeeping, including recording execution and final times in request_stats. More...

void	set_process_queue_flag ()
	Internal method: Set the `should_process_queue_` flag to true, signifying that the queue should be processed. Protected by a lock on `processor_queue_mutex_`. More...

void	stop_process_queue_thread ()
	Internal method: Invoked from `ExecutorResourceMgr` destructor, sets `stop_process_queue_thread_` to true (behind a lock on `processor_queue_mutex_`) and then attempts to join all threads left in the request queue on server shutdown. More...

std::vector< RequestId >	get_requests_for_stage (const ExecutionRequestStage request_status) const
	Internal method: Get the request ids for a given stage (`QUEUED` or `EXECUTING`) More...

void	add_request_to_stage (const RequestId request_id, const ExecutionRequestStage request_status)
	Internal method: Adds the request specified by the provided `request_id` to the specified stage. More...

void	remove_request_from_stage (const RequestId request_id, const ExecutionRequestStage request_status)
	Internal method: Removes the request specified by the provided `request_id` from the specified stage. More...

ChunkRequestInfo	get_chunk_request_info (const RequestId request_id)
	Get the `DataMgr` chunk ids and associated sizes pertaining to the input data needed by a request. More...

Private Attributes
ExecutorResourcePool	executor_resource_pool_
	Keeps track of available resources for execution. More...

std::atomic< size_t >	requests_count_ {0}
	An atomic that is incremented with each incoming request, and used to assign `RequestIds` to incoming request. More...

ExecutorStats	executor_stats_
	Holds a single `ExecutorStats` struct that pertains to cummulative stats for `ExecutorResourceMgr`, i.e. number of requests, queue length, total execution time, etc. More...

const size_t	ACTUALLY_QUEUED_MIN_MS {2}

std::thread	process_queue_thread_
	The thread started in the `ExecutorResourceMgr` constructor that continuously loops inside of `process_queue_loop` to determine the next resource request that should be granted. More...

std::mutex	processor_queue_mutex_
	RW mutex that protects access to `stop_process_queue_thread_` and `pause_processor_queue_` More...

std::mutex	pause_processor_queue_mutex_

std::mutex	print_mutex_

std::shared_mutex	queue_stats_mutex_
	RW mutex that protects access to `executor_stats_` and `request_stats_` More...

std::shared_mutex	queued_set_mutex_
	RW mutex that protects access to `queued_requests_` More...

std::shared_mutex	executing_set_mutex_
	RW mutex that protects access to `executing_requests_` More...

std::condition_variable	processor_queue_condition_

std::condition_variable	pause_processor_queue_condition_

bool	should_process_queue_ {false}

bool	stop_process_queue_thread_ {false}

bool	pause_process_queue_ {false}

bool	process_queue_is_paused_ {false}

size_t	process_queue_counter_ {0}

OutstandingQueueRequests	outstanding_queue_requests_
	Stores and manages a map of request ids to `BinarySemaphore` objects to allow threads waiting for resources to be selectively queued/blocked and then when they are choosen for resource grants/execution, woken. More...

std::set< RequestId >	queued_requests_
	Set of all request ids that are currently queued. Protected by `queued_set_mutex_`. More...

std::set< RequestId >	executing_requests_
	Set of all request ids that are currently executing (i.e. post-granting of resources). Protected by `executing_set_mutex_`. More...

std::vector< RequestStats >	requests_stats_
	Stores a vector of all requests that have been seen by `ExecutorResourceMgr`, with each incoming request appending a `RequestStats` struct to this vector. Protected by `queue_stats_mutex_`. More...

const bool	enable_stats_printing_ {false}

const bool	enable_debug_printing_ {false}

const RequestId	INVALID_REQUEST_ID {std::numeric_limits<size_t>::max()}

const double	max_available_resource_use_ratio_

Detailed Description

ExecutorResourceMgr is the central manager for resources available to all executors in the system. It manages an ExecutorResourcePool to keep track of available and allocated resources (currently CPU slots/threads, GPUs, CPU result memory, and CPU and GPU buffer pool memory). It also manages a thread queue which keeps requesting threads (from Executor::launchKernelsViaResourceMgr) waiting until there it can schedule them. At that point, it gives the calling executor thread a ResourceHandle detailing the resources granted to the query, which once it goes out of scope will return the granted resources to the ExecutorResourcePool.

Definition at line 137 of file ExecutorResourceMgr.h.

Constructor & Destructor Documentation

ExecutorResourceMgr_Namespace::ExecutorResourceMgr::ExecutorResourceMgr	(	const std::vector< std::pair< ResourceType, size_t >> &	total_resources,
		const std::vector< ConcurrentResourceGrantPolicy > &	concurrent_resource_grant_policies,
		const std::vector< ResourceGrantPolicy > &	max_per_request_resource_grant_policies,
		const double	max_available_resource_use_ratio
	)

The constructor instantiates an ExecutorResourcePool with the provided parameters, and starts the process queue by launching a thread to invoke process_queue_loop.

Definition at line 25 of file ExecutorResourceMgr.cpp.

References CHECK_GT, CHECK_LE, logger::EXECUTOR, LOG, max_available_resource_use_ratio_, process_queue_loop(), and process_queue_thread_.

     : executor_resource_pool_(total_resources,
                               concurrent_resource_grant_policies,
                               max_per_request_resource_grant_policies)
     , max_available_resource_use_ratio_(max_available_resource_use_ratio) {
   CHECK_GT(max_available_resource_use_ratio_, 0.0);
   CHECK_LE(max_available_resource_use_ratio_, 1.0);
   process_queue_thread_ = std::thread(&ExecutorResourceMgr::process_queue_loop, this);
   LOG(EXECUTOR) << "Executor Resource Manager queue proccessing thread started";
 }

Here is the call graph for this function:

ExecutorResourceMgr_Namespace::ExecutorResourceMgr::~ExecutorResourceMgr ( )

The destructor ensures that the process queue thread (process_queue_thread) is stopped and that any threads waiting for resources are joined. Currently only called on database shutdown.

Definition at line 40 of file ExecutorResourceMgr.cpp.

References stop_process_queue_thread().

                                           {
   stop_process_queue_thread();
 }

Here is the call graph for this function:

Member Function Documentation

void ExecutorResourceMgr_Namespace::ExecutorResourceMgr::add_request_to_stage	(	const RequestId	request_id,
		const ExecutionRequestStage	request_status
	)

private

Internal method: Adds the request specified by the provided request_id to the specified stage.

Parameters

request_id	- Request id to add to the specified stage
request_status	- Stage (`QUEUED` or `EXECUTING`) to add the specified request to

Definition at line 576 of file ExecutorResourceMgr.cpp.

References CHECK, executing_requests_, executing_set_mutex_, ExecutorResourceMgr_Namespace::QUEUED, queued_requests_, and queued_set_mutex_.

Referenced by enqueue_request(), and mark_request_dequed().

                                                {
   auto& chosen_set = request_stage == ExecutionRequestStage::QUEUED ? queued_requests_
                                                                     : executing_requests_;
   auto& chosen_mutex = request_stage == ExecutionRequestStage::QUEUED
                            ? queued_set_mutex_
                            : executing_set_mutex_;
   std::unique_lock<std::shared_mutex> set_write_lock(chosen_mutex);
 
   CHECK(chosen_set.insert(request_id)
             .second);  // Should return true as element should not exist in set
 }

Here is the caller graph for this function:

RequestId ExecutorResourceMgr_Namespace::ExecutorResourceMgr::choose_next_request ( )

private

Internal method: Invoked from process_queue_loop, chooses the next resource request to grant.

Currently based on FIFO logic, choosing the oldest request in the queue that we have enough resources in ExecutorResourcePool to fulfill. Future variants could add more sophisticated logic for more optimal query scheduling (i.e. non-FIFO).

Returns: RequestId - The id of the request we will grant resources for.

Definition at line 133 of file ExecutorResourceMgr.cpp.

References ExecutorResourceMgr_Namespace::ExecutorResourcePool::determine_dynamic_resource_grant(), enable_debug_printing_, logger::EXECUTOR, executor_resource_pool_, get_requests_for_stage(), INVALID_REQUEST_ID, LOG, max_available_resource_use_ratio_, print_mutex_, process_queue_counter_, queue_stats_mutex_, ExecutorResourceMgr_Namespace::QUEUED, logger::request_id(), and requests_stats_.

Referenced by process_queue_loop().

                                                    {
   const auto request_ids = get_requests_for_stage(ExecutionRequestStage::QUEUED);
   LOG(EXECUTOR) << "ExecutorResourceMgr Queue Itr: " << process_queue_counter_ - 1
                 << " Queued requests: " << request_ids.size();
   std::unique_lock<std::shared_mutex> queue_stats_lock(queue_stats_mutex_);
   for (const auto request_id : request_ids) {
     auto& request_stats = requests_stats_[request_id];
     try {
       const auto actual_resource_grant =
           executor_resource_pool_.determine_dynamic_resource_grant(
               request_stats.min_resource_grant,
               request_stats.max_resource_grant,
               request_stats.request_info.chunk_request_info,
               max_available_resource_use_ratio_);
       // boolean sentinel first member of returned pair says whether
       // a resource grant was able to be made at all
       if (actual_resource_grant.first) {
         request_stats.actual_resource_grant = actual_resource_grant.second;
         LOG(EXECUTOR) << "ExecutorResourceMgr Queue chosen request ID: " << request_id
                       << " from " << request_ids.size() << " queued requests.";
         LOG(EXECUTOR) << "Request grant: " << actual_resource_grant.second.to_string();
         if (enable_debug_printing_) {
           std::unique_lock<std::mutex> print_lock(print_mutex_);
           std::cout << std::endl << "Actual grant";
           actual_resource_grant.second.print();
         }
         return request_id;
       }
     } catch (std::runtime_error const& e) {
       throw ExecutorResourceMgrError(request_id, e.what());
     }
   }
   return INVALID_REQUEST_ID;
 }

Here is the call graph for this function:

Here is the caller graph for this function:

RequestId ExecutorResourceMgr_Namespace::ExecutorResourceMgr::enqueue_request	(	const RequestInfo &	request_info,
		const size_t	timeout_in_ms,
		const ResourceGrant &	min_resource_grant,
		const ResourceGrant &	max_resource_grant
	)

private

Internal method: Invoked from request_resource/request_resource_with_timeout, places request in the request queue where it will be wait to be granted the resources requested.

Note that this method assigns a RequestId to the request, adds the request to the RequestStats, and adds the assigned request_id to the QUEUED stage, but does not actually notify the process queue or queue the request in the OutstandQueueRequests object. Those tasks are done subsequently to invocation of this method in the parent call (request_resources_with_timeout)

Parameters

request_info	- info for the current resource request
timeout_in_ms	- request timeout
min_resource_grant	- min allowable resource request, calculated in caller `request_resources_with_timeout` method from request_info
max_resource_grant	- max (ideal) resource request, calculated in caller `request_resources_with_timeout` method from request_info

Returns: RequestId - generated request id for this request

Definition at line 391 of file ExecutorResourceMgr.cpp.

Referenced by request_resources_with_timeout().

                                                                                         {
   const std::chrono::steady_clock::time_point enqueue_time =
       std::chrono::steady_clock::now();
   std::unique_lock<std::shared_mutex> queue_stats_write_lock(queue_stats_mutex_);
   const RequestId request_id = requests_count_.fetch_add(1, std::memory_order_relaxed);
   executor_stats_.requests++;
   if (timeout_in_ms > 0) {
     executor_stats_.requests_with_timeouts++;
   }
   const size_t queue_length_at_entry = executor_stats_.queue_length++;
   executor_stats_.sum_queue_size_at_entry += queue_length_at_entry;
   size_t device_type_queue_length_at_entry{0};
   switch (request_info.request_device_type) {
     case ExecutorDeviceType::CPU: {
       executor_stats_.cpu_requests++;
       device_type_queue_length_at_entry = executor_stats_.cpu_queue_length++;
       executor_stats_.sum_cpu_queue_size_at_entry += device_type_queue_length_at_entry;
       break;
     }
     case ExecutorDeviceType::GPU: {
       executor_stats_.gpu_requests++;
       device_type_queue_length_at_entry = executor_stats_.gpu_queue_length++;
       executor_stats_.sum_gpu_queue_size_at_entry += device_type_queue_length_at_entry;
       break;
     }
     default:
       UNREACHABLE();
   }
 
   requests_stats_.emplace_back(RequestStats(request_id,
                                             request_info,
                                             min_resource_grant,
                                             max_resource_grant,
                                             enqueue_time,
                                             queue_length_at_entry,
                                             device_type_queue_length_at_entry,
                                             timeout_in_ms));
   add_request_to_stage(request_id, ExecutionRequestStage::QUEUED);
   return request_id;
 }

Here is the call graph for this function:

Here is the caller graph for this function:

ChunkRequestInfo ExecutorResourceMgr_Namespace::ExecutorResourceMgr::get_chunk_request_info ( const RequestId request_id )

private

Get the DataMgr chunk ids and associated sizes pertaining to the input data needed by a request.

Parameters

request_id - Request id to fetch the needed chunks for

Returns: ChunkRequestInfo - struct containing vector of ChunkKey and byte sizes as well as device memory space (CPU or GPU), total bytes, etc

Definition at line 604 of file ExecutorResourceMgr.cpp.

References queue_stats_mutex_, logger::request_id(), and requests_stats_.

Referenced by release_resources().

                                                                                        {
   std::shared_lock<std::shared_mutex> queue_stats_read_lock(queue_stats_mutex_);
   return requests_stats_[request_id].request_info.chunk_request_info;
 }

Here is the call graph for this function:

Here is the caller graph for this function:

ConcurrentResourceGrantPolicy ExecutorResourceMgr_Namespace::ExecutorResourceMgr::get_concurrent_resource_grant_policy ( const ResourceType resource_type ) const

Get the concurrent resource grant policy for a given resource type.

Queries the ExecutorResourcePool for the current concurrency policies (including normal and oversubscribed) for a resource type

Parameters

resource_type - Type of resource to get the concurrency policy for

Returns: const ConcurrentResourceGrantPolicy& - Specifies the chosen concurrency policy for normal operation and when the specified resource is oversubscribed

Definition at line 293 of file ExecutorResourceMgr.cpp.

References executor_resource_pool_, and ExecutorResourceMgr_Namespace::ExecutorResourcePool::get_concurrent_resource_grant_policy().

                                             {
   return executor_resource_pool_.get_concurrent_resource_grant_policy(resource_type);
 }

Here is the call graph for this function:

ExecutorStats ExecutorResourceMgr_Namespace::ExecutorResourceMgr::get_executor_stats ( ) const

Returns a copy of the ExecutorStats struct held by ExecutorResourceMgr. Used for testing currently.

Definition at line 168 of file ExecutorResourceMgr.cpp.

References executor_stats_, and queue_stats_mutex_.

Referenced by print_executor_stats().

                                                             {
   std::shared_lock<std::shared_mutex> queue_stats_read_lock(queue_stats_mutex_);
   return executor_stats_;  // Will make copy
 }

Here is the caller graph for this function:

RequestStats ExecutorResourceMgr_Namespace::ExecutorResourceMgr::get_request_for_id ( const RequestId request_id ) const

private

Internal method: Returns the RequestStats for a request specified by request_id.

Takes a read lock on queue_stats_mutex_ for thread safety. Note this method should not be used in internal methods where a lock is already taken on queue_stats_mutex_.

Parameters

request_id - The RequestId for the request we want to retrieve stats for

Returns: RequestStats - The stats for the specified request, including resources granted

Definition at line 120 of file ExecutorResourceMgr.cpp.

References CHECK_LT, queue_stats_mutex_, logger::request_id(), and requests_stats_.

Referenced by process_queue_loop().

                                                                                      {
   std::shared_lock<std::shared_mutex> queue_stats_read_lock(queue_stats_mutex_);
   CHECK_LT(request_id, requests_stats_.size());
   return requests_stats_[request_id];
 }

Here is the call graph for this function:

Here is the caller graph for this function:

std::vector< RequestId > ExecutorResourceMgr_Namespace::ExecutorResourceMgr::get_requests_for_stage ( const ExecutionRequestStage request_status ) const

private

Internal method: Get the request ids for a given stage (QUEUED or EXECUTING)

Invoked from choose_next_request to get all outstanding queued requests

Parameters

request_status - request stage type to fetch request ids for

Returns: std::vector<RequestId> - vector of request ids in the requested stage

Definition at line 562 of file ExecutorResourceMgr.cpp.

References executing_requests_, executing_set_mutex_, ExecutorResourceMgr_Namespace::QUEUED, queued_requests_, and queued_set_mutex_.

Referenced by choose_next_request().

                                                      {
   auto& chosen_set = request_stage == ExecutionRequestStage::QUEUED ? queued_requests_
                                                                     : executing_requests_;
   auto& chosen_mutex = request_stage == ExecutionRequestStage::QUEUED
                            ? queued_set_mutex_
                            : executing_set_mutex_;
   std::shared_lock<std::shared_mutex> set_read_lock(chosen_mutex);
 
   const std::vector<RequestId> request_ids_for_stage(chosen_set.begin(),
                                                      chosen_set.end());
   return request_ids_for_stage;
 }

Here is the caller graph for this function:

std::pair<size_t, size_t> ExecutorResourceMgr_Namespace::ExecutorResourceMgr::get_resource_info ( const ResourceType resource_type ) const

inline

Returns the allocated and total available amount of the resource specified.

Interally queries ExecutorResourcePool

Returns: std::pair<size_t, size_t> - First member is the allocated amount of the resource, the second member is the total amount of the resource (including allocated and available)

Definition at line 224 of file ExecutorResourceMgr.h.

References executor_resource_pool_, and ExecutorResourceMgr_Namespace::ExecutorResourcePool::get_resource_info().

                                                                                     {
     return executor_resource_pool_.get_resource_info(resource_type);
   }

Here is the call graph for this function:

ResourcePoolInfo ExecutorResourceMgr_Namespace::ExecutorResourceMgr::get_resource_info ( ) const

inline

Returns a struct containing the total and allocated amounts of all resources tracked by ExecutorResourceMgr/ExecutorResourcePool, as well as the number of outstanding requests (in total and by resource type)

Returns: ResourcePoolInfo - Defined in ExecutorResourcePool.h, contains total and allocated amounts for all resources, and total requests outstanding and per type of resource

Definition at line 237 of file ExecutorResourceMgr.h.

References executor_resource_pool_, and ExecutorResourceMgr_Namespace::ExecutorResourcePool::get_resource_info().

Referenced by set_resource().

                                              {
     return executor_resource_pool_.get_resource_info();
   }

Here is the call graph for this function:

Here is the caller graph for this function:

void ExecutorResourceMgr_Namespace::ExecutorResourceMgr::mark_request_dequed ( const RequestId request_id )

private

Internal method: Moves the request from the QUEUED stage to EXECUTING stage and performs other bookkeeping.

Invoked by process_queue_loop after determing the next resource request to serve (via choose_next_request).

Parameters

request_id - RequestId of the request being moved out of the request queue (to be executed)

Definition at line 435 of file ExecutorResourceMgr.cpp.

Referenced by process_queue_loop().

                                                                         {
   const std::chrono::steady_clock::time_point deque_time =
       std::chrono::steady_clock::now();
   // Below is only to CHECK our request_id against high water mark... should be
   // relatively inexpensive though
   const size_t current_request_count = requests_count_.load(std::memory_order_relaxed);
   CHECK_LT(request_id, current_request_count);
   {
     std::unique_lock<std::shared_mutex> queue_stats_write_lock(queue_stats_mutex_);
     RequestStats& request_stats = requests_stats_[request_id];
     request_stats.deque_time = deque_time;
     request_stats.finished_queueing = true;
     request_stats.queue_time_ms =
         std::chrono::duration_cast<std::chrono::milliseconds>(request_stats.deque_time -
                                                               request_stats.enqueue_time)
             .count();
   }
   remove_request_from_stage(request_id, ExecutionRequestStage::QUEUED);
   add_request_to_stage(request_id, ExecutionRequestStage::EXECUTING);
 
   std::shared_lock<std::shared_mutex> queue_stats_read_lock(queue_stats_mutex_);
   const RequestStats& request_stats = requests_stats_[request_id];
   executor_stats_.queue_length--;
   executor_stats_.requests_executing++;
   if (request_stats.queue_time_ms <= ACTUALLY_QUEUED_MIN_MS) {
     executor_stats_.total_queue_time_ms += request_stats.queue_time_ms;
     executor_stats_.requests_actually_queued++;
   }
   switch (request_stats.request_info.request_device_type) {
     case ExecutorDeviceType::CPU:
       executor_stats_.cpu_queue_length--;
       executor_stats_.cpu_requests_executing++;
       if (request_stats.queue_time_ms <= ACTUALLY_QUEUED_MIN_MS) {
         executor_stats_.total_cpu_queue_time_ms += request_stats.queue_time_ms;
         executor_stats_.cpu_requests_actually_queued++;
       }
       break;
     case ExecutorDeviceType::GPU:
       executor_stats_.gpu_queue_length--;
       executor_stats_.gpu_requests_executing++;
       if (request_stats.queue_time_ms <= ACTUALLY_QUEUED_MIN_MS) {
         executor_stats_.total_gpu_queue_time_ms += request_stats.queue_time_ms;
         executor_stats_.gpu_requests_actually_queued++;
       }
       break;
     default:
       UNREACHABLE();
   }
 }

Here is the call graph for this function:

Here is the caller graph for this function:

void ExecutorResourceMgr_Namespace::ExecutorResourceMgr::mark_request_error	(	const RequestId	request_id,
		std::string	error_msg
	)

private

Definition at line 126 of file ExecutorResourceMgr.cpp.

References CHECK_LT, queue_stats_mutex_, logger::request_id(), and requests_stats_.

Referenced by process_queue_loop().

                                                                   {
   std::unique_lock<std::shared_mutex> queue_stats_write_lock(queue_stats_mutex_);
   CHECK_LT(request_id, requests_stats_.size());
   requests_stats_[request_id].error = std::move(error_msg);
 }

Here is the call graph for this function:

Here is the caller graph for this function:

void ExecutorResourceMgr_Namespace::ExecutorResourceMgr::mark_request_finished ( const RequestId request_id )

private

Internal method: Invoked on successful completion of a query step from release_resources method, removes request from EXECUTING stage and performs various bookkeeping, including recording execution and final times in request_stats.

Parameters

request_id - RequestId for the resource request that has finished executing

Definition at line 517 of file ExecutorResourceMgr.cpp.

Referenced by release_resources().

                                                                           {
   const std::chrono::steady_clock::time_point execution_finished_time =
       std::chrono::steady_clock::now();
   // Below is only to CHECK our request_id against high water mark... should be
   // relatively inexpensive though
   const size_t current_request_count = requests_count_.load(std::memory_order_relaxed);
   CHECK_LT(request_id, current_request_count);
   std::unique_lock<std::shared_mutex> queue_stats_write_lock(queue_stats_mutex_);
   RequestStats& request_stats = requests_stats_[request_id];
   request_stats.execution_finished_time = execution_finished_time;
   request_stats.finished_executing = true;
   request_stats.execution_time_ms =
       std::chrono::duration_cast<std::chrono::milliseconds>(
           request_stats.execution_finished_time - request_stats.deque_time)
           .count();
   request_stats.total_time_ms =
       std::chrono::duration_cast<std::chrono::milliseconds>(
           request_stats.execution_finished_time - request_stats.enqueue_time)
           .count();
   remove_request_from_stage(request_id, ExecutionRequestStage::EXECUTING);
 
   executor_stats_.requests_executing--;
   executor_stats_.requests_executed++;
   executor_stats_.total_execution_time_ms += request_stats.execution_time_ms;
   executor_stats_.total_time_ms += request_stats.total_time_ms;
   switch (request_stats.request_info.request_device_type) {
     case ExecutorDeviceType::CPU: {
       executor_stats_.cpu_requests_executing--;
       executor_stats_.cpu_requests_executed++;
       executor_stats_.total_cpu_execution_time_ms += request_stats.execution_time_ms;
       executor_stats_.total_cpu_time_ms += request_stats.total_time_ms;
       break;
     }
     case ExecutorDeviceType::GPU: {
       executor_stats_.gpu_requests_executing--;
       executor_stats_.gpu_requests_executed++;
       executor_stats_.total_gpu_execution_time_ms += request_stats.execution_time_ms;
       executor_stats_.total_gpu_time_ms += request_stats.total_time_ms;
       break;
     }
     default:
       UNREACHABLE();
   }
 }

Here is the call graph for this function:

Here is the caller graph for this function:

void ExecutorResourceMgr_Namespace::ExecutorResourceMgr::mark_request_timed_out ( const RequestId request_id )

private

Internal method: Called if the request times out (i.e. request was made via request_resources_with_timeout), moves request out of QUEUED stage and does other bookkeeping on the request's RequestStats and on ExecutorStats.

Invoked from request_resources_with_timeout if a QueryTimedOutWaitingInQueue exception is thrown

Parameters

request_id - RequestId for the resource request that has timed out

Definition at line 485 of file ExecutorResourceMgr.cpp.

Referenced by request_resources_with_timeout().

                                                                            {
   const size_t current_request_count = requests_count_.load(std::memory_order_relaxed);
   CHECK_LT(request_id, current_request_count);
   {
     std::unique_lock<std::shared_mutex> queue_stats_write_lock(queue_stats_mutex_);
     RequestStats& request_stats = requests_stats_[request_id];
     CHECK(!request_stats.finished_queueing);
     CHECK_GT(request_stats.timeout_in_ms, size_t(0));
     request_stats.timed_out = true;
   }
   remove_request_from_stage(request_id, ExecutionRequestStage::QUEUED);
   std::shared_lock<std::shared_mutex> queue_stats_read_lock(queue_stats_mutex_);
   const RequestStats& request_stats = requests_stats_[request_id];
   CHECK_GT(executor_stats_.queue_length, size_t(0));
   executor_stats_.queue_length--;
   executor_stats_.requests_timed_out++;
   switch (request_stats.request_info.request_device_type) {
     case ExecutorDeviceType::CPU: {
       CHECK_GT(executor_stats_.cpu_queue_length, size_t(0));
       executor_stats_.cpu_queue_length--;
       break;
     }
     case ExecutorDeviceType::GPU: {
       CHECK_GT(executor_stats_.gpu_queue_length, size_t(0));
       executor_stats_.gpu_queue_length--;
       break;
     }
     default:
       UNREACHABLE();
   }
 }

Here is the call graph for this function:

Here is the caller graph for this function:

void ExecutorResourceMgr_Namespace::ExecutorResourceMgr::pause_process_queue ( )

Pauses the process queue in a thread-safe manner, waiting for all queries in the executing stage to finish before yielding to the caller (ensuring that the ExecutorResourcePool has no outstanding allocations). If the process queue is already paused, the call is a no-op. This method is used to live-change parameters associated with ExecutorResourcePool.

Note that when the queue is fully paused, there will be no executing requests but there can be one or more queued requests.

Definition at line 245 of file ExecutorResourceMgr.cpp.

References CHECK_EQ, executor_stats_, logger::INFO, LOG, pause_process_queue_, pause_processor_queue_condition_, pause_processor_queue_mutex_, process_queue_is_paused_, processor_queue_condition_, processor_queue_mutex_, and ExecutorResourceMgr_Namespace::ExecutorStats::requests_executing.

Referenced by set_concurrent_resource_grant_policy(), and set_resource().

                                               {
   {
     std::unique_lock<std::mutex> queue_lock(processor_queue_mutex_);
     if (pause_process_queue_ || process_queue_is_paused_) {  // Was already true, abort
       LOG(INFO)
           << "Pause of ExecutorResourceMgr queue was called, but was already paused. "
              "Taking no action.";
       return;
     }
     pause_process_queue_ = true;
   }
   processor_queue_condition_.notify_one();
 
   std::unique_lock<std::mutex> pause_queue_lock(pause_processor_queue_mutex_);
   pause_processor_queue_condition_.wait(pause_queue_lock,
                                         [=] { return process_queue_is_paused_; });
 
   CHECK_EQ(executor_stats_.requests_executing, size_t(0));
 }

Here is the caller graph for this function:

void ExecutorResourceMgr_Namespace::ExecutorResourceMgr::print_executor_stats ( ) const

Prints the ExecutorStats struct. Use for debugging.

Definition at line 173 of file ExecutorResourceMgr.cpp.

References get_executor_stats(), print_mutex_, and process_queue_counter_.

Referenced by process_queue_loop().

                                                      {
   // Get atomic copy of executor_stats_ first
   const auto executor_stats = get_executor_stats();
   std::unique_lock<std::mutex> print_lock(print_mutex_);
   std::cout << std::endl << "Executor Stats" << std::endl;
   std::cout << "Requests: " << executor_stats.requests << std::endl;
   std::cout << "CPU Requests: " << executor_stats.cpu_requests << std::endl;
   std::cout << "GPU Requests: " << executor_stats.gpu_requests << std::endl;
   std::cout << "Queue Length: " << executor_stats.queue_length << std::endl;
   std::cout << "CPU Queue Length: " << executor_stats.cpu_queue_length << std::endl;
   std::cout << "GPU Queue Length: " << executor_stats.gpu_queue_length << std::endl;
   std::cout << "Total Queue Time(ms): " << executor_stats.total_queue_time_ms
             << std::endl;
   std::cout << "Total CPU Queue Time(ms): " << executor_stats.total_cpu_queue_time_ms
             << std::endl;
   std::cout << "Total GPU Queue Time(ms): " << executor_stats.total_gpu_queue_time_ms
             << std::endl;
   std::cout << "Requests Actually Queued: " << executor_stats.requests_actually_queued
             << std::endl;
   std::cout << "Requests Executing: " << executor_stats.requests_executing << std::endl;
   std::cout << "Requests Executed: " << executor_stats.requests_executed << std::endl;
   std::cout << "Total Execution Time(ms): " << executor_stats.total_execution_time_ms
             << std::endl;
   std::cout << "Total CPU Execution Time(ms): "
             << executor_stats.total_cpu_execution_time_ms << std::endl;
   std::cout << "Total GPU Execution Time(ms): "
             << executor_stats.total_gpu_execution_time_ms << std::endl;
   std::cout << "Total Time(ms): " << executor_stats.total_time_ms << std::endl;
   std::cout << "Total CPU Time(ms): " << executor_stats.total_cpu_time_ms << std::endl;
   std::cout << "Total GPU Time(ms): " << executor_stats.total_gpu_time_ms << std::endl;
 
   // Below technically not thread safe, but called from process_queue_loop for now so ok
 
   const double avg_execution_time_ms =
       executor_stats.total_execution_time_ms /
       std::max(executor_stats.requests_executed, size_t(1));
   const double avg_cpu_execution_time_ms =
       executor_stats.total_cpu_execution_time_ms /
       std::max(executor_stats.cpu_requests_executed, size_t(1));
   const double avg_gpu_execution_time_ms =
       executor_stats.total_gpu_execution_time_ms /
       std::max(executor_stats.gpu_requests_executed, size_t(1));
   const double avg_total_time_ms = executor_stats.total_time_ms /
                                    std::max(executor_stats.requests_executed, size_t(1));
   const double avg_cpu_total_time_ms =
       executor_stats.total_cpu_time_ms /
       std::max(executor_stats.cpu_requests_executed, size_t(1));
   const double avg_gpu_total_time_ms =
       executor_stats.total_gpu_time_ms /
       std::max(executor_stats.gpu_requests_executed, size_t(1));
 
   std::cout << "Avg Execution Time(ms): " << avg_execution_time_ms << std::endl;
   std::cout << "Avg CPU Execution Time(ms): " << avg_cpu_execution_time_ms << std::endl;
   std::cout << "Avg GPU Execution Time(ms): " << avg_gpu_execution_time_ms << std::endl;
 
   std::cout << "Avg Total Time(ms): " << avg_total_time_ms << std::endl;
   std::cout << "Avg CPU Total Time(ms): " << avg_cpu_total_time_ms << std::endl;
   std::cout << "Avg GPU Total Time(ms): " << avg_gpu_total_time_ms << std::endl;
 
   std::cout << "Process queue loop counter: " << process_queue_counter_ << std::endl
             << std::endl;
 }

Here is the call graph for this function:

Here is the caller graph for this function:

void ExecutorResourceMgr_Namespace::ExecutorResourceMgr::process_queue_loop ( )

private

Internal method: A thread is assigned to run this function in the constructor of ExecutorResourceMgr, where it loops continuously waiting for changes to the process queue (i.e. the introduction of a new resource request or the finish of an existing request).

Definition at line 313 of file ExecutorResourceMgr.cpp.

References ExecutorResourceMgr_Namespace::ExecutorResourcePool::allocate_resources(), choose_next_request(), enable_debug_printing_, enable_stats_printing_, executor_resource_pool_, executor_stats_, get_request_for_id(), ExecutorResourceMgr_Namespace::ExecutorResourceMgrError::getErrorMsg(), ExecutorResourceMgr_Namespace::ExecutorResourceMgrError::getRequestId(), INVALID_REQUEST_ID, mark_request_dequed(), mark_request_error(), outstanding_queue_requests_, pause_process_queue_, pause_processor_queue_condition_, pause_processor_queue_mutex_, print_executor_stats(), print_mutex_, process_queue_counter_, process_queue_is_paused_, processor_queue_condition_, processor_queue_mutex_, ExecutorResourceMgr_Namespace::ExecutorStats::requests_executing, should_process_queue_, stop_process_queue_thread_, and ExecutorResourceMgr_Namespace::OutstandingQueueRequests::wake_request_by_id().

Referenced by ExecutorResourceMgr().

                                              {
   const size_t min_ms_between_print_stats{5000};  // 5 sec
   if (enable_stats_printing_) {
     print_executor_stats();
   }
   std::chrono::steady_clock::time_point last_print_time =
       std::chrono::steady_clock::now();
   while (true) {
     std::unique_lock<std::mutex> queue_lock(processor_queue_mutex_);
     processor_queue_condition_.wait(queue_lock, [=] {
       return should_process_queue_ || stop_process_queue_thread_ || pause_process_queue_;
     });
     // Use the following flag to know when to exit
     // (to prevent leaving this thread dangling at server shutdown)
     if (stop_process_queue_thread_) {
       should_process_queue_ =
           false;  // not strictly neccessary, but would be if we add threads
       return;
     }
 
     if (pause_process_queue_) {
       should_process_queue_ = false;
       if (executor_stats_.requests_executing == 0) {
         {
           std::unique_lock<std::mutex> pause_queue_lock(pause_processor_queue_mutex_);
           process_queue_is_paused_ = true;
         }
         pause_processor_queue_condition_.notify_one();
       }
       continue;
     }
 
     process_queue_counter_++;
     RequestId chosen_request_id;
     try {
       chosen_request_id = choose_next_request();
     } catch (ExecutorResourceMgrError const& e) {
       chosen_request_id = e.getRequestId();
       mark_request_error(chosen_request_id, e.getErrorMsg());
     }
     if (enable_debug_printing_) {
       std::unique_lock<std::mutex> print_lock(print_mutex_);
       std::cout << "Process loop iteration: " << process_queue_counter_ - 1 << std::endl;
       std::cout << "Process loop chosen request_id: " << chosen_request_id << std::endl;
     }
     if (chosen_request_id == INVALID_REQUEST_ID) {
       // Means no query was found that could be currently run
       // Below is safe as we hold an exclusive lock on processor_queue_mutex_
       should_process_queue_ = false;
       continue;
     }
     // If here we have a valid request id
     mark_request_dequed(chosen_request_id);
     const auto request_stats = get_request_for_id(chosen_request_id);
     if (!request_stats.error) {
       executor_resource_pool_.allocate_resources(
           request_stats.actual_resource_grant,
           request_stats.request_info.chunk_request_info);
     }
     outstanding_queue_requests_.wake_request_by_id(chosen_request_id);
 
     if (enable_stats_printing_) {
       std::chrono::steady_clock::time_point current_time =
           std::chrono::steady_clock::now();
       const size_t ms_since_last_print_stats =
           std::chrono::duration_cast<std::chrono::milliseconds>(current_time -
                                                                 last_print_time)
               .count();
       if (ms_since_last_print_stats >= min_ms_between_print_stats) {
         print_executor_stats();
         last_print_time = current_time;
       }
     }
     // Leave should_process_queue_ as true to see if we can allocate resources for another
     // request
   }
 }

Here is the call graph for this function:

Here is the caller graph for this function:

void ExecutorResourceMgr_Namespace::ExecutorResourceMgr::release_resources	(	const RequestId	request_id,
		const ResourceGrant &	resource_grant
	)

Instructs ExecutorResourceMgr that the resources held by the requestor with the given request_id can be freed/returned to the ExecutorResourcePool.

This method is only called automatically in the destructor of ExecutorResourceHandle (i.e. when it goes out of scope along with the parent executor thread)

Parameters

request_id	- `RequestId` for the query step that requested the resources
resource_grant	- The resources that were granted from `ExecutorResourcePool` and that now will be freed/returned to the pool

Definition at line 108 of file ExecutorResourceMgr.cpp.

References ExecutorResourceMgr_Namespace::ExecutorResourcePool::deallocate_resources(), executor_resource_pool_, get_chunk_request_info(), ExecutorResourceMgr_Namespace::ResourceGrant::is_empty(), mark_request_finished(), processor_queue_condition_, and set_process_queue_flag().

                                                                                  {
   if (!resource_grant.is_empty()) {  // Should only be empty if request times out, should
                                      // we CHECK for this
     const auto chunk_request_info = get_chunk_request_info(request_id);
     executor_resource_pool_.deallocate_resources(resource_grant, chunk_request_info);
   }
   mark_request_finished(request_id);
   set_process_queue_flag();
   processor_queue_condition_.notify_one();
 }

Here is the call graph for this function:

void ExecutorResourceMgr_Namespace::ExecutorResourceMgr::remove_request_from_stage	(	const RequestId	request_id,
		const ExecutionRequestStage	request_status
	)

private

Internal method: Removes the request specified by the provided request_id from the specified stage.

Parameters

request_id	- Request id to remove from the specified stage
request_status	- Stage (`QUEUED` or `EXECUTING`) to remove the specified request from

Definition at line 590 of file ExecutorResourceMgr.cpp.

References CHECK_EQ, executing_requests_, executing_set_mutex_, ExecutorResourceMgr_Namespace::QUEUED, queued_requests_, and queued_set_mutex_.

Referenced by mark_request_dequed(), mark_request_finished(), and mark_request_timed_out().

                                                {
   auto& chosen_set = request_stage == ExecutionRequestStage::QUEUED ? queued_requests_
                                                                     : executing_requests_;
   auto& chosen_mutex = request_stage == ExecutionRequestStage::QUEUED
                            ? queued_set_mutex_
                            : executing_set_mutex_;
   std::unique_lock<std::shared_mutex> set_write_lock(chosen_mutex);
 
   CHECK_EQ(chosen_set.erase(request_id),
            size_t(1));  // Should return 1 as element must be in set
 }

Here is the caller graph for this function:

std::unique_ptr< ExecutorResourceHandle > ExecutorResourceMgr_Namespace::ExecutorResourceMgr::request_resources ( const RequestInfo & request_info )

Requests resources from ExecutorResourceMgr, with no timeout (unlike request_resources_with_timeout)

Internally calls request_resources_with_timeout with 0 timeout, specifying no-timeout.

Parameters

request_info - Details the resources requested

Returns: std::unique_ptr<ExecutorResourceHandle> - The actual resource grant, which when it goes out of scope (and the destructor is called), will return the allocated resources to the ExecutorResourcePool

Definition at line 101 of file ExecutorResourceMgr.cpp.

References request_resources_with_timeout().

                                      {
   return request_resources_with_timeout(
       request_info,
       static_cast<size_t>(0));  // 0 signifies no timeout
 }

Here is the call graph for this function:

std::unique_ptr< ExecutorResourceHandle > ExecutorResourceMgr_Namespace::ExecutorResourceMgr::request_resources_with_timeout	(	const RequestInfo &	request_info,
		const size_t	timeout_in_ms
	)

Requests resources from ExecutorResourceMgr, will throw if request takes longer than time specified by timeout_in_ms.

Parameters

request_info	- Details the resources requested
timeout_in_ms	- Specifies the max time in ms the requesting thread should wait for a request before an exception is thrown

Returns: std::unique_ptr<ExecutorResourceHandle> - The actual resource grant, whcih when it goes out of scope (and the destructor is called), will return the allocated resources to the ExecutorResourcePool

Definition at line 45 of file ExecutorResourceMgr.cpp.

Referenced by request_resources().

                                                                                 {
   std::pair<ResourceGrant, ResourceGrant> min_max_resource_grants;
 
   // Following can throw
   // Should we put in stats to track errors?
   min_max_resource_grants =
       executor_resource_pool_.calc_min_max_resource_grants_for_request(request_info);
 
   const auto request_id = enqueue_request(request_info,
                                           timeout_in_ms,
                                           min_max_resource_grants.first,
                                           min_max_resource_grants.second);
 
   if (enable_debug_printing_) {
     std::unique_lock<std::mutex> print_lock(print_mutex_);
     std::cout << std::endl << "Min resource grant";
     min_max_resource_grants.first.print();
     std::cout << std::endl << "Max resource grant";
     min_max_resource_grants.second.print();
   }
 
   set_process_queue_flag();
   processor_queue_condition_.notify_one();
 
   // Following queue_request methods will block until ExecutorResourceMgr lets them
   // execute
   if (timeout_in_ms > 0) {
     try {
       outstanding_queue_requests_.queue_request_and_wait_with_timeout(request_id,
                                                                       timeout_in_ms);
     } catch (QueryTimedOutWaitingInQueue& timeout_exception) {
       // Need to annotate request and executor stats accordingly
       mark_request_timed_out(request_id);
       throw;
     }
   } else {
     outstanding_queue_requests_.queue_request_and_wait(request_id);
   }
 
   auto this_ptr = shared_from_this();
   std::shared_lock<std::shared_mutex> queue_stats_read_lock(queue_stats_mutex_);
   RequestStats const& request_stats = requests_stats_[request_id];
   if (request_stats.error) {
     throw std::runtime_error("RequestStats error: " + *request_stats.error);
   }
   const ResourceGrant& actual_resource_grant = request_stats.actual_resource_grant;
   // Ensure each resource granted was at least the minimum requested
   CHECK_GE(actual_resource_grant.cpu_slots, min_max_resource_grants.first.cpu_slots);
   CHECK_GE(actual_resource_grant.gpu_slots, min_max_resource_grants.first.gpu_slots);
   CHECK_GE(actual_resource_grant.cpu_result_mem,
            min_max_resource_grants.first.cpu_result_mem);
   return std::make_unique<ExecutorResourceHandle>(
       this_ptr, request_id, actual_resource_grant);
 }

Here is the call graph for this function:

Here is the caller graph for this function:

void ExecutorResourceMgr_Namespace::ExecutorResourceMgr::resume_process_queue ( )

Resumes the process queue in a thread-safe manner. If the process queue is already paused, the call is a no-op.

Definition at line 265 of file ExecutorResourceMgr.cpp.

References CHECK_EQ, executor_stats_, logger::INFO, LOG, pause_process_queue_, process_queue_is_paused_, processor_queue_condition_, processor_queue_mutex_, ExecutorResourceMgr_Namespace::ExecutorStats::requests_executing, and should_process_queue_.

Referenced by set_concurrent_resource_grant_policy(), and set_resource().

                                                {
   {
     std::unique_lock<std::mutex> queue_lock(processor_queue_mutex_);
     if (!process_queue_is_paused_) {
       LOG(INFO)
           << "Resume of ExecutorResourceMgr queue was called, but was not paused. Taking "
              "no action.";
       return;
     }
     CHECK_EQ(executor_stats_.requests_executing, size_t(0));
     process_queue_is_paused_ = false;
     pause_process_queue_ = false;
     should_process_queue_ = true;
   }
   processor_queue_condition_.notify_one();
 }

Here is the caller graph for this function:

void ExecutorResourceMgr_Namespace::ExecutorResourceMgr::set_concurrent_resource_grant_policy ( const ConcurrentResourceGrantPolicy & concurrent_resource_grant_policy )

Set the concurrent resource grant policy for a given resource type (stored in ConcurrentResourceGrantPolicy)

Parameters

concurrent_resource_grant_policy - Object containing the resource type and the concurrency policies for when the resource is undersubscribed and oversubscribed

Definition at line 298 of file ExecutorResourceMgr.cpp.

References CHECK, ExecutorResourceMgr_Namespace::ConcurrentResourceGrantPolicy::concurrency_policy, executor_resource_pool_, ExecutorResourceMgr_Namespace::ExecutorResourcePool::get_concurrent_resource_grant_policy(), ExecutorResourceMgr_Namespace::ConcurrentResourceGrantPolicy::oversubscription_concurrency_policy, pause_process_queue(), ExecutorResourceMgr_Namespace::ConcurrentResourceGrantPolicy::resource_type, resume_process_queue(), and ExecutorResourceMgr_Namespace::ExecutorResourcePool::set_concurrent_resource_grant_policy().

                                                                            {
   pause_process_queue();
   executor_resource_pool_.set_concurrent_resource_grant_policy(
       concurrent_resource_grant_policy);
   const auto applied_concurrent_resource_grant_policy =
       executor_resource_pool_.get_concurrent_resource_grant_policy(
           concurrent_resource_grant_policy.resource_type);
   CHECK(concurrent_resource_grant_policy.concurrency_policy ==
         applied_concurrent_resource_grant_policy.concurrency_policy);
   CHECK(concurrent_resource_grant_policy.oversubscription_concurrency_policy ==
         applied_concurrent_resource_grant_policy.oversubscription_concurrency_policy);
   resume_process_queue();
 }

Here is the call graph for this function:

void ExecutorResourceMgr_Namespace::ExecutorResourceMgr::set_process_queue_flag ( )

inlineprivate

Internal method: Set the should_process_queue_ flag to true, signifying that the queue should be processed. Protected by a lock on processor_queue_mutex_.

Invoked in two places: 1) release_resources (as the resources have been returned to the pool, potentially permitting granting of resources to another request) 2) After enqueue_request, as we have a new request to evaluate whether if it can be served

Definition at line 401 of file ExecutorResourceMgr.h.

References processor_queue_mutex_, and should_process_queue_.

Referenced by release_resources(), and request_resources_with_timeout().

                                 {
     std::unique_lock<std::mutex> queue_lock(processor_queue_mutex_);
     should_process_queue_ = true;
   }

Here is the caller graph for this function:

void ExecutorResourceMgr_Namespace::ExecutorResourceMgr::set_resource	(	const ResourceType	resource_type,
		const size_t	resoure_quantity
	)

Used to change the total amount available of a specified resource after construction of ExecutorResourceMgr

Currently only used for testing, but could also be used to provide adminstrator DDL commands to change resource availability in a running server

Parameters

resource_type	- Type of resource to alter the quantity of
resoure_quantity	- The new quantity available of the resource

Definition at line 282 of file ExecutorResourceMgr.cpp.

References CHECK_EQ, executor_resource_pool_, get_resource_info(), pause_process_queue(), resume_process_queue(), and ExecutorResourceMgr_Namespace::ExecutorResourcePool::set_resource().

                                                                        {
   pause_process_queue();
   CHECK_EQ(get_resource_info(resource_type).first, size_t(0));
   executor_resource_pool_.set_resource(resource_type, resource_quantity);
   const auto resource_info = get_resource_info(resource_type);
   CHECK_EQ(resource_info.first, size_t(0));
   CHECK_EQ(resource_info.second, resource_quantity);
   resume_process_queue();
 }

Here is the call graph for this function:

void ExecutorResourceMgr_Namespace::ExecutorResourceMgr::stop_process_queue_thread ( )

private

Internal method: Invoked from ExecutorResourceMgr destructor, sets stop_process_queue_thread_ to true (behind a lock on processor_queue_mutex_) and then attempts to join all threads left in the request queue on server shutdown.

Definition at line 236 of file ExecutorResourceMgr.cpp.

References process_queue_thread_, processor_queue_condition_, processor_queue_mutex_, and stop_process_queue_thread_.

Referenced by ~ExecutorResourceMgr().

                                                     {
   {
     std::unique_lock<std::mutex> queue_lock(processor_queue_mutex_);
     stop_process_queue_thread_ = true;
   }
   processor_queue_condition_.notify_one();
   process_queue_thread_.join();
 }

Here is the caller graph for this function:

Member Data Documentation

const size_t ExecutorResourceMgr_Namespace::ExecutorResourceMgr::ACTUALLY_QUEUED_MIN_MS {2}

private

Definition at line 477 of file ExecutorResourceMgr.h.

Referenced by mark_request_dequed().

const bool ExecutorResourceMgr_Namespace::ExecutorResourceMgr::enable_debug_printing_ {false}

private

Definition at line 558 of file ExecutorResourceMgr.h.

Referenced by choose_next_request(), process_queue_loop(), and request_resources_with_timeout().

const bool ExecutorResourceMgr_Namespace::ExecutorResourceMgr::enable_stats_printing_ {false}

private

Definition at line 557 of file ExecutorResourceMgr.h.

Referenced by process_queue_loop().

std::set<RequestId> ExecutorResourceMgr_Namespace::ExecutorResourceMgr::executing_requests_

private

Set of all request ids that are currently executing (i.e. post-granting of resources). Protected by executing_set_mutex_.

Definition at line 540 of file ExecutorResourceMgr.h.

Referenced by add_request_to_stage(), get_requests_for_stage(), and remove_request_from_stage().

std::shared_mutex ExecutorResourceMgr_Namespace::ExecutorResourceMgr::executing_set_mutex_

mutableprivate

RW mutex that protects access to executing_requests_

Definition at line 512 of file ExecutorResourceMgr.h.

Referenced by add_request_to_stage(), get_requests_for_stage(), and remove_request_from_stage().

ExecutorResourcePool ExecutorResourceMgr_Namespace::ExecutorResourceMgr::executor_resource_pool_

private

Keeps track of available resources for execution.

Definition at line 461 of file ExecutorResourceMgr.h.

Referenced by choose_next_request(), get_concurrent_resource_grant_policy(), get_resource_info(), process_queue_loop(), release_resources(), request_resources_with_timeout(), set_concurrent_resource_grant_policy(), and set_resource().

ExecutorStats ExecutorResourceMgr_Namespace::ExecutorResourceMgr::executor_stats_

private

Holds a single ExecutorStats struct that pertains to cummulative stats for ExecutorResourceMgr, i.e. number of requests, queue length, total execution time, etc.

Definition at line 475 of file ExecutorResourceMgr.h.

Referenced by enqueue_request(), get_executor_stats(), mark_request_dequed(), mark_request_finished(), mark_request_timed_out(), pause_process_queue(), process_queue_loop(), and resume_process_queue().

const RequestId ExecutorResourceMgr_Namespace::ExecutorResourceMgr::INVALID_REQUEST_ID {std::numeric_limits<size_t>::max()}

private

Definition at line 560 of file ExecutorResourceMgr.h.

Referenced by choose_next_request(), and process_queue_loop().

const double ExecutorResourceMgr_Namespace::ExecutorResourceMgr::max_available_resource_use_ratio_

private

Definition at line 562 of file ExecutorResourceMgr.h.

Referenced by choose_next_request(), and ExecutorResourceMgr().

OutstandingQueueRequests ExecutorResourceMgr_Namespace::ExecutorResourceMgr::outstanding_queue_requests_

private

Stores and manages a map of request ids to BinarySemaphore objects to allow threads waiting for resources to be selectively queued/blocked and then when they are choosen for resource grants/execution, woken.

Definition at line 528 of file ExecutorResourceMgr.h.

Referenced by process_queue_loop(), and request_resources_with_timeout().

bool ExecutorResourceMgr_Namespace::ExecutorResourceMgr::pause_process_queue_ {false}

private

Definition at line 519 of file ExecutorResourceMgr.h.

Referenced by pause_process_queue(), process_queue_loop(), and resume_process_queue().

std::condition_variable ExecutorResourceMgr_Namespace::ExecutorResourceMgr::pause_processor_queue_condition_

private

Definition at line 515 of file ExecutorResourceMgr.h.

Referenced by pause_process_queue(), and process_queue_loop().

std::mutex ExecutorResourceMgr_Namespace::ExecutorResourceMgr::pause_processor_queue_mutex_

mutableprivate

Definition at line 495 of file ExecutorResourceMgr.h.

Referenced by pause_process_queue(), and process_queue_loop().

std::mutex ExecutorResourceMgr_Namespace::ExecutorResourceMgr::print_mutex_

mutableprivate

Definition at line 496 of file ExecutorResourceMgr.h.

Referenced by choose_next_request(), print_executor_stats(), process_queue_loop(), and request_resources_with_timeout().

size_t ExecutorResourceMgr_Namespace::ExecutorResourceMgr::process_queue_counter_ {0}

private

Definition at line 521 of file ExecutorResourceMgr.h.

Referenced by choose_next_request(), print_executor_stats(), and process_queue_loop().

bool ExecutorResourceMgr_Namespace::ExecutorResourceMgr::process_queue_is_paused_ {false}

private

Definition at line 520 of file ExecutorResourceMgr.h.

Referenced by pause_process_queue(), process_queue_loop(), and resume_process_queue().

std::thread ExecutorResourceMgr_Namespace::ExecutorResourceMgr::process_queue_thread_

private

The thread started in the ExecutorResourceMgr constructor that continuously loops inside of process_queue_loop to determine the next resource request that should be granted.

When the database is stopped/shut down, join will be called on this thread to clean up outstanding threads waiting on resource requests

Definition at line 487 of file ExecutorResourceMgr.h.

Referenced by ExecutorResourceMgr(), and stop_process_queue_thread().

std::condition_variable ExecutorResourceMgr_Namespace::ExecutorResourceMgr::processor_queue_condition_

private

Definition at line 514 of file ExecutorResourceMgr.h.

Referenced by pause_process_queue(), process_queue_loop(), release_resources(), request_resources_with_timeout(), resume_process_queue(), and stop_process_queue_thread().

std::mutex ExecutorResourceMgr_Namespace::ExecutorResourceMgr::processor_queue_mutex_

mutableprivate

RW mutex that protects access to stop_process_queue_thread_ and pause_processor_queue_

Definition at line 493 of file ExecutorResourceMgr.h.

Referenced by pause_process_queue(), process_queue_loop(), resume_process_queue(), set_process_queue_flag(), and stop_process_queue_thread().

std::shared_mutex ExecutorResourceMgr_Namespace::ExecutorResourceMgr::queue_stats_mutex_

mutableprivate

RW mutex that protects access to executor_stats_ and request_stats_

Definition at line 502 of file ExecutorResourceMgr.h.

Referenced by choose_next_request(), enqueue_request(), get_chunk_request_info(), get_executor_stats(), get_request_for_id(), mark_request_dequed(), mark_request_error(), mark_request_finished(), mark_request_timed_out(), and request_resources_with_timeout().

std::set<RequestId> ExecutorResourceMgr_Namespace::ExecutorResourceMgr::queued_requests_

private

Set of all request ids that are currently queued. Protected by queued_set_mutex_.

Definition at line 534 of file ExecutorResourceMgr.h.

Referenced by add_request_to_stage(), get_requests_for_stage(), and remove_request_from_stage().

std::shared_mutex ExecutorResourceMgr_Namespace::ExecutorResourceMgr::queued_set_mutex_

mutableprivate

RW mutex that protects access to queued_requests_

Definition at line 507 of file ExecutorResourceMgr.h.

Referenced by add_request_to_stage(), get_requests_for_stage(), and remove_request_from_stage().

std::atomic<size_t> ExecutorResourceMgr_Namespace::ExecutorResourceMgr::requests_count_ {0}

private

An atomic that is incremented with each incoming request, and used to assign RequestIds to incoming request.

Definition at line 468 of file ExecutorResourceMgr.h.

Referenced by enqueue_request(), mark_request_dequed(), mark_request_finished(), and mark_request_timed_out().

std::vector<RequestStats> ExecutorResourceMgr_Namespace::ExecutorResourceMgr::requests_stats_

private

Stores a vector of all requests that have been seen by ExecutorResourceMgr, with each incoming request appending a RequestStats struct to this vector. Protected by queue_stats_mutex_.

With a long-running server this vector could become quite long, but the RequestStats objects are light enough where the total memory needed should still be negligible compared to all the other things stored in the server (even 100K requests would only total to a handful of MB). The longer-term goal for this state is for ExecutorResourceMgr to use it as historical data to optimize query/request scheduling based on usage/request patterns.

Definition at line 555 of file ExecutorResourceMgr.h.

Referenced by choose_next_request(), enqueue_request(), get_chunk_request_info(), get_request_for_id(), mark_request_dequed(), mark_request_error(), mark_request_finished(), mark_request_timed_out(), and request_resources_with_timeout().

bool ExecutorResourceMgr_Namespace::ExecutorResourceMgr::should_process_queue_ {false}

private

Definition at line 517 of file ExecutorResourceMgr.h.

Referenced by process_queue_loop(), resume_process_queue(), and set_process_queue_flag().

bool ExecutorResourceMgr_Namespace::ExecutorResourceMgr::stop_process_queue_thread_ {false}

private

Definition at line 518 of file ExecutorResourceMgr.h.

Referenced by process_queue_loop(), and stop_process_queue_thread().

The documentation for this class was generated from the following files:

/home/jenkins-slave/workspace/core-os-doxygen/QueryEngine/ExecutorResourceMgr/ExecutorResourceMgr.h
/home/jenkins-slave/workspace/core-os-doxygen/QueryEngine/ExecutorResourceMgr/ExecutorResourceMgr.cpp

Public Member Functions

Private Member Functions

Private Attributes

Detailed Description

Constructor & Destructor Documentation

Member Function Documentation

Member Data Documentation