fiber.h

Go to the documentation of this file.

#ifndef ZTH_FIBER_H
#define ZTH_FIBER_H
/*
 * SPDX-FileCopyrightText: 2019-2026 Jochem Rutgers
 *
 * SPDX-License-Identifier: MPL-2.0
 */
 
#include <libzth/macros.h>
 
typedef struct {
        void* p;
} zth_fiber_t;
 
#ifdef __cplusplus
#  include <libzth/allocator.h>
#  include <libzth/backtrace.h>
#  include <libzth/config.h>
#  include <libzth/context.h>
#  include <libzth/exception.h>
#  include <libzth/init.h>
#  include <libzth/list.h>
#  include <libzth/perf.h>
#  include <libzth/time.h>
#  include <libzth/util.h>
 
#  include <errno.h>
#  include <exception>
#  include <utility>
 
#  if __cplusplus >= 201103L
#    include <functional>
#  endif
 
namespace zth {
 
// See libzth/worker.h
void outOfWork();
 
class Fiber : public Listable, public UniqueID<Fiber>, public RefCounted {
        ZTH_CLASS_NEW_DELETE(Fiber)
        ZTH_CLASS_NOCOPY(Fiber)
public:
        typedef void(FiberHook)(Fiber&);
 
        static FiberHook* hookNew;
 
        static FiberHook* hookDead;
 
        enum State { Uninitialized = 0, New, Ready, Running, Waiting, Suspended, Cancel, Dead };
 
        typedef ContextAttr::EntryArg EntryArg;
        typedef void (*Entry)(EntryArg);
        static uintptr_t const ExitUnknown = (uintptr_t)-1;
 
        explicit Fiber(Entry entry, EntryArg arg = EntryArg())
                : UniqueID("zth::Fiber")
                , m_state(Uninitialized)
                , m_stateNext(Ready)
                , m_entry(entry)
                , m_entryArg(arg)
                , m_stackSize(Config::DefaultFiberStackSize)
                , m_context()
                , m_fls()
                , m_timeslice(Config::MinTimeslice())
                , m_dtMax(Config::CheckTimesliceOverrun
                                  ? TimeInterval(Config::TimesliceOverrunReportThreshold())
                                  : TimeInterval())
        {
                zth_init();
                setState(New);
                zth_assert(m_entry);
                zth_dbg(fiber, "[%s] New fiber %p", id_str(), normptr());
        }
 
        virtual ~Fiber() noexcept override
        {
                m_exit.once(*this);
                m_cleanup.once(*this);
 
                if(state() > Uninitialized && state() < Dead)
                        kill();
 
                setState(Uninitialized);
                zth_assert(!fls());
                size_t stackSize_ __attribute__((unused)) = this->stackSize();
                size_t stackUsage_ __attribute__((unused)) = this->stackUsage();
                context_destroy(m_context);
                m_context = nullptr;
                zth_dbg(fiber, "[%s] Destructed. Stack usage: 0x%x of 0x%x, total CPU: %s",
                        id_str(), (unsigned int)stackUsage_, (unsigned int)stackSize_,
                        m_totalTime.str().c_str());
        }
 
        zth_fiber_t handle() const noexcept
        {
                // cppcheck-suppress cstyleCast
                zth_fiber_t h = {(void*)this}; // NOLINT
                return h;
        }
 
        static Fiber* fromHandle(zth_fiber_t const& h) noexcept
        {
                return (Fiber*)h.p; // NOLINT
        }
 
        int setStackSize(size_t size) noexcept
        {
                if(state() != New)
                        return EPERM;
 
                m_stackSize = size;
                return 0;
        }
 
        size_t stackSize() const noexcept
        {
                return m_stackSize;
        }
 
        size_t stackUsage() const
        {
                return context_stack_usage(context());
        }
 
        Context* context() const noexcept
        {
                return m_context;
        }
 
        State state() const noexcept
        {
                return m_state;
        }
 
