/**
 * weejobs
 * Copyright 2024 Pelican Mapping
 * https://github.com/pelicanmapping/weejobs
 * MIT License
 */
#pragma once
#include <atomic>
#include <cfloat>
#include <chrono>
#include <condition_variable>
#include <cstdlib>
#include <functional>
#include <list>
#include <mutex>
#include <thread>
#include <type_traits>
#include <vector>
#include <string>
#include <algorithm>

 // OPTIONAL: Define WEEJOBS_EXPORT if you want to use this library from multiple modules (DLLs)
#ifndef WEEJOBS_EXPORT
#define WEEJOBS_EXPORT
#endif

// OPTIONAL: Customize the namespace by defining WEEJOBS_NAMESPACE before including this file.
#ifndef WEEJOBS_NAMESPACE
#define WEEJOBS_NAMESPACE jobs
#endif

// Version
#define WEEJOBS_VERSION_MAJOR 1
#define WEEJOBS_VERSION_MINOR 0
#define WEEJOBS_VERSION_REV   1
#define WEEJOBS_STR_NX(s) #s
#define WEEJOBS_STR(s) WEEJOBS_STR_NX(s)
#define WEEJOBS_COMPUTE_VERSION(major, minor, patch) ((major) * 10000 + (minor) * 100 + (patch))
#define WEEJOBS_VERSION_NUMBER WEEJOBS_COMPUTE_VERSION(WEEJOBS_VERSION_MAJOR, WEEJOBS_VERSION_MINOR, WEEJOBS_VERSION_REV)
#define WEEJOBS_VERSION_STRING WEEJOBS_STR(WEEJOBS_VERSION_MAJOR) "." WEEJOBS_STR(WEEJOBS_VERSION_MINOR) "." WEEJOBS_STR(WEEJOBS_VERSION_REV)

#if __cplusplus >= 201703L || _MSVC_LANG >= 201703L
#define WEEJOBS_NO_DISCARD [[nodiscard]]
#else
#define WEEJOBS_NO_DISCARD
#endif

/**
* weejobs is an API for scheduling a task to run in the background.
* Please read the README.md file for more information.
*/
namespace WEEJOBS_NAMESPACE
{
    /**
    * Interface for something that can be canceled
    */
    class cancelable
    {
    public:
        virtual bool canceled() const { return false; }
    };

    namespace detail
    {
        /**
         * Event with a binary signaled state, for multi-threaded sychronization.
         *
         * The event has two states:
         *  "set" means that a call to wait() will not block;
         *  "unset" means that calls to wait() will block until another thread calls set().
         *
         * The event starts out unset.
         *
         * Typical usage: Thread A creates Thread B to run asynchronous code. Thread A
         * then calls wait(), which blocks Thread A. When Thread B is finished, it calls
         * set(). Thread A then wakes up and continues execution.
         *
         * NOTE: ALL waiting threads will wake up when the Event is cleared.
         */
        struct event
        {
        public:
            //! Construct a new event
            event() : _set(false) { }

            //! DTOR
            ~event() {
                _set = false;
                for (int i = 0; i < 255; ++i)  // workaround buggy broadcast
                    _cond.notify_all();
            }

            //! Block until the event is set, then return true.
            inline bool wait() {
                while (!_set) {
                    std::unique_lock<std::mutex> lock(_m);
                    if (!_set)
                        _cond.wait(lock);
                }
                return _set;
            }

            //! Block until the event is set or the timout expires.
            //! Return true if the event has set, otherwise false.
            template<typename T>
            inline bool wait(T timeout) {
                if (!_set) {
                    std::unique_lock<std::mutex> lock(_m);
                    if (!_set)
                        _cond.wait_for(lock, timeout);
                }
                return _set;
            }

            //! Block until the event is set; then reset it.
            inline bool waitAndReset() {
                std::unique_lock<std::mutex> lock(_m);
                if (!_set)
                    _cond.wait(lock);
                _set = false;
                return true;
            }

            //! Set the event state, causing any waiters to unblock.
            inline void set() {
                if (!_set) {
                    std::unique_lock<std::mutex> lock(_m);
                    if (!_set) {
                        _set = true;
                        _cond.notify_all();
                    }
                }
            }

