request_container.h

Go to the documentation of this file.

// Copyright (c) 2018, Smart Projects Holdings Ltd
// All rights reserved.
// See LICENSE file for license details.

#ifndef _UGCS_VSM_REQUEST_CONTAINER_H_
#define _UGCS_VSM_REQUEST_CONTAINER_H_

#include <ugcs/vsm/callback.h>
#include <ugcs/vsm/utils.h>

#include <memory>
#include <mutex>
#include <chrono>
#include <condition_variable>
#include <atomic>
#include <list>

namespace ugcs {
namespace vsm {

class Request_container: public std::enable_shared_from_this<Request_container> {
    DEFINE_COMMON_CLASS(Request_container, Request_container)

public:
    class Request: public std::enable_shared_from_this<Request> {
        DEFINE_COMMON_CLASS(Request, Request)

    public:
        enum class Status {
            PENDING,
            CANCELLATION_PENDING,
            CANCELING,
            PROCESSING,
            ABORT_PENDING,
            ABORTED,

            RESULT_CODES,

            OK = RESULT_CODES,
            CANCELED
        };

        typedef Callback_base<void>::Ptr<> Handler;
        typedef std::unique_lock<std::mutex> Locker;

        virtual
        ~Request();

        void
        Set_processing_handler(const Handler &handler);

        void
        Set_processing_handler(Handler &&handler);

        void
        Set_completion_handler(const Request_container::Ptr &context,
                               const Handler &handler);

        void
        Set_completion_handler(const Request_container::Ptr &context,
                               Handler &&handler);

        void
        Set_cancellation_handler(const Handler &handler);

        void
        Set_cancellation_handler(Handler &&handler);

        void
        Set_done_handler(Handler &handler);

        void
        Set_done_handler(Handler &&handler);

        Request_container::Ptr
        Get_completion_context(Locker locker = Locker()) const;

        void
        Process(bool process_request);

        void
        Complete(Status status = Status::OK, Locker locker = Locker());

        void
        Cancel(Locker locker = Locker());

        void
        Abort(Locker locker = Locker());

        Locker
        Lock(bool acquire = true) const;

        Status
        Get_status() const
        {
            return status;
        }

        bool
        Is_completed() const
        {
            return status >= Status::RESULT_CODES;
        }

        bool
        Is_request_processing_needed() const
        {
            /* Abort pending is only for completion contexts. */
            return !Is_completed() && Get_status() != Status::ABORT_PENDING;
        }

        bool
        Is_aborted() const
        {
            return status == Status::ABORTED || status == Status::ABORT_PENDING;
        }

        bool
        Is_processing() const
        {
            return status == Status::PROCESSING;
        }

        bool
        Is_completion_delivered() const
        {
            return completion_delivered;
        }

        bool
        Is_completion_delivering_started() const
        {
            return completion_processed && !completion_handler;
        }

        bool
        Is_done() const
        {
            return Is_completion_delivered() || Is_aborted();
        }

        bool
        Wait_done(bool process_ctx = true,
                  std::chrono::milliseconds timeout = std::chrono::milliseconds::zero());

        bool&
        Timed_out()
        {
            return timed_out;
        }

    protected:
        virtual void
        Destroy();

        bool
        Is_completion_handler_present();

    private:
        Handler processing_handler;
        Handler completion_handler;
        Handler cancellation_handler;
        Handler done_handler;
        Request_container::Ptr completion_context;
        bool timed_out = false;
        std::atomic<Status> status = { Status::PENDING };
        std::atomic_bool completion_processed = { false },
                         completion_delivered = { false };
        mutable std::mutex mutex;
        std::condition_variable cond_var;
    };

    class Request_waiter: public std::enable_shared_from_this<Request_waiter> {
        DEFINE_COMMON_CLASS(Request_waiter, Request_waiter)

    public:
        typedef Callback_base<bool>::Ptr<> Predicate;

        class Locker;

        virtual void
        Notify();

        Locker
        Lock();

        Locker
        Lock_notify();

        int
        Wait_and_process(const std::initializer_list<Request_container::Ptr> &containers,
                         std::chrono::milliseconds timeout = std::chrono::milliseconds::zero(),
                         int requests_limit = 0, Predicate predicate = Predicate());

        int
        Wait_and_process(const std::list<Request_container::Ptr> &containers,
                         std::chrono::milliseconds timeout = std::chrono::milliseconds::zero(),
                         int requests_limit = 0, Predicate predicate = Predicate());

        virtual
        ~Request_waiter() = default;

    private:
        friend class Locker;

        std::mutex mutex;
        std::condition_variable cond_var;

        template <class Container_list>
        int
        Wait_and_process_impl(const Container_list &containers,
                              std::chrono::milliseconds timeout,
                              int requests_limit, Predicate ext_predicate);
    };

    enum class Type {
        NONE                            = 0x0,
        PROCESSOR                       = 0x1,
        COMPLETION_CONTEXT              = 0x2,
        ANY                             = PROCESSOR | COMPLETION_CONTEXT,
        TEMPORAL                        = 0x4,
        TEMP_COMPLETION_CONTEXT         = COMPLETION_CONTEXT | TEMPORAL
    };

    Request_container(
            const std::string& name,
            Request_waiter::Ptr waiter = Request_waiter::Create());

    virtual
    ~Request_container();

    virtual Type
    Get_type() const
    {
        return Type::NONE;
    }

    bool
    Check_type(Type mask)
    {
        return static_cast<int>(Get_type()) & static_cast<int>(mask);
    }

    void
    Submit_request(Request::Ptr request);

    void
    Submit_request_locked(
                Request::Ptr request,
                Request_waiter::Locker locker);

    int
    Process_requests(int requests_limit = 0);

    int
    Process_requests(std::unique_lock<std::mutex> &lock, int requests_limit = 0);

    void
    Set_waiter(Request_waiter::Ptr waiter);

    Request_waiter::Ptr
    Get_waiter() const
    {
        return waiter;
    }

    const std::string&
    Get_name()
    {
        return name;
    }

    void
    Enable();

    void
    Disable();

    bool
    Is_enabled() const;

protected:
    Request_waiter::Ptr waiter;
    std::list<Request::Ptr> request_queue;

    void
    Processing_loop();

    virtual void
    Process_request(Request::Ptr request) = 0;

    virtual void
    On_enable();

    virtual void
    On_disable();

    void
    Set_disabled();

    virtual void
    On_wait_and_process();

private:
    void
    Submit_request_impl(
            Request::Ptr request,
            Request_waiter::Locker locker);

    void
    Abort_requests();

    std::atomic_bool is_enabled = { false };

    std::atomic_bool disable_ongoing = { false };

    std::atomic_bool abort_ongoing = { false };

    const std::string name;

    std::list<Request::Ptr> aborted_request_queue;
};

typedef Request_container::Request_waiter Request_waiter;

class Request_container::Request_waiter::Locker {
public:
    Locker(Request_waiter::Ptr waiter, bool want_notify = false);

    ~Locker();

    Locker(const Locker &) = delete;

    Locker(Locker &&);

    void
    Lock();

    void
    Unlock();

    bool
    Want_notify() const;

    bool
    Is_same_waiter(const Request_waiter::Ptr&) const;

private:
    Request_waiter::Ptr waiter;
    bool want_notify;
    bool is_locked;
};

typedef Request_container::Request Request;

} /* namespace vsm */
} /* namespace ugcs */

#endif /* _UGCS_VSM_REQUEST_CONTAINER_H_ */