        void* fls() const noexcept
        {
                return m_fls;
        }
 
        void setFls(void* data = nullptr) noexcept
        {
                m_fls = data;
        }
 
        Timestamp const& runningSince() const noexcept
        {
                return m_startRun;
        }
 
        Timestamp const& stateEnd() const noexcept
        {
                return m_stateEnd;
        }
 
        TimeInterval const& totalTime() const noexcept
        {
                return m_totalTime;
        }
 
        typedef Hook<Fiber&> Hook_type;
 
        void atExit(Hook_type::function_type f, Hook_type::arg_type arg = Hook_type::arg_type())
        {
                m_exit.add(f, arg);
        }
 
        void atCleanup(Hook_type::function_type f, Hook_type::arg_type arg = Hook_type::arg_type())
        {
                m_cleanup.add(f, arg);
        }
 
        int init(Timestamp const& now = Timestamp::now())
        {
                if(state() != New)
                        return EPERM;
 
                zth_assert(!m_context);
 
                zth_dbg(fiber, "[%s] Init", id_str());
 
                ContextAttr attr(&fiberEntry, this, m_stackSize);
                int res = context_create(m_context, attr);
                if(res) {
                        // Oops.
                        kill();
                        return res;
                }
 
                setState(m_stateNext, now);
                m_stateNext = Ready;
                m_startRun = now;
 
                return 0;
        }
 
        int run(Fiber& from, Timestamp now = Timestamp::now())
        {
                int res = 0;
 
again:
                switch(state()) {
                case New:
                        // First call, do implicit init.
                        if((res = init(now)))
                                return res;
 
                        if(hookNew)
                                hookNew(*this);
                        goto again;
 
                case Cancel:
                case Ready: {
                        zth_assert(&from != this);
 
                        // Update administration of the current fiber.
                        TimeInterval dt = now - from.m_startRun;
                        from.m_totalTime += dt;
 
                        if(from.state() == Running)
                                from.setState(Ready, now);
 
                        if(unlikely(zth_config(CheckTimesliceOverrun) && from.m_dtMax < dt)) {
                                zth_perf_mark("timeslice overrun reported");
                                from.m_dtMax = dt;
                                log_color(
                                        Config::Print_perf,
                                        ZTH_DBG_PREFIX "Long timeslice by %s of %s\n",
                                        from.id_str(), dt.str().c_str());
                                Backtrace().print(Config::Print_perf);
                                now = Timestamp::now();
                        }
 
                        // Hand over to this.
                        m_startRun = now;
                        if(likely(state() != Cancel))
                                setState(Running, now);
                        m_stateEnd = now + m_timeslice;
 
                        zth_dbg(fiber, "Switch from %s to %s after %s", from.id_str(), id_str(),
                                dt.str().c_str());
                        context_switch(from.context(), context());
 
                        // Ok, got back to the *from* fiber.
                        // Warning! The *this* fiber might already be dead and destructed at this
                        // point!
 
                        if(unlikely(from.state() == Cancel))
                                from.cancelled();
 
                        return 0;
                }
 
                case Running:
                        zth_assert(&from == this);
                        // No switch required.
                        return EAGAIN;
 
                case Dead:
                        // Can't run dead fibers.
                        return EPERM;
 
                case Uninitialized:
                case Waiting:
                case Suspended:
                default:
                        zth_assert(false);
                        return EINVAL;
                }
        }
 
        bool allowYield(Timestamp const& now = Timestamp::now()) const noexcept
        {
                return state() != Running || m_stateEnd < now;
        }
 
        void kill() noexcept
        {
                if(state() == Dead)
                        return;
 
                zth_dbg(fiber, "[%s] Killed", id_str());
                setState(Dead);
        }
 