            //! Reset (unset) the event state; new waiters will block until set() is called.
            inline void reset() {
                std::unique_lock<std::mutex> lock(_m);
                _set = false;
            }

            //! Whether the event state is set (waiters will not block).
            inline bool isSet() const {
                return _set;
            }

        protected:
            bool _set;
            std::condition_variable_any _cond;
            std::mutex _m; // do not use Mutex, we never want tracking
        };


        /**
         * Sempahore lets N users aquire it and then notifies when the
         * count goes back down to zero.
         */
        class semaphore
        {
        public:
            //! Acquire, increasing the usage count by one
            void acquire()
            {
                std::unique_lock<std::mutex> lock(_m);
                ++_count;
            }

            //! Release, decreasing the usage count by one.
            //! When the count reaches zero, joiners will be notified and
            //! the semaphore will reset to its initial state.
            void release()
            {
                std::unique_lock<std::mutex> lock(_m);
                _count = std::max(_count - 1, 0);
                if (_count == 0)
                    _cv.notify_all();
            }

            //! Reset to initialize state; this will cause a join to occur
            //! even if no acquisitions have taken place.
            void reset()
            {
                std::unique_lock<std::mutex> lock(_m);
                _count = 0;
                _cv.notify_all();
            }

            //! Current count in the semaphore
            std::size_t count() const
            {
                std::unique_lock<std::mutex> lock(_m);
                return _count;
            }

            //! Block until the semaphore count returns to zero.
            //! (It must first have left zero)
            //! Warning: this method will block forever if the count
            //! never reaches zero!
            void join()
            {
                std::unique_lock<std::mutex> lock(_m);
                while (_count > 0)
                    _cv.wait(lock);
            }

            //! Block until the semaphore count returns to zero, or
            //! the operation is canceled.
            //! (It must first have left zero)
            void join(cancelable* c)
            {
                _cv.wait_for(_m, std::chrono::seconds(1), [this, c]() {
                    return
                        (_count == 0) ||
                        (c && c->canceled());
                    }
                );
                _count = 0;
            }

        private:
            int _count = 0;
            std::condition_variable_any _cv;
            mutable std::mutex _m;
        };

#if __cplusplus >= 201703L || _MSVC_LANG >= 201703L
        template<typename F, typename...Args>
        using result_of_t = typename std::invoke_result<F, Args...>::type;
#else
        template<typename F, typename...Args>
        using result_of_t = typename std::result_of<F(Args...)>::type;
#endif
    }

    /**
    * Include a jobgroup in a context to group together multiple jobs.
    * You can then call jobgroup::join() to wait for the whole group
    * to finish.
    */
    struct jobgroup : public detail::semaphore
    {
        static std::shared_ptr<jobgroup> create()
        {
            return std::make_shared<jobgroup>();
        }
    };

    /**
    * Context object you can pass to dispatch(...) to control aspects of
    * how the background task is run.
    */
    struct context
    {
        std::string name; // readable name of the job
        class jobpool* pool = nullptr; // job pool to run in
        std::function<float()> priority = {}; // priority of the job
        std::shared_ptr<jobgroup> group = nullptr; // join group for this job
        bool can_cancel = true; // if true, the job will cancel if its future goes out of scope
    };

    /**
     * Future holds the future result of an asynchronous operation.
     *
     * Usage:
     *   Producer (usually an asynchronous function call) creates a future<T>
     *   (the promise of a future result) and immediately returns it. The Consumer
     *   then performs other work, and eventually (or immediately) checks available()
     *   for a result or canceled() for cancelation. If availabile() is true,
     *   Consumer calls value() to fetch the valid result.
     *
     *   As long as at least two equivalent Future object (i.e. Futures pointing to the
     *   same internal shared data) exist, the Future is considered valid. Once
     *   that count goes to one, the Future is either available (the value is ready)
     *   or empty (i.e., canceled or abandoned).
     */
    template<typename T>
    class future : public cancelable
    {
    private:
        // internal structure to track references to the result
        // One instance of this is shared among all Future instances
        // created from the copy constructor.
        struct shared_t
        {
            T _obj;
            mutable detail::event _ev;
            std::mutex _continuation_mutex;
            std::function<void()> _continuation;
            std::atomic_bool _continuation_ran = { false };
        };

