Commit 6dad33c9 authored by Yedidya Feldblum's avatar Yedidya Feldblum Committed by Facebook Github Bot

Fix a regression in per-task allocations in SerialExecutor

Summary: [Folly] Fix a regression in per-task allocations in `SerialExecutor`.

Reviewed By: andriigrynenko

Differential Revision: D15589007

fbshipit-source-id: 4461798435833cb317b41cfea932762eb7ac8662
parent 4c6147af
...@@ -19,7 +19,6 @@ ...@@ -19,7 +19,6 @@
#include <glog/logging.h> #include <glog/logging.h>
#include <folly/ExceptionString.h> #include <folly/ExceptionString.h>
#include <folly/io/async/Request.h>
namespace folly { namespace folly {
...@@ -58,16 +57,12 @@ void SerialExecutor::keepAliveRelease() { ...@@ -58,16 +57,12 @@ void SerialExecutor::keepAliveRelease() {
} }
void SerialExecutor::add(Func func) { void SerialExecutor::add(Func func) {
queue_.enqueue([ctx = folly::RequestContext::saveContext(), queue_.enqueue(Task{std::move(func), RequestContext::saveContext()});
func = std::move(func)]() mutable {
folly::RequestContextScopeGuard ctxGuard(ctx);
func();
});
parent_->add([keepAlive = getKeepAliveToken(this)] { keepAlive->run(); }); parent_->add([keepAlive = getKeepAliveToken(this)] { keepAlive->run(); });
} }
void SerialExecutor::addWithPriority(Func func, int8_t priority) { void SerialExecutor::addWithPriority(Func func, int8_t priority) {
queue_.enqueue(std::move(func)); queue_.enqueue(Task{std::move(func), RequestContext::saveContext()});
parent_->addWithPriority( parent_->addWithPriority(
[keepAlive = getKeepAliveToken(this)] { keepAlive->run(); }, priority); [keepAlive = getKeepAliveToken(this)] { keepAlive->run(); }, priority);
} }
...@@ -80,11 +75,12 @@ void SerialExecutor::run() { ...@@ -80,11 +75,12 @@ void SerialExecutor::run() {
} }
do { do {
Func func; Task task;
queue_.dequeue(func); queue_.dequeue(task);
try { try {
func(); folly::RequestContextScopeGuard ctxGuard(std::move(task.ctx));
task.func();
} catch (std::exception const& ex) { } catch (std::exception const& ex) {
LOG(ERROR) << "SerialExecutor: func threw unhandled exception " LOG(ERROR) << "SerialExecutor: func threw unhandled exception "
<< folly::exceptionStr(ex); << folly::exceptionStr(ex);
......
...@@ -23,6 +23,7 @@ ...@@ -23,6 +23,7 @@
#include <folly/concurrency/UnboundedQueue.h> #include <folly/concurrency/UnboundedQueue.h>
#include <folly/executors/GlobalExecutor.h> #include <folly/executors/GlobalExecutor.h>
#include <folly/executors/SequencedExecutor.h> #include <folly/executors/SequencedExecutor.h>
#include <folly/io/async/Request.h>
namespace folly { namespace folly {
...@@ -105,6 +106,11 @@ class SerialExecutor : public SequencedExecutor { ...@@ -105,6 +106,11 @@ class SerialExecutor : public SequencedExecutor {
void keepAliveRelease() override; void keepAliveRelease() override;
private: private:
struct Task {
Func func;
std::shared_ptr<RequestContext> ctx;
};
explicit SerialExecutor(KeepAlive<Executor> parent); explicit SerialExecutor(KeepAlive<Executor> parent);
~SerialExecutor() override; ~SerialExecutor() override;
...@@ -116,7 +122,7 @@ class SerialExecutor : public SequencedExecutor { ...@@ -116,7 +122,7 @@ class SerialExecutor : public SequencedExecutor {
* Unbounded multi producer single consumer queue where consumers don't block * Unbounded multi producer single consumer queue where consumers don't block
* on dequeue. * on dequeue.
*/ */
folly::UnboundedQueue<Func, false, true, false> queue_; folly::UnboundedQueue<Task, false, true, false> queue_;
std::atomic<ssize_t> keepAliveCounter_{1}; std::atomic<ssize_t> keepAliveCounter_{1};
}; };
......
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