/** \file
* ucontext-based coroutine library designed to emulate a normal control flow around callbacks. */
// http://en.wikipedia.org/wiki/Setcontext; man 3 makecontext; man 2 getcontext
// http://www.boost.org/doc/libs/1_53_0/libs/context/doc/html/index.html
// g++ -std=c++11 -O1 -Wall -g test_cbcoro.cc -pthread && ./a.out
// NB: There is now a coroutine support in Boost ASIO which can be used to make asynchronous APIs look synchronous in a similar way:
// https://svn.boost.org/trac/boost/changeset/84311
#include <ucontext.h>
#include <sys/mman.h> // mmap
#include <string.h> // strerror
#include <mutex>
#include <atomic>
#include <valgrind/valgrind.h>
#include <glim/exception.hpp>
#include <boost/container/flat_map.hpp>
#include <boost/container/slist.hpp>
namespace glim {
/// Simplifies turning callback control flows into normal imperative control flows.
class CBCoro {
public:
/// "Holds" the CBCoro and will delete it when it is no longer used.
struct CBCoroPtr {
CBCoro* _coro;
CBCoroPtr (CBCoro* coro): _coro (coro) {
_coro->_users++;
}
~CBCoroPtr() {
if (--_coro->_users <= 0 && _coro->_delete) delete _coro;
}
CBCoro* operator ->() const {return _coro;}
};
static constexpr size_t defaultStackSize() {return 512 * 1024;}
static constexpr uint8_t defaultCacheSize() {return 2;}
protected:
typedef boost::container::flat_map<size_t, boost::container::slist<void*> > cache_t;
/// The cached stacks; stackSize -> free list.
static cache_t& cache() {static cache_t CACHE; return CACHE;}
static std::mutex& cacheMutex() {static std::mutex CACHE_MUTEX; return CACHE_MUTEX;}
ucontext_t _context;
ucontext_t* _returnTo;
std::recursive_mutex _mutex; ///< This one is locked most of the time.
std::atomic_int_fast32_t _users; ///< Counter used by `CBCoroPtr`.
bool _delete; ///< Whether the `CBCoroPtr` should `delete` this instance when it is no longer used (default is `true`).
bool _invokeFromYield; ///< True if `invokeFromCallback()` was called directly from `yieldForCallback()`.
bool _yieldFromInvoke; ///< True if `yieldForCallback()` now runs from `invokeFromCallback()`.
uint8_t const _cacheStack; ///< Tells `freeStack()` to cache the stack if the number of cached `#_stackSize` stacks is less than it.
void* _stack;
size_t const _stackSize; ///< Keeps the size of the stack.
/// Peek a stack from the cache or allocate one with `mmap` (and register with Valgrind).
virtual void allocateStack() {
if (_cacheStack) {
std::lock_guard<std::mutex> lock (cacheMutex());
auto& freeList = cache()[_stackSize];
if (!freeList.empty()) {_stack = freeList.front(); freeList.pop_front(); return;}
}
_stack = mmap (nullptr, _stackSize, PROT_READ | PROT_WRITE, MAP_PRIVATE | MAP_ANONYMOUS | MAP_STACK | MAP_NORESERVE, -1, 0);
if (_stack == MAP_FAILED) GTHROW (std::string ("mmap allocation failed: ") + ::strerror (errno));
#pragma GCC diagnostic ignored "-Wunused-value"
VALGRIND_STACK_REGISTER (_stack, (char*) _stack + _stackSize);
}
/// Release a stack into the cache or free it with `munmap` (and deregister with Valgrind).
virtual void freeStack() {
if (_cacheStack) {
std::lock_guard<std::mutex> lock (cacheMutex());
auto& freeList = cache()[_stackSize];
if (freeList.size() < _cacheStack) {freeList.push_front (_stack); _stack = nullptr; return;}
}
VALGRIND_STACK_DEREGISTER (_stack);
if (munmap (_stack, _stackSize)) GTHROW (std::string ("!munmap: ") + ::strerror (errno));;
_stack = nullptr;
}
/// Prepare the coroutine (initialize context, allocate stack and register it with Valgrind).
CBCoro (uint8_t cacheStack = defaultCacheSize(), size_t stackSize = defaultStackSize()):
_returnTo (nullptr), _users (0), _delete (true), _invokeFromYield (false), _yieldFromInvoke (false),
_cacheStack (cacheStack), _stack (nullptr), _stackSize (stackSize) {
if (getcontext (&_context)) GTHROW ("!getcontext");
allocateStack();
_context.uc_stack.ss_sp = _stack;
_context.uc_stack.ss_size = stackSize;
}
virtual ~CBCoro() {
freeStack();
}
public:
/// Starts the coroutine on the `_stack` (makecontext, swapcontext), calling the `CBCoro::run`.
CBCoroPtr start() {
CBCoroPtr ptr (this);
ucontext_t back; _context.uc_link = &back;
makecontext (&_context, (void(*)()) cbcRun, 1, (intptr_t) this);
// Since we have to "return" from inside the `yieldForCallback`,
// we're not actually using the `_context.uc_link` and `return`, we use `setcontext (_returnTo)` instead.
_returnTo = &back;
_mutex.lock();
swapcontext (&back, &_context); // Now our stack lives and the caller stack is no longer in control.
_mutex.unlock();
return ptr;
}
protected:
/// Logs exception thrown from `CBCoro::run`.
virtual void log (const std::exception& ex) {
std::cerr << "glim::CBCoro, exception: " << ex.what() << std::endl;
}
static void cbcRun (CBCoro* cbCoro) {
try {
cbCoro->run();
} catch (const std::exception& ex) {
cbCoro->log (ex);
}
cbCoro->cbcReturn(); // Return the control to the rightful owner, e.g. to a last callback who ran `invokeFromCallback`, or otherwise to `cbcStart`.
}
/// Relinquish the control to the original owner of the thread, restoring its stack.
void cbcReturn() {
ucontext_t* returnTo = _returnTo;
if (returnTo != nullptr) {_returnTo = nullptr; setcontext (returnTo);}
}
/// This method is performed on the CBCoro stack, allowing it to be suspended and then reanimated from callbacks.
virtual void run() = 0;
public:
/** Use this method to wrap a return-via-callback code.
* For example, the callback code \code
* startSomeWork ([=]() {
* continueWhenWorkIsFinished();
* });
* \endcode should be turned into \code
* yieldForCallback ([&]() {
* startSomeWork ([&]() {
* invokeFromCallback();
* });
* });
* continueWhenWorkIsFinished();
* \endcode
*
* Captures the stack, runs the `fun` and relinquish the control to `_returnTo`.\n
* This method will never "return" by itself, in order for it to "return" the
* `fun` MUST call `invokeFromCallback`, maybe later and from a different stack. */
template <typename F> CBCoroPtr yieldForCallback (F fun) {
CBCoroPtr ptr (this);
_yieldFromInvoke = false;
if (getcontext (&_context)) GTHROW ("!getcontext"); // Capture.
if (_yieldFromInvoke) {
// We're now in the future, revived by the `invokeFromCallback`.
// All we do now is "return" to the caller whose stack we captured earlier.
} else {
// We're still in the present, still have some work to do.
fun(); // The `fun` is supposed to do something resulting in the `invokeFromCallback` being called later.
if (_invokeFromYield) {
// The `fun` used the `invokeFromCallback` directly, not resorting to callbacks, meaning we don't have to do our magick.
_invokeFromYield = false;
} else {
// So, the `fun` took measures to revive us later, it's time for us to go into torpor and return the control to whoever we've borrowed it from.
cbcReturn();
}
}
return ptr;
}
/// To be called from a callback in order to lend the control to CBCoro, continuing it from where it called `yieldForCallback`.
CBCoroPtr invokeFromCallback() {
CBCoroPtr ptr (this);
_mutex.lock(); // Wait for an other-thready `yieldForCallback` to finish.
if (_returnTo != nullptr) {
// We have not yet "returned" from the `yieldForCallback`,
// meaning that the `invokeFromCallback` was executed immediately from inside the `yieldForCallback`.
// In that case we must DO NOTHING, we must simply continue running on the current stack.
_invokeFromYield = true; // Tells `yieldForCallback` to do nothing.
} else {
// Revive the CBCoro, letting it continue from where it was suspended in `yieldForCallback`.
ucontext_t cbContext; _returnTo = &cbContext; _yieldFromInvoke = true;
if (swapcontext (&cbContext, &_context)) GTHROW ("!swapcontext");
// NB: When the CBCoro is suspended or exits, the control returns back there and then back to the callback from which we borrowed it.
if (_returnTo == &cbContext) _returnTo = nullptr;
}
_mutex.unlock(); // Other-thready `yieldForCallback` has finished and `cbcReturn`ed here.
return ptr;
}
};
/** CBCoro running a given functor.
* The functor's first argument must be a CBCoro pointer, like this: \code (new CBCoroForFunctor ([](CBCoro* cbcoro) {}))->start(); \endcode */
template <typename FUN> struct CBCoroForFunctor: public CBCoro {
FUN _fun;
template <typename CFUN> CBCoroForFunctor (CFUN&& fun, uint8_t cacheStack, size_t stackSize): CBCoro (cacheStack, stackSize), _fun (std::forward<CFUN> (fun)) {}
virtual void run() {_fun (this);}
virtual ~CBCoroForFunctor() {}
};
/** Syntactic sugar: Runs a given functor in a CBCoro instance.
* Example: \code glim::cbCoro ([](glim::CBCoro* cbcoro) {}); \endcode
* Returns a `CBCoroPtr` to the CBCoro instance holding the `fun` which might be held somewhere in order to delay the deletion of `fun`. */
template <typename FUN> inline CBCoro::CBCoroPtr cbCoro (FUN&& fun, uint8_t cacheStack = CBCoro::defaultCacheSize(), size_t stackSize = CBCoro::defaultStackSize()) {
return (new CBCoroForFunctor<FUN> (std::forward<FUN> (fun), cacheStack, stackSize))->start();
}
}