    public:
        //! Default constructor
        future()
        {
            _shared = std::make_shared<shared_t>();
        }

        //! Default copy constructor
        future(const future& rhs) = default;

        //! True is this Future is unused and not connected to any other Future
        bool empty() const
        {
            return !available() && _shared.use_count() == 1;
        }

        //! True if the promise was resolved and a result if available.
        bool available() const
        {
            return _shared->_ev.isSet();
        }

        //! True if a promise exists, but has not yet been resolved;
        //! Presumably the asynchronous task is still working.
        bool working() const
        {
            return !empty() && !available();
        }

        // cancelable interface
        bool canceled() const override
        {
            return empty();
        }

        //! Deference the result object. Make sure you check available()
        //! to check that the future was actually resolved; otherwise you
        //! will just get the default object.
        const T& value() const
        {
            return _shared->_obj;
        }

        //! Dereference this object to const pointer to the result.
        const T* operator -> () const
        {
            return &_shared->_obj;
        }

        //! Result is available AND equal to the argument.
        bool has_value(const T& arg) const
        {
            return available() && value() == arg;
        }

        //! Same as value(), but if the result is available will reset the
        //! future before returning the result object.
        T release()
        {
            bool avail = available();
            T result = value();
            if (avail)
                reset();
            return result;
        }

        //! Blocks until the result becomes available or the future is abandoned;
        //! then returns the result object.
        const T& join() const
        {
            while (
                !empty() &&
                !_shared->_ev.wait(std::chrono::milliseconds(1)));
            return value();
        }

        //! Blocks until the result becomes available or the future is abandoned
        //! or a cancelation flag is set; then returns the result object. Be sure to
        //! check canceled() after calling join() to see if the return value is valid.
        const T& join(const cancelable* p) const
        {
            while (working() && (p == nullptr || !p->canceled()))
            {
                _shared->_ev.wait(std::chrono::milliseconds(1));
            }
            return value();
        }

        //! Blocks until the result becomes available or the future is abandoned
        //! or a cancelation flag is set; then returns the result object. Be sure to
        //! check canceled() after calling join() to see if the return value is valid.
        const T& join(const cancelable& p) const
        {
            return join(&p);
        }

        //! Release reference to a promise, resetting this future to its default state
        void abandon()
        {
            _shared.reset(new shared_t());
        }

        //! synonym for abandon.
        void reset()
        {
            abandon();
        }

        //! Resolve (fulfill) the promise with the provided result value.
        void resolve(const T& value)
        {
            _shared->_obj = value;
            _shared->_ev.set();
            fire_continuation();
        }

        //! Resolve (fulfill) the promise with an rvalue
        void resolve(T&& value)
        {
            _shared->_obj = std::move(value);
            _shared->_ev.set();
            fire_continuation();
        }

        //! Resolve (fulfill) the promise with a default result
        void resolve()
        {
            _shared->_ev.set();
            fire_continuation();
        }

        //! The number of objects, including this one, that
        //! reference the shared container. If this method
        //! returns 1, that means this is the only object with
        //! access to the data. This method will never return zero.
        unsigned refs() const
        {
            return _shared.use_count();
        }

        //! Add a continuation to this future. The continuation will be dispatched
        //! when this object's result becomes available; that result will be the input
        //! value to the continuation function. The continuation function in turn must
        //! return a value (cannot be void).
        template<typename F, typename R = typename detail::result_of_t<F, const T&, cancelable&>>
        WEEJOBS_NO_DISCARD inline future<R> then_dispatch(F func, const context& con = {});

        //! Add a continuation to this future. Instead of the functor returning a value,
        //! it will instead have the option of resolving the incoming future/promise object.
        //! This is useful for operations that have their own way of running asynchronous code.
        //! Note: for some reason when you use this variant you must specific the template
        //! argument, e.g. result.then<int>(auto i, promise<int> p)
        template<typename R>
        WEEJOBS_NO_DISCARD inline future<R> then_dispatch(std::function<void(const T&, future<R>&)> func, const context& con = {});