        void cancel()
        {
                if(state() == Cancel)
                        return;
 
                zth_dbg(fiber, "[%s] Cancel", id_str());
 
                bool running = state() == Running;
                if(state() != Dead)
                        setState(Cancel);
 
                if(running)
                        cancelled();
        }
 
protected:
        void cancelled()
        {
                zth_dbg(fiber, "[%s] Cancelled", id_str());
#  ifdef __cpp_exceptions
                zth_throw(zth::cancelled());
#  else
                kill();
                outOfWork();
#  endif
        }
 
public:
        void nap(Timestamp const& sleepUntil = Timestamp::null()) noexcept
        {
                switch(state()) {
                case New:
                        // Postpone actual sleep
                        zth_dbg(fiber, "[%s] Sleep upon startup", id_str());
                        m_stateNext = Waiting;
                        break;
                case Ready:
                case Running:
                        if(sleepUntil.isNull())
                                zth_dbg(fiber, "[%s] Sleep", id_str());
                        else
                                zth_dbg(fiber, "[%s] Sleep for %s", id_str(),
                                        (sleepUntil - Timestamp::now()).str().c_str());
                        setState(Waiting);
                        m_stateNext = Ready;
                        break;
                case Suspended:
                        zth_dbg(fiber, "[%s] Schedule sleep after resume", id_str());
                        m_stateNext = Waiting;
                        break;
                case Cancel:
                case Waiting:
                case Uninitialized:
                case Dead:
                default:
                        return;
                }
 
                m_stateEnd = sleepUntil;
        }
 
        void wakeup() noexcept
        {
                if(likely(state() == Waiting)) {
                        setState(m_stateNext);
                        switch(state()) {
                        case Suspended:
                                zth_dbg(fiber, "[%s] Suspend after wakeup", id_str());
                                m_stateNext = Ready;
                                break;
                        case Uninitialized:
                        case New:
                        case Ready:
                        case Running:
                        case Waiting:
                        case Cancel:
                        case Dead:
                        default:
                                zth_dbg(fiber, "[%s] Wakeup", id_str());
                        }
                } else if(state() == Suspended && m_stateNext == Waiting) {
                        zth_dbg(fiber, "[%s] Set wakeup after suspend", id_str());
                        m_stateNext = Ready;
                }
        }
 
        void suspend() noexcept
        {
                switch(state()) {
                case New:
                        m_stateNext = New;
                        break;
                case Running:
                case Ready:
                        m_stateNext = Ready;
                        break;
                case Waiting:
                        m_stateNext = Waiting;
                        // fall-through
                case Suspended:
                case Uninitialized:
                case Cancel:
                case Dead:
                default:
                        // Ignore.
                        return;
                }
 
                zth_dbg(fiber, "[%s] Suspend", id_str());
                setState(Suspended);
        }
 
        void resume() noexcept
        {
                if(state() != Suspended)
                        return;
 
                zth_dbg(fiber, "[%s] Resume", id_str());
                setState(m_stateNext);
                if(state() == New)
                        m_stateNext = Ready;
        }
 
        string str() const
        {
                string res = format("%s", id_str());
 
                switch(state()) {
                case Uninitialized:
                        res += " Uninitialized";
                        break;
                case New:
                        res += " New";
                        break;
                case Ready:
                        res += " Ready";
                        break;
                case Running:
                        res += " Running";
                        break;
                case Waiting:
                        res += " Waiting";
                        if(!m_stateEnd.isNull())
                                res +=
                                        format(" (%s remaining)",
                                               Timestamp::now().timeTo(stateEnd()).str().c_str());
                        break;
                case Suspended:
                        res += " Suspended";
                        break;
                case Cancel:
                        res += " Cancel";
                        break;
                case Dead:
                        res += " Dead";
                        break;
                default:;
                }
 
                res += format(" t=%s", m_totalTime.str().c_str());
                return res;
        }
 
private:
        virtual void changedName(string const& name) override
        {
                zth_dbg(fiber, "[%s] Renamed to %s", id_str(), name.c_str());
 
                if(Config::EnablePerfEvent)
                        perf_fiber(*this);
        }
 
protected:
        void setState(State state, Timestamp const& t = Timestamp::now()) noexcept
        {
                if(m_state == state)
                        return;
 
                m_state = state;
 
                if(Config::EnablePerfEvent)
                        perf_fiber_state(*this, m_state, t);
 
                if(state == Dead && hookDead)
                        hookDead(*this);
        }
 
