Commit a876ba8b authored by Andrii Grynenko's avatar Andrii Grynenko Committed by Facebook GitHub Bot

Minor AtomicNotificationQueue cleanup

Summary: Following up on some comments on the initial diff.

Differential Revision: D24489715

fbshipit-source-id: baea66c7b811ebabf4476d7fd6a1136ab9ec2282
parent a7c4b5ba
...@@ -22,6 +22,17 @@ ...@@ -22,6 +22,17 @@
namespace folly { namespace folly {
void AtomicNotificationQueue::Task::execute() && noexcept {
RequestContextScopeGuard rctx(std::move(rctx_));
try {
func_();
func_ = nullptr;
} catch (...) {
LOG(FATAL) << "Exception thrown by a task in AtomicNotificationQueue: "
<< exceptionStr(std::current_exception());
}
}
AtomicNotificationQueue::Queue::Queue(Queue&& other) noexcept AtomicNotificationQueue::Queue::Queue(Queue&& other) noexcept
: head_(std::exchange(other.head_, nullptr)), : head_(std::exchange(other.head_, nullptr)),
size_(std::exchange(other.size_, 0)) {} size_(std::exchange(other.size_, 0)) {}
...@@ -273,13 +284,7 @@ bool AtomicNotificationQueue::drive() { ...@@ -273,13 +284,7 @@ bool AtomicNotificationQueue::drive() {
} }
} }
queueSize_.fetch_sub(1, std::memory_order_relaxed); queueSize_.fetch_sub(1, std::memory_order_relaxed);
RequestContextScopeGuard rctx(std::move(queue_.front().second)); std::move(queue_.front()).execute();
try {
queue_.front().first();
} catch (...) {
LOG(FATAL) << "Exception thrown by a task in AtomicNotificationQueue: "
<< exceptionStr(std::current_exception());
}
queue_.pop(); queue_.pop();
} }
return i > 0; return i > 0;
......
...@@ -33,11 +33,21 @@ namespace folly { ...@@ -33,11 +33,21 @@ namespace folly {
* A producer-consumer queue for passing tasks to EventBase thread. * A producer-consumer queue for passing tasks to EventBase thread.
* *
* Tasks can be added to the queue from any thread. A single EventBase * Tasks can be added to the queue from any thread. A single EventBase
* thread can be listening to the queue. Tasks are processed in the FIFO order. * thread can be listening to the queue. Tasks are processed in FIFO order.
*/ */
class AtomicNotificationQueue : private EventBase::LoopCallback, class AtomicNotificationQueue : private EventBase::LoopCallback,
private EventHandler { private EventHandler {
using Task = std::pair<Func, std::shared_ptr<RequestContext>>; class Task {
public:
Task(Func&& func, std::shared_ptr<RequestContext> rctx)
: func_(std::move(func)), rctx_(std::move(rctx)) {}
void execute() && noexcept;
private:
Func func_;
std::shared_ptr<RequestContext> rctx_;
};
class AtomicQueue; class AtomicQueue;
class Queue { class Queue {
public: public:
......
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