Stores and allows access to a binary semaphore per RequestId (using an std::unordered_map), as well as accessing all outstanding RequestIds for waiting requests. More...

#include <OutstandingQueueRequests.h>

Public Member Functions
void	queue_request_and_wait (const RequestId request_id)
	Submits a request with id `request_id` into the queue, waiting on a `BinarySemaphore` until `ExecutorResourceMgr` decides to grant the request resources and wakes the waiting thread. More...

void	queue_request_and_wait_with_timeout (const RequestId request_id, const size_t max_wait_in_ms)
	Submits a request with id `request_id` into the queue, waiting on a `BinarySemaphore` until `ExecutorResourceMgr` decides to grant the request resources and wakes the waiting thread. If it waits for a period longer than `max_wait_in_ms`, a `QueryTimedOutWaitingInQueue` exception is thrown. More...

std::vector< RequestId >	get_outstanding_request_ids ()
	Get the `RequestId`s of all outsanding requests in the queue. More...

void	wake_request_by_id (const RequestId request_id)
	Wakes a waiting thread in the queue. Invoked by `ExecutorResourceMgr::process_queue_loop()` More...

Private Types
using	OutstandingQueueRequestsMap = std::unordered_map< RequestId, SemaphoreShim_Namespace::BinarySemaphore >

Private Member Functions
SemaphoreShim_Namespace::BinarySemaphore &	get_semaphore_for_request (const RequestId request_id)
	Creates a new entry in `outstanding_requests_map_`, assigning a `BinarySemaphore` for the given requesting thread with id `request_id` More...

void	delete_semaphore_for_request (const RequestId request_id)
	Internal method: removes a `RequestId`-`BinarySemaphore` entry from `outstanding_requests_map_`. Invoked after a request thread is awoken (including on timeout). More...

Private Attributes
OutstandingQueueRequestsMap	outstanding_requests_map_
	Stores a map of `RequestId` to `BinarySemaphore` More...

std::shared_mutex	requests_map_mutex_
	Read-write lock protecting the `outstanding_requests_map_` More...

Detailed Description

Stores and allows access to a binary semaphore per RequestId (using an std::unordered_map), as well as accessing all outstanding RequestIds for waiting requests.

OutstandingQueueRequests

Definition at line 35 of file OutstandingQueueRequests.h.

Member Typedef Documentation

using ExecutorResourceMgr_Namespace::OutstandingQueueRequests::OutstandingQueueRequestsMap = std::unordered_map<RequestId, SemaphoreShim_Namespace::BinarySemaphore>

private

Definition at line 37 of file OutstandingQueueRequests.h.

Member Function Documentation

void ExecutorResourceMgr_Namespace::OutstandingQueueRequests::delete_semaphore_for_request ( const RequestId request_id )

inlineprivate

Internal method: removes a RequestId-BinarySemaphore entry from outstanding_requests_map_. Invoked after a request thread is awoken (including on timeout).

Parameters

request_id - RequestId key to be removed from the outstanding_requests_map_

Definition at line 130 of file OutstandingQueueRequests.h.

References CHECK_EQ, outstanding_requests_map_, and requests_map_mutex_.

Referenced by queue_request_and_wait(), and queue_request_and_wait_with_timeout().

                                                                 {
     std::unique_lock<std::shared_mutex> requests_write_lock(requests_map_mutex_);
     CHECK_EQ(outstanding_requests_map_.erase(request_id),
              size_t(1));  // Ensure the erase call returns 1, meaning there was actually
                           // an entry in the map matching this request_id to delete
   }

Here is the caller graph for this function:

std::vector<RequestId> ExecutorResourceMgr_Namespace::OutstandingQueueRequests::get_outstanding_request_ids ( )

inline

Get the RequestIds of all outsanding requests in the queue.

Returns: std::vector<RequestId> - The vector of request ids for outstanding requests in the queue

Definition at line 79 of file OutstandingQueueRequests.h.

References outstanding_requests_map_, and requests_map_mutex_.

                                                      {
     std::vector<RequestId> outstanding_request_ids;
     std::shared_lock<std::shared_mutex> requests_read_lock(requests_map_mutex_);
     outstanding_request_ids.reserve(outstanding_requests_map_.size());
     for (const auto& request_entry : outstanding_requests_map_) {
       outstanding_request_ids.emplace_back(request_entry.first);
     }
     return outstanding_request_ids;
   }