        static void fiberEntry(void* that) noexcept
        {
                zth_assert(that);
                // cppcheck-suppress nullPointerRedundantCheck
                static_cast<Fiber*>(that)->fiberEntry_();
        }
 
        void fiberEntry_() noexcept
        {
                zth_dbg(fiber, "[%s] Entry", id_str());
 
                try {
                        m_entry(m_entryArg);
                } catch(std::exception const& e) {
#  ifdef __cpp_exceptions
                        zth_dbg(fiber, "[%s] Uncaught exception; %s", id_str(), e.what());
#  endif
                } catch(...) {
                        zth_dbg(fiber, "[%s] Uncaught exception", id_str());
                }
 
                zth_dbg(fiber, "[%s] Exit", id_str());
                m_exit.once(*this);
 
                kill();
        }
 
private:
        State m_state;
        State m_stateNext;
        Entry m_entry;
        EntryArg m_entryArg;
        size_t m_stackSize;
        Context* m_context;
        void* m_fls;
        TimeInterval m_totalTime;
        Timestamp m_startRun;
        Timestamp m_stateEnd;
        TimeInterval m_timeslice;
        TimeInterval m_dtMax;
        Hook_type m_exit;
        Hook_type m_cleanup;
};
 
class Runnable {
        ZTH_CLASS_NOCOPY(Runnable)
        ZTH_CLASS_NEW_DELETE(Runnable)
protected:
        constexpr Runnable() noexcept
                : m_fiber()
        {}
 
public:
        virtual ~Runnable() is_default
 
        int run();
 
        Fiber* fiber() const noexcept
        {
                return m_fiber;
        }
 
        operator Fiber&() const noexcept
        {
                // cppcheck-suppress constVariablePointer
                Fiber* const f = fiber();
                zth_assert(f);
                // cppcheck-suppress nullPointerRedundantCheck
                return *f;
        }
 
        char const* id_str() const
        {
                return fiber() ? fiber()->id_str() : "detached Runnable";
        }
 
protected:
        virtual int fiberHook(Fiber& f)
        {
                f.atCleanup(&cleanup_, this);
                m_fiber = &f;
                return 0;
        }
 
        virtual void cleanup() noexcept
        {
                m_fiber = nullptr;
        }
 
        virtual void entry() = 0;
 
private:
        static void cleanup_(Fiber& UNUSED_PAR(f), void* that) noexcept
        {
                Runnable* r = static_cast<Runnable*>(that);
                if(likely(r)) {
                        zth_assert(&f == r->m_fiber);
                        r->cleanup();
                }
        }
 
        static void entry_(void* that)
        {
                if(likely(that))
                        static_cast<Runnable*>(that)->entry();
        }
 
private:
        Fiber* m_fiber;
};
 
__attribute__((pure)) Fiber& currentFiber() noexcept;
 
ZTH_EXPORT inline void* fls() noexcept
{
        return currentFiber().fls();
}
 
ZTH_EXPORT inline void setFls(void* data = nullptr) noexcept
{
        return currentFiber().setFls(data);
}
 
} // namespace zth
 
EXTERN_C ZTH_EXPORT ZTH_INLINE void* zth_fls() noexcept
{
        return zth::fls();
}
 
EXTERN_C ZTH_EXPORT ZTH_INLINE void zth_setFls(void* data = nullptr) noexcept
{
        zth::setFls(data);
}
 
#else // !__cplusplus
 
ZTH_EXPORT void* zth_fls();
ZTH_EXPORT void* zth_setFls(void* data);
 
#endif // __cplusplus
 
EXTERN_C ZTH_EXPORT int main_fiber(int argc, char** argv);
 
#endif // ZTH_FIBER_H