PipeLine.hpp

//
// 3D Shape Analysis Library.
// Copyright (C) 2010 AQSENSE, S.L.
//
#if !defined (AQSENSE_SAL3DPP_PIPELINE_HPP)
#define AQSENSE_SAL3DPP_PIPELINE_HPP

#if defined (_MSC_VER) && (_MSC_VER >= 1020)
#pragma once
#endif // _MSC_VER && _MSC_VER >= 1020

#include <cassert>
#include <stdexcept>
#include <string>
#include <sal3d/threadpool.h>
#include <sal3dpp/Common.hpp>

namespace sal3d
{
    template<typename Task>
    class PipeLine
    {
        public:
            PipeLine(int nthreads)
                :_nthreads(nthreads)
            {
                Error e;
                sal3d_threadpool_new(nthreads, &_threadpool, 0, &e.e);
                if (e.e.value < 0)
                    throw e;
            } 

            PipeLine(int nthreads, const char *basename)
                :_nthreads(nthreads)
            {
                Error e;
                sal3d_threadpool_new(nthreads, &_threadpool, basename, &e.e);
                if (e.e.value < 0)
                    throw e;
            } 

            ~PipeLine()
            {
                sal3d_threadpool_delete(&_threadpool, 0);
            }

            void post(const Task &t)
            {
                // This will be deleted after running, on callback.
                Task *copy = new Task(t);

                Error e;
                sal3d_threadpool_post_sequential(_threadpool,
                        callbackParallel,
                        reinterpret_cast<void *>(copy),
                        callbackSequential,
                        reinterpret_cast<void *>(copy),
                        &e.e
                        );
                if (e.e.value < 0)
                {
                    delete copy;
                    throw e;
                }
            }

            void wait()
            {
                Error e;
                sal3d_threadpool_wait(_threadpool, &e.e);
                if (e.e.value < 0)
                    throw e;
            }

            bool wait(int timeout_ms)
            {
                Error e;
                sal3d_threadpool_wait_timeout(_threadpool,
                        timeout_ms, &e.e);
                if (e.e.value < 0)
                {
                    if (e.e.value == SAL3D_ERROR_TIMEOUT)
                        return false;
                    throw e;
                }
                return true;
            }

            int nthreads() const
            {
                return _nthreads;
            }

        private:

            static void callbackParallel(void *data)
            {
                Task *t = reinterpret_cast<Task *>(data);
                try {
                    t->parallel();
                }
                catch(...)
                {
                    assert(0 && "Exception thrown in callback for Task");
                }
            }

            static void callbackSequential(void *data)
            {
                Task *t = reinterpret_cast<Task *>(data);
                try {
                    t->sequential();
                }
                catch(...)
                {
                    assert(0 && "Exception thrown in callback for Task");
                }

                delete t;
            }

            /* Uncopiable */
            PipeLine(const PipeLine &);
            PipeLine& operator=(const PipeLine &);

            sal3d_threadpool _threadpool;
            int _nthreads;
    };
}
#endif // !AQSENSE_SAL3DPP_PIPELINE_HPP