        //! Add a continuation to this future. The functor only takes an input value and has no
        //! return value (fire and forget).
        inline void then_dispatch(std::function<void(const T&)> func, const context& con = {});

    private:
        std::shared_ptr<shared_t> _shared;

        void fire_continuation()
        {
            std::lock_guard<std::mutex> lock(_shared->_continuation_mutex);

            if (_shared->_continuation && !_shared->_continuation_ran.exchange(true))
                _shared->_continuation();

            // Zero out the continuation function immediately after running it.
            // This is important because the continuation might hold a reference to a promise
            // that might hamper cancelation.
            _shared->_continuation = nullptr;
        }
    };

    //! in the "promise/future" pattern, we use the same object for both,
    //! but here's an alias for clarity.
    template<class T> using promise = future<T>;

    namespace detail
    {
        struct job
        {
            context ctx;
            std::function<bool()> _delegate;

            bool operator < (const job& rhs) const
            {
                float lp = ctx.priority ? ctx.priority() : -FLT_MAX;
                float rp = rhs.ctx.priority ? rhs.ctx.priority() : -FLT_MAX;
                return lp < rp;
            }
        };

        inline bool steal_job(class jobpool* thief, detail::job& stolen);
    }

    /**
    * A priority-sorted collection of jobs that are running or waiting
    * to run in a thread pool.
    */
    class jobpool
    {
    public:
        /**
        * Metrics of a thread pool.
        */
        struct metrics_t
        {
            std::string name;
            std::atomic_uint concurrency = { 0u };
            std::atomic_uint pending = { 0u };
            std::atomic_uint running = { 0u };
            std::atomic_uint postprocessing = { 0u };
            std::atomic_uint canceled = { 0u };
            std::atomic_uint total = { 0u };
        };

    public:
        //! Destroy
        ~jobpool()
        {
            stop_threads();
        }

        //! Name of this job pool
        const std::string& name() const
        {
            return _metrics.name;
        }

        metrics_t* metrics()
        {
            return &_metrics;
        }

        //! Set the concurrency of this job scheduler
        void set_concurrency(unsigned value)
        {
            value = std::max(value, 1u);
            if (_target_concurrency != value)
            {
                _target_concurrency = value;
                start_threads();
            }
        }

        //! Get the target concurrency (thread count) 
        unsigned concurrency() const
        {
            return _target_concurrency;
        }

        //! Whether this job pool is allowed to steal work from other job pools
        //! when it is idle. Default = true.
        void set_can_steal_work(bool value)
        {
            _can_steal_work = value;
        }

        //! Discard all queued jobs
        void cancel_all()
        {
            std::lock_guard<std::mutex> lock(_queue_mutex);
            _queue.clear();
            _queue_size = 0;
            _metrics.canceled += _metrics.pending;
            _metrics.pending = 0;
        }

        //! Schedule an asynchronous task on this scheduler
        //! Use job::dispatch to run jobs (usually no need to call this directly)
        //! @param delegate Function to execute
        //! @param context Job details
        void _dispatch_delegate(std::function<bool()>& delegate, const context& context)
        {
            if (!_done)
            {
                // If we have a group semaphore, acquire it BEFORE queuing the job
                if (context.group)
                {
                    context.group->acquire();
                }

                if (_target_concurrency > 0)
                {
                    std::lock_guard<std::mutex> lock(_queue_mutex);

                    _queue.emplace_back(detail::job{ context, delegate });
                    _queue_size++;

                    _metrics.pending++;
                    _metrics.total++;
                    _block.notify_one();
                }
                else
                {
                    // no threads? run synchronously.
                    delegate();

                    if (context.group)
                    {
                        context.group->release();
                    }
                }
            }
        }

