threader.h

Go to the documentation of this file.

#ifndef AUDIOGRAPHER_THREADER_H
#define AUDIOGRAPHER_THREADER_H
 
#include <atomic>
#include <vector>
#include <algorithm>
 
#include <glibmm/threadpool.h>
#include <glibmm/timeval.h>
#include <sigc++/slot.h>
#include <boost/format.hpp>
 
#include <glib.h>
 
#include "pbd/atomic.h"
 
#include "audiographer/visibility.h"
#include "audiographer/source.h"
#include "audiographer/sink.h"
#include "audiographer/exception.h"
 
namespace AudioGrapher
{
 
class /*LIBAUDIOGRAPHER_API*/ ThreaderException : public Exception
{
  public:
        template<typename T>
        ThreaderException (T const & thrower, std::exception const & e)
                : Exception (thrower,
                        boost::str ( boost::format
                        ("\n\t- Dynamic type: %1%\n\t- what(): %2%")
                        % DebugUtils::demangled_name (e) % e.what() ))
        { }
};
 
template <typename T = DefaultSampleType>
class /*LIBAUDIOGRAPHER_API*/ Threader : public Source<T>, public Sink<T>
{
  private:
        typedef std::vector<typename Source<T>::SinkPtr> OutputVec;
 
  public:
 
        Threader (Glib::ThreadPool & thread_pool, long wait_timeout_milliseconds = 500)
          : thread_pool (thread_pool)
          , wait_timeout (wait_timeout_milliseconds)
        {
                readers.store (0);
        }
 
        virtual ~Threader () {}
 
        void add_output (typename Source<T>::SinkPtr output) { outputs.push_back (output); }
 
        void clear_outputs () { outputs.clear (); }
 
        void remove_output (typename Source<T>::SinkPtr output) {
                typename OutputVec::iterator new_end = std::remove(outputs.begin(), outputs.end(), output);
                outputs.erase (new_end, outputs.end());
        }
 
        void process (ProcessContext<T> const & c)
        {
                wait_mutex.lock();
 
                exception.reset();
 
                unsigned int outs = outputs.size();
                (void) readers.fetch_add (outs);
                for (unsigned int i = 0; i < outs; ++i) {
                        thread_pool.push (sigc::bind (sigc::mem_fun (this, &Threader::process_output), c, i));
                }
 
                wait();
        }
 
        using Sink<T>::process;
 
  private:
 
        void wait()
        {
                while (readers.load () != 0) {
                        gint64 end_time = g_get_monotonic_time () + (wait_timeout * G_TIME_SPAN_MILLISECOND);
                        wait_cond.wait_until(wait_mutex, end_time);
                }
 
                wait_mutex.unlock();
 
                if (exception) {
                        throw *exception;
                }
        }
 
        void process_output(ProcessContext<T> const & c, unsigned int output)
        {
                try {
                        outputs[output]->process (c);
                } catch (std::exception const & e) {
                        // Only first exception will be passed on
                        exception_mutex.lock();
                        if(!exception) { exception.reset (new ThreaderException (*this, e)); }
                        exception_mutex.unlock();
                }
 
                if (PBD::atomic_dec_and_test (readers)) {
                        wait_cond.signal();
                }
        }
 
        OutputVec outputs;
 
        Glib::ThreadPool&    thread_pool;
        Glib::Threads::Mutex wait_mutex;
        Glib::Threads::Cond  wait_cond;
 
        std::atomic<int> readers;
        long         wait_timeout;
 
        Glib::Threads::Mutex exception_mutex;
        std::shared_ptr<ThreaderException> exception;
 
};
 
} // namespace
 
#endif //AUDIOGRAPHER_THREADER_H