concurrency.h

Go to the documentation of this file.

// License: Apache 2.0. See LICENSE file in root directory.
// Copyright(c) 2015 Intel Corporation. All Rights Reserved.

#pragma once
#include <queue>
#include <mutex>
#include <condition_variable>
#include <thread>
#include <atomic>
#include <functional>
#include <cassert>

const int QUEUE_MAX_SIZE = 10;
// Simplest implementation of a blocking concurrent queue for thread messaging
template<class T>
class single_consumer_queue
{
    std::deque<T> _queue;
    mutable std::mutex _mutex;
    std::condition_variable _deq_cv; // not empty signal
    std::condition_variable _enq_cv; // not full signal

    unsigned int const _cap;
    bool _accepting;

    std::function<void(T const &)> const _on_drop_callback;

public:
    explicit single_consumer_queue< T >( unsigned int cap = QUEUE_MAX_SIZE,
                                         std::function< void( T const & ) > on_drop_callback = nullptr )
        : _cap( cap )
        , _accepting( true )
        , _on_drop_callback( on_drop_callback )
    {
    }

    // Enqueue an item onto the queue.
    // If the queue grows beyond capacity, the front will be removed, losing whatever was there!
    bool enqueue(T&& item)
    {
        std::unique_lock<std::mutex> lock(_mutex);
        if( ! _accepting )
        {
            if( _on_drop_callback )
                _on_drop_callback( item );
            return false;
        }

        _queue.push_back(std::move(item));

        if( _queue.size() > _cap )
        {
            if( _on_drop_callback )
                _on_drop_callback( _queue.front() );
            _queue.pop_front();
        }

        lock.unlock();

        // We pushed something -- let others know there's something to dequeue
        _deq_cv.notify_one();

        return true;
    }


    // Enqueue an item, but wait for room if there isn't any
    // Returns true if the enqueue succeeded
    bool blocking_enqueue(T&& item)
    {
        std::unique_lock<std::mutex> lock(_mutex);
        _enq_cv.wait( lock, [this]() {
            return _queue.size() < _cap; } );
        if( ! _accepting )
        {
            // We shouldn't be adding anything to the queue when we're stopping
            if( _on_drop_callback )
                _on_drop_callback( item );
            return false;
        }

        _queue.push_back(std::move(item));
        lock.unlock();

        // We pushed something -- let another know there's something to dequeue
        _deq_cv.notify_one();

        return true;
    }


    // Remove one item; if unavailable, wait for it
    // Return true if an item was removed -- otherwise, false
    bool dequeue( T * item, unsigned int timeout_ms )
    {
        std::unique_lock<std::mutex> lock(_mutex);
        if( ! _deq_cv.wait_for( lock,
                                std::chrono::milliseconds( timeout_ms ),
                                [this]() { return ! _accepting || ! _queue.empty(); } )
            || _queue.empty() )
        {
            return false;
        }

        *item = std::move(_queue.front());
        _queue.pop_front();

        // We've made room -- let whoever is waiting for room know about it
        _enq_cv.notify_one();

        return true;
    }

    // Remove one item if available; do not wait for one
    // Return true if an item was removed -- otherwise, false
    bool try_dequeue(T* item)
    {
        std::lock_guard< std::mutex > lock( _mutex );
        if( _queue.empty() )
            return false;

        *item = std::move(_queue.front());
        _queue.pop_front();

        // We've made room -- let whoever is waiting for room know about it
        _enq_cv.notify_one();

        return true;
    }

    template< class Fn >
    bool peek( Fn fn ) const
    {
        std::lock_guard< std::mutex > lock( _mutex );
        if( _queue.empty() )
            return false;
        fn( _queue.front() );
        return true;
    }

    template< class Fn >
    bool peek( Fn fn )
    {
        std::lock_guard< std::mutex > lock( _mutex );
        if( _queue.empty() )
            return false;
        fn( _queue.front() );
        return true;
    }

    void stop()
    {
        std::lock_guard< std::mutex > lock( _mutex );

        // We no longer accept any more items!
        _accepting = false;

        _clear();
    }

