channel.hpp

//SPDX-License-Identifier: LicenseRef-Apache-License-2.0
//Author: Blayne Dennis
 
#ifndef __SIMPLE_THREADING_CHANNEL__
#define __SIMPLE_THREADING_CHANNEL__
 
#include <memory>
#include <mutex>
#include <condition_variable>
#include <deque>
#include <utility>
#include <thread>
#include <future>
#include <chrono>
#include <type_traits>
 
#include "utility.hpp"
#include "context.hpp"
#include "message.hpp"
 
namespace st { // simple thread  
 
enum state {
    closed = 0, 
    failure, 
    success 
};
 
namespace detail {
namespace channel {
 
// private class used to implement `recv()` blocking behavior
struct blocker {
    // stack condition data (not allocated!)
    struct data {
        inline void wait(std::unique_lock<std::mutex>& lk) {
            do {
                cv.wait(lk);
            } while(!flag);
        }
 
        inline void notify() {
            flag = true;
            cv.notify_one(); 
        }
 
        bool flag = false;
        std::condition_variable cv;
    };
 
    blocker() = delete;
    blocker(data* d) : m_data(d) { }
    ~blocker(){ m_data->notify(); }
 
    // a pointer to stack condition data 
    data* m_data;
};
 
struct context {
    context() : m_closed(false) { }
 
    virtual ~context(){ 
        close(true); 
    }
 
    inline bool closed() const { 
        std::lock_guard<std::mutex> lk(m_mtx);
        return m_closed;
    }
    
    inline void close(bool soft=true) {
        std::unique_lock<std::mutex> lk(m_mtx);
        if(!m_closed) {
            m_closed = true;
 
            if(!soft) {
                m_msg_q.clear();
            }
 
            // unblock all receivers if no messages
            if(m_msg_q.empty()) {
                m_blockers.clear(); 
            } 
 
            // unblock excess receivers if messages are available
            while(m_blockers.size() > m_msg_q.size()) {
                m_blockers.pop_front();
            }
        }
    }
    
    inline std::size_t queued() const {
        std::lock_guard<std::mutex> lk(m_mtx);
        return m_msg_q.size();
    }
 
    inline std::size_t blocked_receivers() const {
        std::lock_guard<std::mutex> lk(m_mtx);
        return m_blockers.size();
    }
 
    inline bool send(st::message msg) {
        std::unique_lock<std::mutex> lk(m_mtx);
 
        if(m_closed || !msg) {
            return false;
        } else {
            m_msg_q.push_back(std::move(msg));
 
            while(m_msg_q.size() && m_blockers.size()) {
                m_blockers.pop_front();
            }
            return true;
        } 
    }
 
    inline state recv(st::message& msg, bool block) {
        msg.reset();
 
        std::unique_lock<std::mutex> lk(m_mtx);
 
        do {
            if(!m_msg_q.empty()) {
                msg = std::move(m_msg_q.front());
                m_msg_q.pop_front();
                return st::state::success;
            } else if(m_closed) {
                return st::state::closed;
            } else if(block) {
                // block until message is available or channel termination
                blocker::data d; // stack condition and flag 
                // ~blocker() will notify condition
                m_blockers.push_back(std::shared_ptr<blocker>(new blocker(&d)));
                d.wait(lk);
            } 
        } while(block); 
 
        // failed try_recv()
        return st::state::failure;
    }
 
    bool m_closed;
    mutable std::mutex m_mtx;
    std::deque<st::message> m_msg_q;
    std::deque<std::shared_ptr<blocker>> m_blockers;
};
 
}
}
 
struct channel : public st::shared_context<channel,detail::channel::context> {
    inline virtual ~channel() { }
 
    static inline channel make() {
        channel ch;
        ch.ctx(std::make_shared<detail::channel::context>());
        return ch;
    }
 
    inline bool closed() const {
        return ctx() ? ctx()->closed() : true;
    }
 
