dethreader.hpp

Go to the documentation of this file.

#pragma once
 
#include "exceptions.hpp"
#include "timing.hpp"
#include <functional>
#include <libco/libco.h>
#include <memory>
#include <queue>
#include <set>
 
#define __JAFFAR_COMMON_DETHREADER_STATE                                                                                                                                           \
  namespace jaffarCommon                                                                                                                                                           \
  {                                                                                                                                                                                \
  namespace dethreader                                                                                                                                                             \
  {                                                                                                                                                                                \
  Runtime* __runtime = nullptr;                                                                                                                                                    \
  }                                                                                                                                                                                \
  }
 
#define __JAFFAR_COMMON_DETHREADER_STACK_SIZE 4 * 1024 * 1024
 
namespace jaffarCommon
{
 
namespace dethreader
{
 
class Runtime;
 
extern Runtime* __runtime;
 
typedef std::function<void()> threadFc_t;
 
typedef uint64_t threadId_t;
 
class Runtime
{
public:
  class Thread
  {
  public:
    enum returnReason_t
    {
      none,
 
      finished,
 
      sleeping,
 
      waiting
    };
 
    friend class Runtime;
 
    Thread()                      = delete;
    Thread(const Thread&)         = delete;
    void operator=(const Thread&) = delete;
 
    Thread(const threadFc_t fc, const threadId_t id) : _id(id), _fc(fc)
    {
      constexpr size_t stackSize = __JAFFAR_COMMON_DETHREADER_STACK_SIZE;
      _coroutine                 = co_create(stackSize, Thread::coroutineWrapper);
    }
 
    ~Thread() { co_delete(_coroutine); }
 
    __JAFFAR_COMMON__INLINE__ void run() const
    {
      // If the thread is sleep, checking if it has finished
      if (_returnReason == returnReason_t::sleeping)
      {
        auto timeDelta = timing::timeDeltaMicroseconds(timing::now(), _sleepStartTime);
 
        // If not, return now without continuing
        if (timeDelta < _sleepDuration) return;
      }
 
      // If the thread is waiting for another to finish, check it now
      if (_returnReason == returnReason_t::waiting)
        if (__runtime->getFinishedThreads().contains(_threadWaitedFor) == false) return;
 
      // Starting or continuing execution
      co_switch(_coroutine);
    }
 
    __JAFFAR_COMMON__INLINE__ void yield()
    {
      _returnReason = returnReason_t::none;
      __runtime->yieldToRuntime();
    }
 
    __JAFFAR_COMMON__INLINE__ returnReason_t getReturnReason() const { return _returnReason; }
 
    __JAFFAR_COMMON__INLINE__ void sleep(const size_t sleepDuration)
    {
      _sleepDuration  = sleepDuration;
      _returnReason   = returnReason_t::sleeping;
      _sleepStartTime = timing::now();
      __runtime->yieldToRuntime();
    }
 
    __JAFFAR_COMMON__INLINE__ bool joinable() { return true; }
 
    __JAFFAR_COMMON__INLINE__ void join(const threadId_t threadId)
    {
      _threadWaitedFor = threadId;
      _returnReason    = returnReason_t::waiting;
      __runtime->yieldToRuntime();
    }
 
    __JAFFAR_COMMON__INLINE__ threadId_t getThreadId() const { return _id; }
 
    threadId_t _threadWaitedFor;
 
  private:
    __JAFFAR_COMMON__INLINE__ void setReturnReason(const returnReason_t returnReason) { _returnReason = returnReason; }
 
    __JAFFAR_COMMON__INLINE__ static void coroutineWrapper()
    {
      auto currentThread = __runtime->getCurrentThread();
      currentThread->_fc();
      currentThread->setReturnReason(returnReason_t::finished);
      __runtime->yieldToRuntime();
    }
 
    const threadId_t _id;
 
    const threadFc_t _fc;
 
    cothread_t _coroutine;
 
    returnReason_t _returnReason = none;
 
    size_t _sleepDuration;
 
    timing::timePoint _sleepStartTime;
  };
 
  Runtime() = default;
 
  __JAFFAR_COMMON__INLINE__ threadId_t createThread(const threadFc_t fc)
  {
    const auto threadId = _uniqueThreadIdCounter;
    _threadQueue.push(std::make_unique<Thread>(fc, threadId));
    _uniqueThreadIdCounter++;
    return threadId;
  }
 
  __JAFFAR_COMMON__INLINE__ void initialize()
  {
    // Setting singleton
    jaffarCommon::dethreader::__runtime = this;
  }
 
  __JAFFAR_COMMON__INLINE__ void finalize()
  {
    // unsetting singleton
    jaffarCommon::dethreader::__runtime = nullptr;
  }
 
  __JAFFAR_COMMON__INLINE__ void run()
  {
    // Getting main coroutine
    _coroutine = co_active();
 
    // Starting to run threads until they are all finished
    while (_threadQueue.empty() == false)
    {
      // Obtaining next thread to run
      auto thread = std::move(_threadQueue.front());
 
      // Removing thread from the front
      _threadQueue.pop();
 
      // Setting current thread for execution
      setCurrentThread(thread.get());
 
      // Running thread
      thread->run();
 
      // If thread not finished, re-add to the back of the queue
      if (thread->getReturnReason() != Thread::returnReason_t::finished)
        _threadQueue.push(std::move(thread));
      else // Otherwise add it to the set of finished threads
        _finishedThreads.insert(thread->getThreadId());
    }
  }
 
  __JAFFAR_COMMON__INLINE__ void setCurrentThread(Thread* const thread) { _currentThread = thread; }
 
  __JAFFAR_COMMON__INLINE__ Thread* getCurrentThread() const { return _currentThread; }
 
  __JAFFAR_COMMON__INLINE__ void yieldToRuntime() { co_switch(_coroutine); }
 
  __JAFFAR_COMMON__INLINE__ const std::set<threadId_t>& getFinishedThreads() const { return _finishedThreads; }
 
private:
  threadId_t _uniqueThreadIdCounter = 0;
 
  std::set<threadId_t> _finishedThreads;
 
  Thread* _currentThread = nullptr;
 
  std::queue<std::unique_ptr<Thread>> _threadQueue;
 
  cothread_t _coroutine;
};
 
__JAFFAR_COMMON__INLINE__ Runtime::Thread* getCurrentThread() { return __runtime->getCurrentThread(); }
 
__JAFFAR_COMMON__INLINE__ threadId_t createThread(const threadFc_t fc)
{
  if (__runtime == nullptr) JAFFAR_THROW_LOGIC("Trying to use dethreader runtime before it is initialized");
  return __runtime->createThread(fc);
}
 
__JAFFAR_COMMON__INLINE__ void yield()
{
  if (__runtime == nullptr) JAFFAR_THROW_LOGIC("Trying to use dethreader runtime before it is initialized");
  __runtime->getCurrentThread()->yield();
}
 
__JAFFAR_COMMON__INLINE__ void sleep(const size_t sleepDuration)
{
  if (__runtime == nullptr) JAFFAR_THROW_LOGIC("Trying to use dethreader runtime before it is initialized");
  __runtime->getCurrentThread()->sleep(sleepDuration);
}
 
__JAFFAR_COMMON__INLINE__ void join(const threadId_t threadId)
{
  if (__runtime == nullptr) JAFFAR_THROW_LOGIC("Trying to use dethreader runtime before it is initialized");
  __runtime->getCurrentThread()->join(threadId);
}
 
} // namespace dethreader
 
} // namespace jaffarCommon