        //! removes the highest priority job from the queue and places it
        //! in output. Returns true if a job was taken, false if the queue
        //! was empty.
        inline bool _take_job(detail::job& output, bool lock)
        {
            if (lock)
            {
                std::lock_guard<std::mutex> lock(_queue_mutex);
                return _take_job(output, false);
            }
            else if (!_done && _queue_size > 0)
            {
                auto ptr = _queue.end();
                float highest_priority = -FLT_MAX;
                for (auto iter = _queue.begin(); iter != _queue.end(); ++iter)
                {
                    float priority = iter->ctx.priority != nullptr ?
                        iter->ctx.priority() :
                        0.0f;

                    if (ptr == _queue.end() || priority > highest_priority)
                    {
                        ptr = iter;
                        highest_priority = priority;
                    }
                }

                if (ptr == _queue.end())
                    ptr = _queue.begin();

                output = std::move(*ptr);
                _queue.erase(ptr);
                _queue_size--;
                _metrics.pending--;
                return true;
            }
            return false;
        }

        //! Construct a new job pool.
        //! Do not call this directly - call getPool(name) instead.
        jobpool(const std::string& name, unsigned concurrency) :
            _target_concurrency(concurrency)
        {
            _metrics.name = name;
            _metrics.concurrency = 0;
        }

        //! Pulls queued jobs and runs them in whatever thread run() is called from.
        //! Runs in a loop until _done is set.
        inline void run();

        //! Spawn all threads in this scheduler
        inline void start_threads();

        //! Signall all threads to stop
        inline void stop_threads();

        //! Wait for all threads to exit (after calling stop_threads)
        inline void join_threads();

        bool _can_steal_work = true;
        std::list<detail::job> _queue;
        std::atomic_int _queue_size = { 0 }; // don't use list::size(), it's slow and not atomic
        mutable std::mutex _queue_mutex; // protect access to the queue
        mutable std::mutex _quit_mutex; // protects access to _done
        std::atomic<unsigned> _target_concurrency; // target number of concurrent threads in the pool
        std::condition_variable_any _block; // thread waiter block
        bool _done = false; // set to true when threads should exit
        std::vector<std::thread> _threads; // threads in the pool
        metrics_t _metrics; // metrics for this pool
    };

    class metrics
    {
    public:
        //! Total number of pending jobs across all schedulers
        int total_pending() const;

        //! Total number of running jobs across all schedulers
        int total_running() const;

        //! Total number of running jobs across all schedulers
        int total_postprocessing() const;

        //! Total number of canceled jobs across all schedulers
        int total_canceled() const;

        //! Total number of active jobs in the system
        int total() const;

        //! Gets a vector of all jobpool metrics structures.
        inline const std::vector<struct jobpool::metrics_t*> all()
        {
            return _pools;
        }

        std::vector<struct jobpool::metrics_t*> _pools;
    };

    /**
    * Runtime singleton object;
    * Declare with WEEJOBS_INSTANCE in one of your .cpp files.
    */
    namespace detail
    {
        struct runtime
        {
            inline runtime();
            inline ~runtime();
            inline void shutdown();

            bool _alive = true;
            bool _stealing_allowed = false;
            std::mutex _pools_mutex;
            std::vector<jobpool*> _pools;
            metrics _metrics;
            std::function<void(const char*)> _set_thread_name;
        };
    }

    //! Access to the runtime singleton - users need not call this
    extern WEEJOBS_EXPORT detail::runtime& instance();

    //! Returns the job pool with the given name, creating a new one if it doesn't 
    //! already exist. If you don't specify a name, a default pool is used.
    inline jobpool* get_pool(const std::string& name = {})
    {
        std::lock_guard<std::mutex> lock(instance()._pools_mutex);

        for (auto pool : instance()._pools)
        {
            if (pool->name() == name)
                return pool;
        }
        auto new_pool = new jobpool(name, 2u);
        instance()._pools.push_back(new_pool);
        instance()._metrics._pools.push_back(&new_pool->_metrics);
        new_pool->start_threads();
        return new_pool;
    }

    namespace detail
    {
        // dispatches a function to the appropriate job pool.
        inline void pool_dispatch(std::function<bool()> delegate, const context& context)
        {
            auto pool = context.pool ? context.pool : get_pool({});
            if (pool)
            {
                pool->_dispatch_delegate(delegate, context);

                // if work stealing is enabled, wake up all pools
                if (instance()._stealing_allowed)
                {
                    std::lock_guard<std::mutex> lock(instance()._pools_mutex);

                    for (auto pool : instance()._pools)
                    {
                        pool->_block.notify_all();
                    }
                }
            }
        }
    }

