Commit 8202eccd authored by Dave Watson's avatar Dave Watson Committed by Facebook Github Bot

BlockingQueue returns bool for add

Summary:
Add a return value to BlockingQueue interface - add() returns true if
we were able to successfully handoff this work item to an existing thread.

Implementation is straightforward in LifoSem.

Reviewed By: magedm

Differential Revision: D7477279

fbshipit-source-id: 56ea17672a97ad722fd190baf0433ac68c440750
parent 42506a07
...@@ -206,11 +206,11 @@ void ThreadPoolExecutor::subscribeToTaskStats(TaskStatsCallback cb) { ...@@ -206,11 +206,11 @@ void ThreadPoolExecutor::subscribeToTaskStats(TaskStatsCallback cb) {
taskStatsCallbacks_->callbackList.wlock()->push_back(std::move(cb)); taskStatsCallbacks_->callbackList.wlock()->push_back(std::move(cb));
} }
void ThreadPoolExecutor::StoppedThreadQueue::add( bool ThreadPoolExecutor::StoppedThreadQueue::add(
ThreadPoolExecutor::ThreadPtr item) { ThreadPoolExecutor::ThreadPtr item) {
std::lock_guard<std::mutex> guard(mutex_); std::lock_guard<std::mutex> guard(mutex_);
queue_.push(std::move(item)); queue_.push(std::move(item));
sem_.post(); return sem_.post();
} }
ThreadPoolExecutor::ThreadPtr ThreadPoolExecutor::StoppedThreadQueue::take() { ThreadPoolExecutor::ThreadPtr ThreadPoolExecutor::StoppedThreadQueue::take() {
......
...@@ -234,7 +234,7 @@ class ThreadPoolExecutor : public virtual folly::Executor { ...@@ -234,7 +234,7 @@ class ThreadPoolExecutor : public virtual folly::Executor {
class StoppedThreadQueue : public BlockingQueue<ThreadPtr> { class StoppedThreadQueue : public BlockingQueue<ThreadPtr> {
public: public:
void add(ThreadPtr item) override; bool add(ThreadPtr item) override;
ThreadPtr take() override; ThreadPtr take() override;
size_t size() override; size_t size() override;
folly::Optional<ThreadPtr> try_take_for( folly::Optional<ThreadPtr> try_take_for(
......
...@@ -38,9 +38,14 @@ template <class T> ...@@ -38,9 +38,14 @@ template <class T>
class BlockingQueue { class BlockingQueue {
public: public:
virtual ~BlockingQueue() = default; virtual ~BlockingQueue() = default;
virtual void add(T item) = 0; // Adds item to the queue (with priority).
virtual void addWithPriority(T item, int8_t /* priority */) { //
add(std::move(item)); // Returns true if an existing thread was able to work on it (used
// for dynamically sizing thread pools), false otherwise. Return false
// if this feature is not supported.
virtual bool add(T item) = 0;
virtual bool addWithPriority(T item, int8_t /* priority */) {
return add(std::move(item));
} }
virtual uint8_t getNumPriorities() { virtual uint8_t getNumPriorities() {
return 1; return 1;
......
...@@ -28,7 +28,7 @@ class LifoSemMPMCQueue : public BlockingQueue<T> { ...@@ -28,7 +28,7 @@ class LifoSemMPMCQueue : public BlockingQueue<T> {
// Note: The queue pre-allocates all memory for max_capacity // Note: The queue pre-allocates all memory for max_capacity
explicit LifoSemMPMCQueue(size_t max_capacity) : queue_(max_capacity) {} explicit LifoSemMPMCQueue(size_t max_capacity) : queue_(max_capacity) {}
void add(T item) override { bool add(T item) override {
switch (kBehavior) { // static switch (kBehavior) { // static
case QueueBehaviorIfFull::THROW: case QueueBehaviorIfFull::THROW:
if (!queue_.write(std::move(item))) { if (!queue_.write(std::move(item))) {
...@@ -39,7 +39,7 @@ class LifoSemMPMCQueue : public BlockingQueue<T> { ...@@ -39,7 +39,7 @@ class LifoSemMPMCQueue : public BlockingQueue<T> {
queue_.blockingWrite(std::move(item)); queue_.blockingWrite(std::move(item));
break; break;
} }
sem_.post(); return sem_.post();
} }
T take() override { T take() override {
......
...@@ -52,11 +52,11 @@ class PriorityLifoSemMPMCQueue : public BlockingQueue<T> { ...@@ -52,11 +52,11 @@ class PriorityLifoSemMPMCQueue : public BlockingQueue<T> {
} }
// Add at medium priority by default // Add at medium priority by default
void add(T item) override { bool add(T item) override {
addWithPriority(std::move(item), folly::Executor::MID_PRI); return addWithPriority(std::move(item), folly::Executor::MID_PRI);
} }
void addWithPriority(T item, int8_t priority) override { bool addWithPriority(T item, int8_t priority) override {
int mid = getNumPriorities() / 2; int mid = getNumPriorities() / 2;
size_t queue = priority < 0 size_t queue = priority < 0
? std::max(0, mid + priority) ? std::max(0, mid + priority)
...@@ -72,7 +72,7 @@ class PriorityLifoSemMPMCQueue : public BlockingQueue<T> { ...@@ -72,7 +72,7 @@ class PriorityLifoSemMPMCQueue : public BlockingQueue<T> {
queues_[queue].blockingWrite(std::move(item)); queues_[queue].blockingWrite(std::move(item));
break; break;
} }
sem_.post(); return sem_.post();
} }
T take() override { T take() override {
......
...@@ -27,9 +27,9 @@ class UnboundedBlockingQueue : public BlockingQueue<T> { ...@@ -27,9 +27,9 @@ class UnboundedBlockingQueue : public BlockingQueue<T> {
public: public:
virtual ~UnboundedBlockingQueue() {} virtual ~UnboundedBlockingQueue() {}
void add(T item) override { bool add(T item) override {
queue_.enqueue(std::move(item)); queue_.enqueue(std::move(item));
sem_.post(); return sem_.post();
} }
T take() override { T take() override {
......
...@@ -356,11 +356,13 @@ struct LifoSemBase { ...@@ -356,11 +356,13 @@ struct LifoSemBase {
LifoSemBase& operator=(LifoSemBase const&) = delete; LifoSemBase& operator=(LifoSemBase const&) = delete;
/// Silently saturates if value is already 2^32-1 /// Silently saturates if value is already 2^32-1
void post() { bool post() {
auto idx = incrOrPop(1); auto idx = incrOrPop(1);
if (idx != 0) { if (idx != 0) {
idxToNode(idx).handoff().post(); idxToNode(idx).handoff().post();
return true;
} }
return false;
} }
/// Equivalent to n calls to post(), except may be much more efficient. /// Equivalent to n calls to post(), except may be much more efficient.
......
Markdown is supported
0%
or
You are about to add 0 people to the discussion. Proceed with caution.
Finish editing this message first!
Please register or to comment