Commit a6ddaecc authored by Andrii Grynenko's avatar Andrii Grynenko Committed by Facebook Github Bot

Fix NotificationQueueExecutor to use RequestContextScopeGuard

Reviewed By: yfeldblum

Differential Revision: D7819076

fbshipit-source-id: 1dae43c27b88ea1a78d2a54f556687e8c4bbc008
parent 1f5581d8
...@@ -237,6 +237,9 @@ class NotificationQueue { ...@@ -237,6 +237,9 @@ class NotificationQueue {
return queue_.eventfd_ >= 0 ? queue_.eventfd_ : queue_.pipeFds_[0]; return queue_.eventfd_ >= 0 ? queue_.eventfd_ : queue_.pipeFds_[0];
} }
template <typename F>
void consumeUntilDrained(F&& foreach);
private: private:
NotificationQueue& queue_; NotificationQueue& queue_;
}; };
...@@ -843,6 +846,36 @@ bool NotificationQueue<MessageT>::Consumer::consumeUntilDrained( ...@@ -843,6 +846,36 @@ bool NotificationQueue<MessageT>::Consumer::consumeUntilDrained(
return true; return true;
} }
template <typename MessageT>
template <typename F>
void NotificationQueue<MessageT>::SimpleConsumer::consumeUntilDrained(
F&& foreach) {
SCOPE_EXIT {
queue_.syncSignalAndQueue();
};
queue_.checkPid();
while (true) {
std::unique_ptr<Node> data;
{
folly::SpinLockGuard g(queue_.spinlock_);
if (UNLIKELY(queue_.queue_.empty())) {
return;
}
data.reset(&queue_.queue_.front());
queue_.queue_.pop_front();
}
RequestContextScopeGuard rctx(std::move(data->ctx_));
foreach(std::move(data->msg_));
// Make sure message destructor is called with the correct RequestContext.
data.reset();
}
}
/** /**
* Creates a NotificationQueue::Consumer wrapping a function object * Creates a NotificationQueue::Consumer wrapping a function object
* Modeled after AsyncTimeout::make * Modeled after AsyncTimeout::make
......
...@@ -37,18 +37,15 @@ class AsyncioExecutor : public DrivableExecutor, public SequencedExecutor { ...@@ -37,18 +37,15 @@ class AsyncioExecutor : public DrivableExecutor, public SequencedExecutor {
} }
void drive() noexcept override { void drive() noexcept override {
Func func; consumer_.consumeUntilDrained([](Func&& func) {
while (queue_.tryConsume(func)) {
try { try {
func(); func();
} catch (const std::exception& ex) {
LOG(ERROR) << "Exception thrown by AsyncioExecutor task."
<< "Exception message: " << folly::exceptionStr(ex);
} catch (...) { } catch (...) {
LOG(ERROR) << "Unknown Exception thrown " LOG(ERROR) << "Exception thrown by NotificationQueueExecutor task."
<< "by AsyncioExecutor task."; << "Exception message: "
<< folly::exceptionStr(std::current_exception());
} }
} });
} }
private: private:
......
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