    //! Dispatches a job with no return value. Fire and forget.
    //! @param task Function to run in a thread. Prototype is void(void).
    //! @param context Optional configuration for the asynchronous function call
    inline void dispatch(std::function<void()> task, const context& context = {})
    {
        auto delegate = [task]() mutable -> bool { task(); return true; };
        detail::pool_dispatch(delegate, context);
    }

    //! Dispatches a job and immediately returns a future result.
    //! @param task Function to run in a thread. Prototype is T(cancelable&)
    //! @param context Optional configuration for the asynchronous function call
    //! @return Future result of the async function call
    template<typename F, typename T = typename detail::result_of_t<F, cancelable&>>
    WEEJOBS_NO_DISCARD inline future<T> dispatch(F task, const context& context = {})
    {
        future<T> promise;
        bool can_cancel = context.can_cancel;

        std::function<bool()> delegate = [task, promise, can_cancel]() mutable
            {
                bool good = true;
                if (can_cancel)
                {
                    good = !promise.canceled();
                    if (good)
                        promise.resolve(task(promise));
                }
                else
                {
                    cancelable dummy;
                    promise.resolve(task(dummy));
                }
                return good;
            };

        detail::pool_dispatch(delegate, context);

        return promise;
    }

    //! Dispatches a job and immediately returns a future result.
    //! @param task Function to run in a thread. Prototype is T(cancelable&)
    //! @param promise Optional user-supplied promise object
    //! @param context Optional configuration for the asynchronous function call
    //! @return Future result of the async function call
    template<typename F, typename T = typename detail::result_of_t<F, cancelable&>>
    WEEJOBS_NO_DISCARD inline future<T> dispatch(F task, future<T> promise, const context& context = {})
    {
        bool can_cancel = context.can_cancel;

        std::function<bool()> delegate = [task, promise, can_cancel]() mutable
            {
                bool run = !can_cancel || !promise.canceled();
                if (run)
                {
                    task(promise);
                }
                return run;
            };

        detail::pool_dispatch(delegate, context);

        return promise;
    }

    //! Metrics for all job pool
    inline metrics* get_metrics()
    {
        return &instance()._metrics;
    }

    //! stop all threads, wait for them to exit, and shut down the system
    inline void shutdown()
    {
        instance().shutdown();
    }

    //! Whether the weejobs runtime is still alive (has not been shutdown)
    inline bool alive()
    {
        return instance()._alive;
    }

    //! Install a function that the SDK can use to set job pool thread names
    //! when it spawns them.
    inline void set_thread_name_function(std::function<void(const char*)> f)
    {
        instance()._set_thread_name = f;
    }

    //! Whether to allow jobpools to steal work from other jobpools when they are idle.
    inline void set_allow_work_stealing(bool value)
    {
        instance()._stealing_allowed = value;
    }

    inline detail::runtime::runtime()
    {
        //nop
    }

    inline detail::runtime::~runtime()
    {
        shutdown();
    }

    inline void detail::runtime::shutdown()
    {
        _alive = false;

        //std::cout << "stopping " << _pools.size() << " threads..." << std::endl;
        for (auto& pool : _pools)
            if (pool)
                pool->stop_threads();

        //std::cout << "joining " << _pools.size() << " threads..." << std::endl;
        for (auto& pool : _pools)
            if (pool)
                pool->join_threads();
    }

    inline void jobpool::run()
    {
        while (!_done)
        {
            detail::job next;
            bool have_next = false;
            {
                if (_can_steal_work && instance()._stealing_allowed)
                {
                    {
                        std::unique_lock<std::mutex> lock(_queue_mutex);

                        // work-stealing enabled: wait until any queue is non-empty
                        _block.wait(lock, [this]() { return get_metrics()->total_pending() > 0 || _done; });

                        if (!_done && _queue_size > 0)
                        {
                            have_next = _take_job(next, false);
                        }
                    }

                    if (!_done && !have_next)
                    {
                        have_next = detail::steal_job(this, next);
                    }
                }
                else
                {
                    std::unique_lock<std::mutex> lock(_queue_mutex);

                    // wait until just our local queue is non-empty
                    _block.wait(lock, [this] { return (_queue_size > 0) || _done; });

                    if (!_done && _queue_size > 0)
                    {
                        have_next = _take_job(next, false);
                    }
                }
            }

            if (have_next)
            {
                _metrics.running++;

                auto t0 = std::chrono::steady_clock::now();

                bool job_executed = next._delegate();

                auto duration = std::chrono::steady_clock::now() - t0;

                if (job_executed == false)
                {
                    _metrics.canceled++;
                }

                // release the group semaphore if necessary
                if (next.ctx.group != nullptr)
                {
                    next.ctx.group->release();
                }

                _metrics.running--;
            }

            // See if we no longer need this thread because the
            // target concurrency has been reduced
            std::lock_guard<std::mutex> lock(_quit_mutex);

            if (_target_concurrency < _metrics.concurrency)
            {
                _metrics.concurrency--;
                break;
            }
        }
    }