    void clear()
    {
        std::lock_guard< std::mutex > lock( _mutex );
        _clear();
    }

protected:
    void _clear()
    {
        _queue.clear();

        // Wake up anyone who is waiting for room to enqueue, or waiting for something to dequeue -- there's nothing now
        _enq_cv.notify_all();
        _deq_cv.notify_all();
    }

public:
    void start()
    {
        std::lock_guard< std::mutex > lock( _mutex );
        _accepting = true;
    }

    bool started() const { return _accepting; }
    bool stopped() const { return ! started(); }

    size_t size() const
    {
        std::lock_guard< std::mutex > lock( _mutex );
        return _queue.size();
    }

    bool empty() const { return ! size(); }
};

// A single_consumer_queue meant to hold frame_holder objects
template<class T>
class single_consumer_frame_queue
{
    single_consumer_queue<T> _queue;

public:
    single_consumer_frame_queue< T >( unsigned int cap = QUEUE_MAX_SIZE,
                                      std::function< void( T const & ) > on_drop_callback = nullptr )
        : _queue( cap, on_drop_callback )
    {
    }

    bool enqueue( T && item )
    {
        if( item->is_blocking() )
            return _queue.blocking_enqueue( std::move( item ) );
        else
            return _queue.enqueue( std::move( item ) );
    }

    bool dequeue(T* item, unsigned int timeout_ms)
    {
        return _queue.dequeue(item, timeout_ms);
    }

    bool try_dequeue(T* item)
    {
        return _queue.try_dequeue(item);
    }

    template< class Fn >
    bool peek( Fn fn ) const
    {
        return _queue.peek( fn );
    }

    template< class Fn >
    bool peek( Fn fn )
    {
        return _queue.peek( fn );
    }

    void clear()
    {
        _queue.clear();
    }

    void stop()
    {
        _queue.stop();
    }

    void start()
    {
        _queue.start();
    }

    size_t size() const
    {
        return _queue.size();
    }

    bool empty() const
    {
        return _queue.empty();
    }

    bool started() const { return _queue.started(); }
    bool stopped() const { return _queue.stopped(); }
};

// The dispatcher is responsible for dispatching generic 'actions': any thread can queue an action
// (lambda) for dispatch, while the dispatcher maintains a single thread that runs these actions one
// at a time.
//
class dispatcher
{
public:
    // An action, when run, takes a 'cancellable_timer', which it may ignore. This class allows the
    // action to perform some sleep and know that, if the dispatcher is shutting down, its sleep
    // will still be interrupted.
    //
    class cancellable_timer
    {
        dispatcher* _owner;
    
    public:
        cancellable_timer(dispatcher* owner)
            : _owner(owner)
        {}

        bool was_stopped() const { return _owner->_was_stopped.load(); }

        // Replacement for sleep() -- try to sleep for a time, but stop if the
        // dispatcher is stopped
        // 
        // Return false if the dispatcher was stopped, true otherwise
        //
        template< class Duration >
        bool try_sleep( Duration sleep_time )
        {
            using namespace std::chrono;

            std::unique_lock<std::mutex> lock(_owner->_was_stopped_mutex);
            if( was_stopped() )
                return false;
            // wait_for() returns "false if the predicate pred still evaluates to false after the
            // rel_time timeout expired, otherwise true"
            return ! (
                _owner->_was_stopped_cv.wait_for( lock, sleep_time, [&]() { return was_stopped(); } ) );
        }
    };

    // An action is any functor that accepts a cancellable timer
    typedef std::function<void(cancellable_timer const &)> action;

    // Certain conditions (see invoke()) may cause actions to be lost, e.g. when the queue gets full
    // and we're non-blocking. The on_drop_callback allows caputring of these instances, if we
    // want...
    //
    dispatcher( unsigned int queue_capacity,
                std::function< void( action ) > on_drop_callback = nullptr );

    ~dispatcher();

    bool empty() const { return _queue.empty(); }

    // Main invocation of an action: this will be called from any thread, and basically just queues
    // up the actions for our dispatching thread to handle them.
    // 
    // A blocking invocation means that it will wait until there's room in the queue: if not
    // blocking and the queue is full, the action will be queued at the expense of the next action
    // in line (the oldest) for dispatch!
    //
    template<class T>
    void invoke(T item, bool is_blocking = false)
    {
        if (!_was_stopped)
        {
            if(is_blocking)
                _queue.blocking_enqueue(std::move(item));
            else
                _queue.enqueue(std::move(item));
        }
    }

