Commit ea078b87 authored by James Sedgwick's avatar James Sedgwick Committed by dcsommer

merge wangle/Executor.h and experimental/wangle/concurrent/Executor.h

Summary:
the one in concurrent/ is a bit more generic, so I kept that as Executor and renamed the existing one ScheduledExecutor

because Hans is surfing I took the liberty of renaming Action->Func as an alias for std::function<void()>, because I think it's more reflective
also kept the version of add() that doesn't force rvalue-reference as it's more user friendly and probably not less performant in common cases (insert reference to "want speed? pass by value" here)

Test Plan: compiled some major relevant bits, will let contbuild show me anything I missed

Reviewed By: hans@fb.com

Subscribers: trunkagent, rushix, fbcode-common-diffs@, fugalh, msk, njormrod

FB internal diff: D1591237

Tasks: 5279196
parent e5e3fde0
...@@ -15,7 +15,7 @@ ...@@ -15,7 +15,7 @@
*/ */
#pragma once #pragma once
#include <folly/experimental/wangle/concurrent/Executor.h> #include <folly/wangle/Executor.h>
#include <thread> #include <thread>
......
...@@ -15,7 +15,7 @@ ...@@ -15,7 +15,7 @@
*/ */
#pragma once #pragma once
#include <folly/experimental/wangle/concurrent/Executor.h> #include <folly/wangle/Executor.h>
#include <folly/experimental/wangle/concurrent/LifoSemMPMCQueue.h> #include <folly/experimental/wangle/concurrent/LifoSemMPMCQueue.h>
#include <folly/experimental/wangle/concurrent/NamedThreadFactory.h> #include <folly/experimental/wangle/concurrent/NamedThreadFactory.h>
#include <folly/experimental/wangle/rx/Observable.h> #include <folly/experimental/wangle/rx/Observable.h>
...@@ -31,7 +31,7 @@ ...@@ -31,7 +31,7 @@
namespace folly { namespace wangle { namespace folly { namespace wangle {
class ThreadPoolExecutor : public experimental::Executor { class ThreadPoolExecutor : public Executor {
public: public:
explicit ThreadPoolExecutor( explicit ThreadPoolExecutor(
size_t numThreads, size_t numThreads,
......
...@@ -17,59 +17,29 @@ ...@@ -17,59 +17,29 @@
#pragma once #pragma once
#include <boost/noncopyable.hpp> #include <boost/noncopyable.hpp>
#include <chrono>
#include <functional> #include <functional>
#include <memory>
#include <stdexcept>
namespace folly { namespace wangle { namespace folly { namespace wangle {
typedef std::function<void()> Func;
/// An Executor accepts units of work with add(), which should be /// An Executor accepts units of work with add(), which should be
/// threadsafe. /// threadsafe.
/// Like an Rx Scheduler. We should probably rename it to match now that it
/// has scheduling semantics too, but that's a codemod for another lazy
/// summer afternoon.
class Executor : boost::noncopyable { class Executor : boost::noncopyable {
public: public:
typedef std::function<void()> Action; virtual ~Executor() = default;
// Reality is that better than millisecond resolution is very hard to
// achieve. However, we reserve the right to be incredible. /// Enqueue a function to executed by this executor. This and all
typedef std::chrono::microseconds Duration; /// variants must be threadsafe.
typedef std::chrono::steady_clock::time_point TimePoint; virtual void add(Func) = 0;
virtual ~Executor() = default; /// A convenience function for shared_ptr to legacy functors.
///
/// Enqueue an action to be performed by this executor. This and all /// Sometimes you have a functor that is move-only, and therefore can't be
/// schedule variants must be threadsafe. /// converted to a std::function (e.g. std::packaged_task). In that case,
virtual void add(Action&&) = 0; /// wrap it in a shared_ptr (or maybe folly::MoveWrapper) and use this.
template <class P>
/// A convenience function for shared_ptr to legacy functors. void addPtr(P fn) {
/// this->add([fn]() mutable { (*fn)(); });
/// Sometimes you have a functor that is move-only, and therefore can't be }
/// converted to a std::function (e.g. std::packaged_task). In that case,
/// wrap it in a shared_ptr (or maybe folly::MoveWrapper) and use this.
template <class P>
void addPtr(P fn) {
this->add([fn]() mutable { (*fn)(); });
}
/// Alias for add() (for Rx consistency)
void schedule(Action&& a) { add(std::move(a)); }
/// Schedule an action to be executed after dur time has elapsed
/// Expect millisecond resolution at best.
void schedule(Action&& a, Duration const& dur) {
scheduleAt(std::move(a), now() + dur);
}
/// Schedule an action to be executed at time t, or as soon afterward as
/// possible. Expect millisecond resolution at best. Must be threadsafe.
virtual void scheduleAt(Action&& a, TimePoint const& t) {
throw std::logic_error("unimplemented");
}
/// Get this executor's notion of time. Must be threadsafe.
virtual TimePoint now() {
return std::chrono::steady_clock::now();
}
}; };
}} }}
...@@ -24,7 +24,7 @@ namespace folly { namespace wangle { ...@@ -24,7 +24,7 @@ namespace folly { namespace wangle {
/// QueuedImmediateExecutor. /// QueuedImmediateExecutor.
class InlineExecutor : public Executor { class InlineExecutor : public Executor {
public: public:
void add(std::function<void()>&& f) override { void add(Func f) override {
f(); f();
} }
}; };
......
...@@ -30,35 +30,35 @@ ManualExecutor::ManualExecutor() { ...@@ -30,35 +30,35 @@ ManualExecutor::ManualExecutor() {
} }
} }
void ManualExecutor::add(std::function<void()>&& callback) { void ManualExecutor::add(Func callback) {
std::lock_guard<std::mutex> lock(lock_); std::lock_guard<std::mutex> lock(lock_);
actions_.push(callback); funcs_.push(std::move(callback));
sem_post(&sem_); sem_post(&sem_);
} }
size_t ManualExecutor::run() { size_t ManualExecutor::run() {
size_t count; size_t count;
size_t n; size_t n;
Action action; Func func;
{ {
std::lock_guard<std::mutex> lock(lock_); std::lock_guard<std::mutex> lock(lock_);
while (!scheduledActions_.empty()) { while (!scheduledFuncs_.empty()) {
auto& sa = scheduledActions_.top(); auto& sf = scheduledFuncs_.top();
if (sa.time > now_) if (sf.time > now_)
break; break;
actions_.push(sa.action); funcs_.push(sf.func);
scheduledActions_.pop(); scheduledFuncs_.pop();
} }
n = actions_.size(); n = funcs_.size();
} }
for (count = 0; count < n; count++) { for (count = 0; count < n; count++) {
{ {
std::lock_guard<std::mutex> lock(lock_); std::lock_guard<std::mutex> lock(lock_);
if (actions_.empty()) { if (funcs_.empty()) {
break; break;
} }
...@@ -67,10 +67,10 @@ size_t ManualExecutor::run() { ...@@ -67,10 +67,10 @@ size_t ManualExecutor::run() {
// This may fail (with EAGAIN), that's fine. // This may fail (with EAGAIN), that's fine.
sem_trywait(&sem_); sem_trywait(&sem_);
action = std::move(actions_.front()); func = std::move(funcs_.front());
actions_.pop(); funcs_.pop();
} }
action(); func();
} }
return count; return count;
...@@ -80,7 +80,7 @@ void ManualExecutor::wait() { ...@@ -80,7 +80,7 @@ void ManualExecutor::wait() {
while (true) { while (true) {
{ {
std::lock_guard<std::mutex> lock(lock_); std::lock_guard<std::mutex> lock(lock_);
if (!actions_.empty()) if (!funcs_.empty())
break; break;
} }
......
...@@ -15,7 +15,7 @@ ...@@ -15,7 +15,7 @@
*/ */
#pragma once #pragma once
#include <folly/wangle/Executor.h> #include <folly/wangle/ScheduledExecutor.h>
#include <semaphore.h> #include <semaphore.h>
#include <memory> #include <memory>
#include <mutex> #include <mutex>
...@@ -31,15 +31,15 @@ namespace folly { namespace wangle { ...@@ -31,15 +31,15 @@ namespace folly { namespace wangle {
/// ///
/// NB No attempt has been made to make anything other than add and schedule /// NB No attempt has been made to make anything other than add and schedule
/// threadsafe. /// threadsafe.
class ManualExecutor : public Executor { class ManualExecutor : public ScheduledExecutor {
public: public:
ManualExecutor(); ManualExecutor();
void add(Action&&) override; void add(Func) override;
/// Do work. Returns the number of actions that were executed (maybe 0). /// Do work. Returns the number of functions that were executed (maybe 0).
/// Non-blocking, in the sense that we don't wait for work (we can't /// Non-blocking, in the sense that we don't wait for work (we can't
/// control whether one of the actions blocks). /// control whether one of the functions blocks).
/// This is stable, it will not chase an ever-increasing tail of work. /// This is stable, it will not chase an ever-increasing tail of work.
/// This also means, there may be more work available to perform at the /// This also means, there may be more work available to perform at the
/// moment that this returns. /// moment that this returns.
...@@ -60,9 +60,9 @@ namespace folly { namespace wangle { ...@@ -60,9 +60,9 @@ namespace folly { namespace wangle {
makeProgress(); makeProgress();
} }
virtual void scheduleAt(Action&& a, TimePoint const& t) override { virtual void scheduleAt(Func&& f, TimePoint const& t) override {
std::lock_guard<std::mutex> lock(lock_); std::lock_guard<std::mutex> lock(lock_);
scheduledActions_.emplace(t, std::move(a)); scheduledFuncs_.emplace(t, std::move(f));
sem_post(&sem_); sem_post(&sem_);
} }
...@@ -82,30 +82,30 @@ namespace folly { namespace wangle { ...@@ -82,30 +82,30 @@ namespace folly { namespace wangle {
private: private:
std::mutex lock_; std::mutex lock_;
std::queue<Action> actions_; std::queue<Func> funcs_;
sem_t sem_; sem_t sem_;
// helper class to enable ordering of scheduled events in the priority // helper class to enable ordering of scheduled events in the priority
// queue // queue
struct ScheduledAction { struct ScheduledFunc {
TimePoint time; TimePoint time;
size_t ordinal; size_t ordinal;
Action action; Func func;
ScheduledAction(TimePoint const& t, Action&& a) ScheduledFunc(TimePoint const& t, Func&& f)
: time(t), action(std::move(a)) : time(t), func(std::move(f))
{ {
static size_t seq = 0; static size_t seq = 0;
ordinal = seq++; ordinal = seq++;
} }
bool operator<(ScheduledAction const& b) const { bool operator<(ScheduledFunc const& b) const {
if (time == b.time) if (time == b.time)
return ordinal < b.ordinal; return ordinal < b.ordinal;
return time < b.time; return time < b.time;
} }
}; };
std::priority_queue<ScheduledAction> scheduledActions_; std::priority_queue<ScheduledFunc> scheduledFuncs_;
TimePoint now_ = now_.min(); TimePoint now_ = now_.min();
}; };
......
...@@ -20,9 +20,8 @@ ...@@ -20,9 +20,8 @@
namespace folly { namespace wangle { namespace folly { namespace wangle {
void QueuedImmediateExecutor::add(Action&& callback) void QueuedImmediateExecutor::add(Func callback) {
{ thread_local std::queue<Func> q;
thread_local std::queue<Action> q;
if (q.empty()) { if (q.empty()) {
q.push(std::move(callback)); q.push(std::move(callback));
......
...@@ -27,7 +27,7 @@ namespace folly { namespace wangle { ...@@ -27,7 +27,7 @@ namespace folly { namespace wangle {
*/ */
class QueuedImmediateExecutor : public Executor { class QueuedImmediateExecutor : public Executor {
public: public:
void add(Action&&) override; void add(Func) override;
}; };
}} // namespace }} // namespace
...@@ -16,20 +16,42 @@ ...@@ -16,20 +16,42 @@
#pragma once #pragma once
#include <functional> #include <folly/wangle/Executor.h>
#include <chrono>
#include <memory>
#include <stdexcept>
namespace folly { namespace wangle { namespace folly { namespace wangle {
// An executor that supports timed scheduling. Like RxScheduler.
typedef std::function<void()> Func; class ScheduledExecutor : public Executor {
public:
namespace experimental { // TODO(jsedgwick) merge with folly/wangle/Executor.h // Reality is that better than millisecond resolution is very hard to
// achieve. However, we reserve the right to be incredible.
class Executor { typedef std::chrono::microseconds Duration;
public: typedef std::chrono::steady_clock::time_point TimePoint;
virtual ~Executor() {};
virtual void add(Func func) = 0; virtual ~ScheduledExecutor() = default;
};
virtual void add(Func) override = 0;
}
/// Alias for add() (for Rx consistency)
}} // folly::wangle void schedule(Func&& a) { add(std::move(a)); }
/// Schedule a Func to be executed after dur time has elapsed
/// Expect millisecond resolution at best.
void schedule(Func&& a, Duration const& dur) {
scheduleAt(std::move(a), now() + dur);
}
/// Schedule a Func to be executed at time t, or as soon afterward as
/// possible. Expect millisecond resolution at best. Must be threadsafe.
virtual void scheduleAt(Func&& a, TimePoint const& t) {
throw std::logic_error("unimplemented");
}
/// Get this executor's notion of time. Must be threadsafe.
virtual TimePoint now() {
return std::chrono::steady_clock::now();
}
};
}}
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