    inline void jobpool::start_threads()
    {
        _done = false;

        // Not enough? Start up more
        while (_metrics.concurrency < _target_concurrency)
        {
            _metrics.concurrency++;

            _threads.push_back(std::thread([this]
                {
                    if (instance()._set_thread_name)
                    {
                        instance()._set_thread_name(_metrics.name.c_str());
                    }
                    run();
                }
            ));
        }
    }

    inline void jobpool::stop_threads()
    {
        _done = true;

        // Clear out the queue
        std::lock_guard<std::mutex> lock(_queue_mutex);

        // reset any group semaphores so that JobGroup.join()
        // will not deadlock.
        for (auto& queuedjob : _queue)
        {
            if (queuedjob.ctx.group != nullptr)
            {
                queuedjob.ctx.group->release();
            }
        }
        _queue.clear();
        _queue_size = 0;

        // wake up all threads so they can exit
        _block.notify_all();
    }

    //! Wait for all threads to exit (after calling stop_threads)
    inline void jobpool::join_threads()
    {
        // wait for them to exit
        for (unsigned i = 0; i < _threads.size(); ++i)
        {
            if (_threads[i].joinable())
            {
                _threads[i].join();
            }
        }

        _threads.clear();
    }

    // steal a job from another jobpool's queue (other than "thief").
    inline bool detail::steal_job(jobpool* thief, detail::job& stolen)
    {
        jobpool* pool_with_most_jobs = nullptr;
        {
            std::lock_guard<std::mutex> lock(instance()._pools_mutex);

            std::size_t max_num_jobs = 0u;
            for (auto pool : instance()._pools)
            {
                if (pool != thief)
                {
                    if (static_cast<std::size_t>(pool->_queue_size) > max_num_jobs)
                    {
                        max_num_jobs = pool->_queue_size;
                        pool_with_most_jobs = pool;
                    }
                }
            }
        }

        if (pool_with_most_jobs)
        {
            return pool_with_most_jobs->_take_job(stolen, true);
        }

        return false;
    }

    template<typename T>
    template<typename F, typename R>
    inline future<R> future<T>::then_dispatch(F func, const context& con)
    {
        // The future result of F. 
        // In this case, the continuation task will return a value that the system will use to resolve the promise.
        future<R> continuation_promise;

        // lock the continuation and set it:
        {
            std::lock_guard<std::mutex> lock(_shared->_continuation_mutex);

            if (_shared->_continuation)
            {
                return {}; // only one continuation allowed
            }

            // take a weak ptr to this future's shared data. If this future goes away we'll still
            // have access to its result.
            std::weak_ptr<shared_t> weak_shared = _shared;

            context copy_of_con = con;

            _shared->_continuation = [func, copy_of_con, weak_shared, continuation_promise]() mutable
                {
                    auto shared = weak_shared.lock();

                    // verify the parent's result is actually available (simulate available())
                    if (shared && shared->_ev.isSet())
                    {
                        // copy it and dispatch it as the input to a new job:
                        T copy_of_value = shared->_obj;

                        // Once this wrapper gets created, note that we now have 2 refereces to the continuation_promise.
                        // To prevent this from hampering cancelation, the continuation fuction is set to nullptr
                        // immediately after being called.
                        auto wrapper = [func, copy_of_value, continuation_promise]() mutable
                            {
                                continuation_promise.resolve(func(copy_of_value, continuation_promise));
                            };

                        jobs::dispatch(wrapper, copy_of_con);
                    }
                };
        }

        // maybe the future is already available?
        if (available())
        {
            fire_continuation();
        }

        return continuation_promise;
    }