SemaphoreShim_Namespace::BinarySemaphore& ExecutorResourceMgr_Namespace::OutstandingQueueRequests::get_semaphore_for_request ( const RequestId request_id )

inlineprivate

Creates a new entry in outstanding_requests_map_, assigning a BinarySemaphore for the given requesting thread with id request_id

Parameters

request_id - RequestId for the requesting thread - will be used as a key in outstanding_requests_map_

Returns: SemaphoreShim_Namespace::BinarySemaphore& - a reference to the BinarySemaphore that was added to outstanding_requests_map_ for this request

Definition at line 116 of file OutstandingQueueRequests.h.

References outstanding_requests_map_, logger::request_id(), and requests_map_mutex_.

Referenced by queue_request_and_wait(), and queue_request_and_wait_with_timeout().

                                   {
     std::unique_lock<std::shared_mutex> requests_write_lock(requests_map_mutex_);
     return outstanding_requests_map_[request_id];
   }

Here is the call graph for this function:

Here is the caller graph for this function:

void ExecutorResourceMgr_Namespace::OutstandingQueueRequests::queue_request_and_wait ( const RequestId request_id )

inline

Submits a request with id request_id into the queue, waiting on a BinarySemaphore until ExecutorResourceMgr decides to grant the request resources and wakes the waiting thread.

Parameters

request_id - RequestId for this request

Definition at line 47 of file OutstandingQueueRequests.h.

References delete_semaphore_for_request(), and get_semaphore_for_request().

Referenced by ExecutorResourceMgr_Namespace::ExecutorResourceMgr::request_resources_with_timeout().

                                                           {
     auto& wait_semaphore = get_semaphore_for_request(request_id);
     wait_semaphore.try_acquire();
     delete_semaphore_for_request(request_id);
   }

Here is the call graph for this function:

Here is the caller graph for this function:

void ExecutorResourceMgr_Namespace::OutstandingQueueRequests::queue_request_and_wait_with_timeout	(	const RequestId	request_id,
		const size_t	max_wait_in_ms
	)

inline

Submits a request with id request_id into the queue, waiting on a BinarySemaphore until ExecutorResourceMgr decides to grant the request resources and wakes the waiting thread. If it waits for a period longer than max_wait_in_ms, a QueryTimedOutWaitingInQueue exception is thrown.

Parameters

request_id - RequestId for this request

Definition at line 61 of file OutstandingQueueRequests.h.

References CHECK_GT, delete_semaphore_for_request(), and get_semaphore_for_request().

Referenced by ExecutorResourceMgr_Namespace::ExecutorResourceMgr::request_resources_with_timeout().

                                                                         {
     CHECK_GT(max_wait_in_ms, size_t(0));
     auto& wait_semaphore = get_semaphore_for_request(request_id);
     // Binary semaphore returns false if it was not unblocked before the specified timeout
     const bool did_timeout = !(wait_semaphore.try_acquire_for(max_wait_in_ms));
     delete_semaphore_for_request(request_id);
     if (did_timeout) {
       throw QueryTimedOutWaitingInQueue(max_wait_in_ms);
     }
   }

Here is the call graph for this function:

Here is the caller graph for this function:

void ExecutorResourceMgr_Namespace::OutstandingQueueRequests::wake_request_by_id ( const RequestId request_id )

inline

Wakes a waiting thread in the queue. Invoked by ExecutorResourceMgr::process_queue_loop()

Parameters

request_id - RequestId of the request/thread that should be awoken

Definition at line 95 of file OutstandingQueueRequests.h.

References outstanding_requests_map_, logger::request_id(), and requests_map_mutex_.

Referenced by ExecutorResourceMgr_Namespace::ExecutorResourceMgr::process_queue_loop().

                                                       {
     std::unique_lock<std::shared_mutex> requests_write_lock(requests_map_mutex_);
     const auto request_itr = outstanding_requests_map_.find(request_id);
     if (request_itr == outstanding_requests_map_.end()) {
       outstanding_requests_map_[request_id].set_ready();
     } else {
       request_itr->second.release();
     }
   }