    inline void close(bool soft=true) {
        if(ctx()) {
            ctx()->close(soft);
        }
    }
 
    inline std::size_t blocked_receivers() const {
        return ctx() ? ctx()->blocked_receivers() : 0;
    }
 
    inline bool recv(st::message& msg) {
        return ctx() ? ctx()->recv(msg, true) == state::success : false;
    }
 
    inline state try_recv(st::message& msg) {
        return ctx() ? ctx()->recv(msg, false) : state::closed;
    }
   
    inline std::size_t queued() const {
        return ctx() ? ctx()->queued() : 0;
    }
 
    template <typename... As>
    bool send(As&&... as) {
        return ctx() ? ctx()->send(st::message::make(std::forward<As>(as)...)) : false;
    }
    
    //--------------------------------------------------------------------------
    // Iteration 
   
    class iterator : 
        public st::shared_context<iterator, detail::channel::context>, 
        public std::input_iterator_tag 
    {
        inline void increment() {
            bool ret = ctx() ? ctx()->recv(msg, true) == state::success : false;
 
            if(!ret) {
                ctx().reset();
            }
        }
 
        st::message msg;
 
    public:
        iterator() { }
 
        iterator(std::shared_ptr<detail::channel::context> rhs) { 
            ctx(rhs);
        }
        
        inline virtual ~iterator() { }
 
        inline iterator& operator=(const iterator& rhs) {
            ctx() = rhs.ctx();
            msg = rhs.msg;
            return *this;
        }
 
        inline st::message& operator*() { 
            return msg; 
        }
 
        inline st::message* operator->() { 
            return &msg;
        }
 
        inline iterator& operator++() {
            increment();
            return *this;
        }
 
        inline iterator operator++(int) {
            increment();
            return *this;
        }
    };
 
    inline iterator begin() const {
        iterator it(ctx());
        ++it; // get an initial value
        return it;
    }
 
    inline iterator end() const { 
        return iterator(); 
    } 
 
    //--------------------------------------------------------------------------
    // Asynchronous Execution
 
    template <typename F, typename... As>
    bool async(std::size_t resp_id, F&& f, As&&... as) {
        using isv = typename std::is_void<detail::function_return_type<F,As...>>::type;
        return async_impl(
                isv(),
                resp_id,
                std::forward<F>(f), 
                std::forward<As>(as)...);
    }
    
    template< class Rep, class Period, typename P>
    bool timer(std::size_t resp_id, const std::chrono::duration<Rep, Period>& timeout, P&& payload) {
        return async(
                resp_id, 
                [=]() mutable -> decltype(std::forward<P>(payload)) { 
                    std::this_thread::sleep_for(timeout); 
                    return std::forward<P>(payload);
                });
    }
    
    template< class Rep, class Period>
    bool timer(std::size_t resp_id, const std::chrono::duration<Rep, Period>& timeout) {
        return async(
                resp_id, 
                [=]() mutable -> void { 
                    std::this_thread::sleep_for(timeout); 
                });
    }
 
private:
    template <typename F, typename... As>
    bool async_impl(std::true_type, std::size_t resp_id, F&& f, As&&... as) {
        if(ctx()) {
            auto self = ctx();
 
            // launch a thread and schedule the call
            std::async([=]() mutable { 
                 f(std::forward<As>(as)...);
                 // capture a copy of the shared send context
                 self->send(st::message::make(resp_id));
            }); 
 
            return true;
        } else {
            return false;
        }
    }
    
    template <typename F, typename... As>
    bool async_impl(std::false_type, std::size_t resp_id, F&& f, As&&... as) {
        if(ctx()) {
            auto self = ctx();
 
            // launch a thread and schedule the call
            std::async([=]() mutable { 
                 auto result = f(std::forward<As>(as)...);
                 // capture a copy of the shared send context
                 self->send(st::message::make(resp_id, result));
            }); 
 
            return true;
        } else {
            return false;
        }
    }
};
 
}
 
#endif