Commit 99cba475 authored by Dan Melnic's avatar Dan Melnic Committed by Facebook Github Bot

Add support for EventBase::runInEventBaseThreadAlwaysEnqueue

Summary: Add support for EventBase::runInEventBaseThreadAlwaysEnqueue

Reviewed By: yfeldblum

Differential Revision: D13366773

fbshipit-source-id: 72eacdbe699509d771fcefe31ee4d742eaadd4d1
parent e572a62c
......@@ -579,6 +579,28 @@ bool EventBase::runInEventBaseThread(Func fn) {
return true;
}
bool EventBase::runInEventBaseThreadAlwaysEnqueue(Func fn) {
// Send the message.
// It will be received by the FunctionRunner in the EventBase's thread.
// We try not to schedule nullptr callbacks
if (!fn) {
LOG(ERROR) << "EventBase " << this
<< ": Scheduling nullptr callbacks is not allowed";
return false;
}
try {
queue_->putMessage(std::move(fn));
} catch (const std::exception& ex) {
LOG(ERROR) << "EventBase " << this << ": failed to schedule function "
<< "for EventBase thread: " << ex.what();
return false;
}
return true;
}
bool EventBase::runInEventBaseThreadAndWait(Func fn) {
if (inRunningEventBaseThread()) {
LOG(ERROR) << "EventBase " << this << ": Waiting in the event loop is not "
......
......@@ -425,6 +425,55 @@ class EventBase : private boost::noncopyable,
*/
bool runInEventBaseThread(Func fn);
/**
* Run the specified function in the EventBase's thread.
*
* This method is thread-safe, and may be called from another thread.
*
* If runInEventBaseThreadAlwaysEnqueue() is called when the EventBase loop is
* not running, the function call will be delayed until the next time the loop
* is started.
*
* If runInEventBaseThreadAlwaysEnqueue() returns true the function has
* successfully been scheduled to run in the loop thread. However, if the
* loop is terminated (and never later restarted) before it has a chance to
* run the requested function, the function will be run upon the EventBase's
* destruction.
*
* If two calls to runInEventBaseThreadAlwaysEnqueue() are made from the same
* thread, the functions will always be run in the order that they were
* scheduled. Ordering between functions scheduled from separate threads is
* not guaranteed. If a call is made from the EventBase thread, the function
* will not be executed inline and will be queued to the same queue as if the
* call would have been made from a different thread
*
* @param fn The function to run. The function must not throw any
* exceptions.
* @param arg An argument to pass to the function.
*
* @return Returns true if the function was successfully scheduled, or false
* if there was an error scheduling the function.
*/
template <typename T>
bool runInEventBaseThreadAlwaysEnqueue(void (*fn)(T*), T* arg);
/**
* Run the specified function in the EventBase's thread
*
* This version of runInEventBaseThreadAlwaysEnqueue() takes a folly::Function
* object. Note that this may be less efficient than the version that takes a
* plain function pointer and void* argument, if moving the function is
* expensive (e.g., if it wraps a lambda which captures some values with
* expensive move constructors).
*
* If the loop is terminated (and never later restarted) before it has a
* chance to run the requested function, the function will be run upon the
* EventBase's destruction.
*
* The function must not throw any exceptions.
*/
bool runInEventBaseThreadAlwaysEnqueue(Func fn);
/*
* Like runInEventBaseThread, but the caller waits for the callback to be
* executed.
......@@ -795,6 +844,11 @@ bool EventBase::runInEventBaseThread(void (*fn)(T*), T* arg) {
return runInEventBaseThread([=] { fn(arg); });
}
template <typename T>
bool EventBase::runInEventBaseThreadAlwaysEnqueue(void (*fn)(T*), T* arg) {
return runInEventBaseThreadAlwaysEnqueue([=] { fn(arg); });
}
template <typename T>
bool EventBase::runInEventBaseThreadAndWait(void (*fn)(T*), T* arg) {
return runInEventBaseThreadAndWait([=] { fn(arg); });
......
......@@ -1559,6 +1559,56 @@ TEST(EventBaseTest, LoopTermination) {
close(pipeFds[0]);
}
TEST(EventBaseTest, CallbackOrderTest) {
size_t num = 0;
EventBase evb;
evb.runInEventBaseThread([&]() {
std::thread t([&]() {
evb.runInEventBaseThread([&]() {
num++;
EXPECT_EQ(num, 2);
});
});
t.join();
// this callback will run first
// even if it is scheduled after the first one
evb.runInEventBaseThread([&]() {
num++;
EXPECT_EQ(num, 1);
});
});
evb.loop();
EXPECT_EQ(num, 2);
}
TEST(EventBaseTest, AlwaysEnqueueCallbackOrderTest) {
size_t num = 0;
EventBase evb;
evb.runInEventBaseThread([&]() {
std::thread t([&]() {
evb.runInEventBaseThreadAlwaysEnqueue([&]() {
num++;
EXPECT_EQ(num, 1);
});
});
t.join();
// this callback will run second
// since it was enqueued after the first one
evb.runInEventBaseThreadAlwaysEnqueue([&]() {
num++;
EXPECT_EQ(num, 2);
});
});
evb.loop();
EXPECT_EQ(num, 2);
}
///////////////////////////////////////////////////////////////////////////
// Tests for latency calculations
///////////////////////////////////////////////////////////////////////////
......
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