    // Like above, but synchronous: will return only when the action has actually been dispatched.
    //
    template<class T>
    void invoke_and_wait(T item, std::function<bool()> exit_condition, bool is_blocking = false)
    {
        bool done = false;

        //action
        auto func = std::move(item);
        invoke([&, func](dispatcher::cancellable_timer c)
        {
            std::lock_guard<std::mutex> lk(_blocking_invoke_mutex);
            func(c);

            done = true;
            _blocking_invoke_cv.notify_one();
        }, is_blocking);

        //wait
        std::unique_lock<std::mutex> lk(_blocking_invoke_mutex);
        _blocking_invoke_cv.wait(lk, [&](){ return done || exit_condition(); });
    }

    // Stops the dispatcher. This is not a pause: it will clear out the queue, losing any pending
    // actions!
    //
    void stop();

    // A dispatcher starts out 'started' after construction. It can then be stopped and started
    // again if needed.
    //
    void start();

    // Return when all items in the queue are finished (within a timeout).
    // If additional items are added while we're waiting, those will not be waited on!
    //
    bool flush(std::chrono::steady_clock::duration timeout = std::chrono::seconds(10) );


private:
    // Return true if dispatcher is started (within a timeout).
    // false if not or the dispatcher is no longer alive
    //
    bool _wait_for_start( int timeout_ms );

    friend cancellable_timer;

    single_consumer_queue<std::function<void(cancellable_timer)>> _queue;
    std::thread _thread;

    std::atomic<bool> _was_stopped;
    std::condition_variable _was_stopped_cv;
    std::mutex _was_stopped_mutex;

    std::mutex _dispatch_mutex;

    std::condition_variable _blocking_invoke_cv;
    std::mutex _blocking_invoke_mutex;

    std::atomic<bool> _is_alive;
};

template<class T = std::function<void(dispatcher::cancellable_timer)>>
class active_object
{
public:
    active_object(T operation)
        : _operation(std::move(operation)), _dispatcher(1), _stopped(true)
    {
    }

    void start()
    {
        _stopped = false;
        _dispatcher.start();

        do_loop();
    }

    void stop()
    {
        if (!_stopped.load()) {
            _stopped = true;
            _dispatcher.stop();
        }
    }

    ~active_object()
    {
        stop();
    }

    bool is_active() const
    {
        return !_stopped;
    }
private:
    void do_loop()
    {
        _dispatcher.invoke([this](dispatcher::cancellable_timer ct)
        {
            _operation(ct);
            if (!_stopped)
            {
                do_loop();
            }
        });
    }

    T _operation;
    dispatcher _dispatcher;
    std::atomic<bool> _stopped;
};

class watchdog
{
public:
    watchdog(std::function<void()> operation, uint64_t timeout_ms) :
            _timeout_ms(timeout_ms), _operation(std::move(operation))
    {
        _watcher = std::make_shared<active_object<>>([this](dispatcher::cancellable_timer cancellable_timer)
        {
            if(cancellable_timer.try_sleep( std::chrono::milliseconds( _timeout_ms )))
            {
                if(!_kicked)
                    _operation();
                std::lock_guard<std::mutex> lk(_m);
                _kicked = false;
            }
        });
    }

    ~watchdog()
    {
        if(_running)
            stop();
    }

    void start() { std::lock_guard<std::mutex> lk(_m); _watcher->start(); _running = true; }
    void stop() { { std::lock_guard<std::mutex> lk(_m); _running = false; } _watcher->stop(); }
    bool running() { std::lock_guard<std::mutex> lk(_m); return _running; }
    void set_timeout(uint64_t timeout_ms) { std::lock_guard<std::mutex> lk(_m); _timeout_ms = timeout_ms; }
    void kick() { std::lock_guard<std::mutex> lk(_m); _kicked = true; }

private:
    std::mutex _m;
    uint64_t _timeout_ms;
    bool _kicked = false;
    bool _running = false;
    std::function<void()> _operation;
    std::shared_ptr<active_object<>> _watcher;
};