Commit 126f7ac8 authored by Dave Watson's avatar Dave Watson Committed by Facebook Github Bot

Make BlockingQueue::add() return a result struct

Summary: For greater clarity, return a BlockingQueueResult struct from add instead of a raw bool

Reviewed By: yfeldblum, magedm

Differential Revision: D7709300

fbshipit-source-id: 36fce6b1006ccf29985f431515c1c61deb84b6d3
parent d42832d2
...@@ -101,8 +101,9 @@ void CPUThreadPoolExecutor::add( ...@@ -101,8 +101,9 @@ void CPUThreadPoolExecutor::add(
Func func, Func func,
std::chrono::milliseconds expiration, std::chrono::milliseconds expiration,
Func expireCallback) { Func expireCallback) {
if (!taskQueue_->add( auto result = taskQueue_->add(
CPUTask(std::move(func), expiration, std::move(expireCallback)))) { CPUTask(std::move(func), expiration, std::move(expireCallback)));
if (!result.reusedThread) {
ensureActiveThreads(); ensureActiveThreads();
} }
} }
...@@ -117,9 +118,10 @@ void CPUThreadPoolExecutor::add( ...@@ -117,9 +118,10 @@ void CPUThreadPoolExecutor::add(
std::chrono::milliseconds expiration, std::chrono::milliseconds expiration,
Func expireCallback) { Func expireCallback) {
CHECK(getNumPriorities() > 0); CHECK(getNumPriorities() > 0);
if (!taskQueue_->addWithPriority( auto result = taskQueue_->addWithPriority(
CPUTask(std::move(func), expiration, std::move(expireCallback)), CPUTask(std::move(func), expiration, std::move(expireCallback)),
priority)) { priority);
if (!result.reusedThread) {
ensureActiveThreads(); ensureActiveThreads();
} }
} }
......
...@@ -294,7 +294,7 @@ void ThreadPoolExecutor::subscribeToTaskStats(TaskStatsCallback cb) { ...@@ -294,7 +294,7 @@ void ThreadPoolExecutor::subscribeToTaskStats(TaskStatsCallback cb) {
taskStatsCallbacks_->callbackList.wlock()->push_back(std::move(cb)); taskStatsCallbacks_->callbackList.wlock()->push_back(std::move(cb));
} }
bool ThreadPoolExecutor::StoppedThreadQueue::add( BlockingQueueAddResult 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));
......
...@@ -267,7 +267,7 @@ class ThreadPoolExecutor : public virtual folly::Executor { ...@@ -267,7 +267,7 @@ class ThreadPoolExecutor : public virtual folly::Executor {
class StoppedThreadQueue : public BlockingQueue<ThreadPtr> { class StoppedThreadQueue : public BlockingQueue<ThreadPtr> {
public: public:
bool add(ThreadPtr item) override; BlockingQueueAddResult 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(
......
...@@ -34,6 +34,11 @@ class FOLLY_EXPORT QueueFullException : public std::runtime_error { ...@@ -34,6 +34,11 @@ class FOLLY_EXPORT QueueFullException : public std::runtime_error {
using std::runtime_error::runtime_error; // Inherit constructors. using std::runtime_error::runtime_error; // Inherit constructors.
}; };
struct BlockingQueueAddResult {
BlockingQueueAddResult(bool reused = false) : reusedThread(reused) {}
bool reusedThread;
};
template <class T> template <class T>
class BlockingQueue { class BlockingQueue {
public: public:
...@@ -43,8 +48,10 @@ class BlockingQueue { ...@@ -43,8 +48,10 @@ class BlockingQueue {
// Returns true if an existing thread was able to work on it (used // Returns true if an existing thread was able to work on it (used
// for dynamically sizing thread pools), false otherwise. Return false // for dynamically sizing thread pools), false otherwise. Return false
// if this feature is not supported. // if this feature is not supported.
virtual bool add(T item) = 0; virtual BlockingQueueAddResult add(T item) = 0;
virtual bool addWithPriority(T item, int8_t /* priority */) { virtual BlockingQueueAddResult addWithPriority(
T item,
int8_t /* priority */) {
return add(std::move(item)); return add(std::move(item));
} }
virtual uint8_t getNumPriorities() { virtual uint8_t getNumPriorities() {
......
...@@ -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) {}
bool add(T item) override { BlockingQueueAddResult 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))) {
......
...@@ -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
bool add(T item) override { BlockingQueueAddResult add(T item) override {
return addWithPriority(std::move(item), folly::Executor::MID_PRI); return addWithPriority(std::move(item), folly::Executor::MID_PRI);
} }
bool addWithPriority(T item, int8_t priority) override { BlockingQueueAddResult 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)
......
...@@ -27,7 +27,7 @@ class UnboundedBlockingQueue : public BlockingQueue<T> { ...@@ -27,7 +27,7 @@ class UnboundedBlockingQueue : public BlockingQueue<T> {
public: public:
virtual ~UnboundedBlockingQueue() {} virtual ~UnboundedBlockingQueue() {}
bool add(T item) override { BlockingQueueAddResult add(T item) override {
queue_.enqueue(std::move(item)); queue_.enqueue(std::move(item));
return sem_.post(); return sem_.post();
} }
......
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