QueryDispatchQueue Class Reference

#include <QueryDispatchQueue.h>

Public Types
using	Task = std::packaged_task< void(size_t)>

Public Member Functions
	QueryDispatchQueue (const size_t parallel_executors_max)
void	submit (std::shared_ptr< Task > task, const bool is_update_delete)
bool	hasIdleWorker ()
	~QueryDispatchQueue ()

Private Member Functions
void	worker (const size_t worker_idx)

Private Attributes
std::mutex	queue_mutex_
std::condition_variable	cv_
std::mutex	update_delete_mutex_
bool	threads_should_exit_ {false}
std::queue< std::shared_ptr < Task > >	queue_
std::vector< std::thread >	workers_
int	num_running_workers_
int	num_workers_

Detailed Description

QueryDispatchQueue maintains a list of pending queries and dispatches those queries as Executors become available

Definition at line 29 of file QueryDispatchQueue.h.

Member Typedef Documentation

using QueryDispatchQueue::Task = std::packaged_task<void(size_t)>

Definition at line 31 of file QueryDispatchQueue.h.

Constructor & Destructor Documentation

QueryDispatchQueue::QueryDispatchQueue ( const size_t  parallel_executors_max )

inline

Definition at line 33 of file QueryDispatchQueue.h.

References num_running_workers_, num_workers_, worker(), and workers_.

                                                          {
    workers_.resize(parallel_executors_max);
    for (size_t i = 0; i < workers_.size(); i++) {
      // worker IDs are 1-indexed, leaving Executor 0 for non-dispatch queue worker tasks
      workers_[i] = std::thread(&QueryDispatchQueue::worker, this, i + 1);
    }
    num_running_workers_ = 0;
    num_workers_ = parallel_executors_max;
  }

Here is the call graph for this function:

QueryDispatchQueue::~QueryDispatchQueue ( )

inline

Definition at line 74 of file QueryDispatchQueue.h.

References cv_, queue_mutex_, threads_should_exit_, worker(), and workers_.

                        {
    {
      std::lock_guard<decltype(queue_mutex_)> lock(queue_mutex_);
      threads_should_exit_ = true;
    }
    cv_.notify_all();
    for (auto& worker : workers_) {
      worker.join();
    }
  }

Here is the call graph for this function:

Member Function Documentation

bool QueryDispatchQueue::hasIdleWorker ( )

inline

Definition at line 69 of file QueryDispatchQueue.h.

References num_running_workers_, num_workers_, and queue_mutex_.

                       {
    std::lock_guard<decltype(queue_mutex_)> lock(queue_mutex_);
    return num_running_workers_ < num_workers_;
  }

void QueryDispatchQueue::submit ( std::shared_ptr< Task >  task, const bool  is_update_delete  )

inline

Submit a new task to the queue. Blocks until the task begins execution. The caller is expected to maintain a copy of the shared_ptr which will be used to access results once the task runs.

Definition at line 48 of file QueryDispatchQueue.h.

References CHECK, cv_, logger::INFO, LOG, queue_, queue_mutex_, update_delete_mutex_, and workers_.

                                                                     {
    if (workers_.size() == 1 && is_update_delete) {
      std::lock_guard<decltype(update_delete_mutex_)> update_delete_lock(
          update_delete_mutex_);
      CHECK(task);
      // We only have 1 worker. Run this task on the calling thread on a special, second
      // worker. The task is under the update delete lock, so we don't have to worry about
      // contention on the special worker. This protects against deadlocks should the
      // query running (or any pending queries) hold a read lock on something that
      // requires a write lock during update/delete.
      (*task)(2);
      return;
    }
    std::unique_lock<decltype(queue_mutex_)> lock(queue_mutex_);

    LOG(INFO) << "Dispatching query with " << queue_.size() << " queries in the queue.";
    queue_.push(task);
    lock.unlock();
    cv_.notify_all();
  }

void QueryDispatchQueue::worker ( const size_t  worker_idx )

inlineprivate

Definition at line 86 of file QueryDispatchQueue.h.

References CHECK, cv_, logger::INFO, LOG, num_running_workers_, queue_, queue_mutex_, and threads_should_exit_.

Referenced by QueryDispatchQueue(), and ~QueryDispatchQueue().

                                       {
    std::unique_lock<std::mutex> lock(queue_mutex_);
    while (true) {
      cv_.wait(lock, [this] { return !queue_.empty() || threads_should_exit_; });

      if (threads_should_exit_) {
        return;
      }

      if (!queue_.empty()) {
        auto task = queue_.front();
        queue_.pop();
        ++num_running_workers_;

        LOG(INFO) << "Worker " << worker_idx
                  << " running query and returning control. There are now "
                  << num_running_workers_ << " workers are running and " << queue_.size()
                  << " queries in the queue.";
        // allow other threads to pick up tasks
        lock.unlock();
        CHECK(task);
        (*task)(worker_idx);
        // wait for signal
        lock.lock();
        --num_running_workers_;
      }
    }
  }

Here is the caller graph for this function:

Member Data Documentation

std::condition_variable QueryDispatchQueue::cv_

private

Definition at line 116 of file QueryDispatchQueue.h.

Referenced by submit(), worker(), and ~QueryDispatchQueue().

int QueryDispatchQueue::num_running_workers_

private

Definition at line 123 of file QueryDispatchQueue.h.

Referenced by hasIdleWorker(), QueryDispatchQueue(), and worker().

int QueryDispatchQueue::num_workers_

private

Definition at line 124 of file QueryDispatchQueue.h.

Referenced by hasIdleWorker(), and QueryDispatchQueue().

std::queue<std::shared_ptr<Task> > QueryDispatchQueue::queue_

private

Definition at line 121 of file QueryDispatchQueue.h.

Referenced by submit(), and worker().

std::mutex QueryDispatchQueue::queue_mutex_

private

Definition at line 115 of file QueryDispatchQueue.h.

Referenced by hasIdleWorker(), submit(), worker(), and ~QueryDispatchQueue().

bool QueryDispatchQueue::threads_should_exit_ {false}

private

Definition at line 120 of file QueryDispatchQueue.h.

Referenced by worker(), and ~QueryDispatchQueue().

std::mutex QueryDispatchQueue::update_delete_mutex_

private

Definition at line 118 of file QueryDispatchQueue.h.

Referenced by submit().

std::vector<std::thread> QueryDispatchQueue::workers_

private

Definition at line 122 of file QueryDispatchQueue.h.

Referenced by QueryDispatchQueue(), submit(), and ~QueryDispatchQueue().

---

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

• /home/jenkins-slave/workspace/core-os-doxygen/QueryEngine/QueryDispatchQueue.h