    template<typename T>
    template<typename R>
    inline future<R> future<T>::then_dispatch(std::function<void(const T&, future<R>&)> func, const context& con)
    {
        // The future we will return to the caller.
        // Note, the user function "func" is responsible for resolving this promise.
        future<R> continuation_promise;

        // lock the continuation and set it:
        {
            std::lock_guard<std::mutex> lock(_shared->_continuation_mutex);

            if (_shared->_continuation)
            {
                return {}; // only one continuation allowed
            }

            // take a weak ptr to this future's shared data. If this future goes away we'll still
            // have access to its result.
            std::weak_ptr<shared_t> weak_shared = _shared;

            // The user task is responsible for resolving the promise.
            // This continuation executes the user function directly instead of dispatching it
            // to the job pool. This is because we expect the user function to use some external
            // asynchronous mechanism to resolve the promise.
            _shared->_continuation = [func, weak_shared, continuation_promise]() mutable
                {
                    auto shared = weak_shared.lock();
                    if (shared)
                    {
                        func(shared->_obj, continuation_promise);
                    }
                };
        }

        if (available())
        {
            fire_continuation();
        }

        return continuation_promise;
    }

    template<typename T>
    inline void future<T>::then_dispatch(std::function<void(const T&)> func, const context& con)
    {
        // lock the continuation and set it:
        {
            std::lock_guard<std::mutex> lock(_shared->_continuation_mutex);

            if (_shared->_continuation)
            {
                return; // only one continuation allowed
            }

            // take a weak ptr to this future's shared data. If this future goes away we'll still
            // have access to its result.
            std::weak_ptr<shared_t> weak_shared = _shared;
            auto copy_of_con = con;

            _shared->_continuation = [func, weak_shared, copy_of_con]() mutable
                {
                    auto shared = weak_shared.lock();
                    if (shared)
                    {
                        auto copy_of_value = shared->_obj;
                        auto fire_and_forget_delegate = [func, copy_of_value]() mutable
                            {
                                func(copy_of_value);
                                return true;
                            };

                        detail::pool_dispatch(fire_and_forget_delegate, copy_of_con);
                    }
                };
        }

        if (available())
        {
            fire_continuation();
        }
    }

    //! Total number of pending jobs across all schedulers
    inline int metrics::total_pending() const
    {
        std::lock_guard<std::mutex> lock(instance()._pools_mutex);
        int count = 0;
        for (auto pool : _pools)
            count += pool->pending;
        return count;
    }

    //! Total number of running jobs across all schedulers
    inline int metrics::total_running() const
    {
        std::lock_guard<std::mutex> lock(instance()._pools_mutex);
        int count = 0;
        for (auto pool : _pools)
            count += pool->running;
        return count;
    }

    //! Total number of running jobs across all schedulers
    inline int metrics::total_postprocessing() const
    {
        std::lock_guard<std::mutex> lock(instance()._pools_mutex);
        int count = 0;
        for (auto pool : _pools)
            count += pool->postprocessing;
        return count;
    }

    //! Total number of canceled jobs across all schedulers
    inline int metrics::total_canceled() const
    {
        std::lock_guard<std::mutex> lock(instance()._pools_mutex);
        int count = 0;
        for (auto pool : _pools)
            count += pool->canceled;
        return count;
    }

    //! Total number of active jobs in the system
    inline int metrics::total() const
    {
        std::lock_guard<std::mutex> lock(instance()._pools_mutex);
        int count = 0;
        for (auto pool : _pools)
            count += pool->pending + pool->running + pool->postprocessing;
        return count;
    }

    // Use this macro ONCE in your application in a .cpp file to 
    // instaniate the weejobs runtime singleton.
#define WEEJOBS_INSTANCE \
    namespace WEEJOBS_NAMESPACE { \
        static detail::runtime runtime_singleton_instance; \
        detail::runtime& instance() { return runtime_singleton_instance; } \
    }
}