libosgearth-dev 2.9.0+dfsg-1 / usr / include / osgEarth / TaskService

/* -*-c++-*- */
/* osgEarth - Dynamic map generation toolkit for OpenSceneGraph
 * Copyright 2016 Pelican Mapping
 * http://osgearth.org
 *
 * osgEarth is free software; you can redistribute it and/or modify
 * it under the terms of the GNU Lesser General Public License as published by
 * the Free Software Foundation; either version 2 of the License, or
 * (at your option) any later version.
 *
 * This program is distributed in the hope that it will be useful,
 * but WITHOUT ANY WARRANTY; without even the implied warranty of
 * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
 * GNU Lesser General Public License for more details.
 *
 * You should have received a copy of the GNU Lesser General Public License
 * along with this program.  If not, see <http://www.gnu.org/licenses/>
 */
#ifndef OSGEARTH_TASK_SERVICE
#define OSGEARTH_TASK_SERVICE 1

#include <osgEarth/Common>
#include <osgEarth/Progress>
#include <osgEarth/ThreadingUtils>
#include <osg/Referenced>
#include <osg/Timer>
#include <OpenThreads/ReentrantMutex>
#include <queue>
#include <list>
#include <string>
#include <map>

namespace osgEarth
{
    class OSGEARTH_EXPORT TaskRequest : public osg::Referenced
    {
    public:
        enum State {
            STATE_IDLE,
            STATE_PENDING,
            STATE_IN_PROGRESS,
            STATE_COMPLETED
        };
    public:
        TaskRequest( float priority =0.0f );

        /** dtor */
        virtual ~TaskRequest() { }

        // the actual task code
        virtual void operator()( ProgressCallback* progress ) =0;

        void run();
        void cancel();

        bool isIdle() const { return _state == STATE_IDLE; }
        bool isPending() const { return _state == STATE_PENDING; }
        bool isCompleted() const { return _state == STATE_COMPLETED; }
        bool isInProgress() const { return _state == STATE_IN_PROGRESS; }
        bool isRunning() const { return isPending() || isInProgress(); }

        bool wasCanceled() const;

        void setPriority( float value ) { _priority = value; }
        float getPriority() const { return _priority; }
        State getState() const { return _state; }
        void setState(State s) { _state = s; }
        void setStamp(int stamp) { _stamp = stamp; }
        int getStamp() const { return _stamp; }
        osg::Referenced* getResult() const { return _result.get(); }
        ProgressCallback* getProgressCallback() { return _progress.get(); }
        void setProgressCallback(ProgressCallback* progress) { _progress = progress? progress : new ProgressCallback(); }
        const std::string& getName() const { return _name; }
        void setName( const std::string& name ) { _name = name; }
        void reset() { _result = 0L; }
        osg::Timer_t startTime() const { return _startTime; }
        osg::Timer_t endTime() const { return _endTime; }
        double runTime() const { return osg::Timer::instance()->delta_s(_startTime,_endTime); }

        void setCompletedEvent( Threading::Event* value ) { _completedEvent = value; }
        Threading::Event* getCompletedEvent() const { return _completedEvent; }

    protected:
        float _priority;
        volatile State _state;
        volatile int _stamp;
        osg::ref_ptr<osg::Referenced> _result;
        osg::ref_ptr< ProgressCallback > _progress;
        std::string _name;
        osg::Timer_t _startTime;
        osg::Timer_t _endTime;
        Threading::Event* _completedEvent;
    };

    /**
     * Special marker class that is used to signify the end of the work.  TaskRequestThreads should shut down when they receive a PoisonPill
     */
    class OSGEARTH_EXPORT PoisonPill : public TaskRequest
    {        
        virtual void operator()( ProgressCallback* progress )
        {
        }
    };

    typedef std::list< osg::ref_ptr<TaskRequest> > TaskRequestList;

    typedef std::vector< osg::ref_ptr<TaskRequest> > TaskRequestVector;

    typedef std::multimap< float, osg::ref_ptr<TaskRequest> > TaskRequestPriorityMap;
    
    /**
     * Convenience template for creating a task that synchronized with an event.
     * Initialze multiple ParallelTask's with a common MultiEvent (semaphore) to
     * run them in parallel and wait for them all to complete.
     */
    template<typename T>
    struct ParallelTask : public TaskRequest, T
    {
        ParallelTask() : _mev(0L), _sev(0L) { }
        ParallelTask( Threading::MultiEvent* ev ) : _mev(ev), _sev(0L) { }
        ParallelTask( Threading::Event* ev ) : _sev(ev), _mev(0L) { }

        void operator()( ProgressCallback* pc ) 
        {
            this->execute();
            if ( _mev )
                _mev->notify();
            else if ( _sev )
                _sev->set();
        }

        Threading::MultiEvent* _mev;
        Threading::Event*      _sev;
    };

    class TaskRequestQueue : public osg::Referenced
    {
    public:
        TaskRequestQueue(unsigned int maxSize=0);

        void add( TaskRequest* request );
        TaskRequest* get();
        void clear();
        void cancel();

        void setDone();

        bool isFull() const;
        bool isEmpty() const;

        unsigned int getMaxSize() const { return _maxSize;}

        void setStamp( int value ) { _stamp = value; }
        int getStamp() const { return _stamp; }

        unsigned int getNumRequests() const;


    private:
        TaskRequestPriorityMap _requests;
        OpenThreads::Mutex _mutex;
        OpenThreads::Condition _notFull;
        OpenThreads::Condition _notEmpty;
        volatile bool _done;
        unsigned int _maxSize;

        int _stamp;
    };
    
    struct TaskThread : public OpenThreads::Thread
    {
        TaskThread( TaskRequestQueue* queue );
        bool getDone() { return _done;}
        void setDone( bool done) { _done = done; }
        void run();
        int cancel();

    private:
        osg::ref_ptr<TaskRequestQueue> _queue;
        osg::ref_ptr<TaskRequest> _request;
        volatile bool _done;
    };

    /** 
     * Manages a priority task queue and associated thread pool.
     */
    class OSGEARTH_EXPORT TaskService : public osg::Referenced
    {
    public:
        TaskService( const std::string& name ="", int numThreads =4, unsigned int maxSize=0 );

        void add( TaskRequest* request );

        void setName( const std::string& value ) { _name = value; }
        const std::string& getName() const { return _name; }

        int getStamp() const;
        void setStamp( int stamp );

        int getNumThreads() const;
        void setNumThreads( int numThreads );

        /**
         *Gets the number of requets left in the queue
         */
        unsigned int getNumRequests() const;

        void waitforThreadsToComplete();

        bool areThreadsRunning();

        void cancelAll();

    private:
        void adjustThreadCount();
        void removeFinishedThreads();

        OpenThreads::ReentrantMutex _threadMutex;
        typedef std::list<TaskThread*> TaskThreads;
        TaskThreads _threads;
        osg::ref_ptr<TaskRequestQueue> _queue;
        int _numThreads;
        int _lastRemoveFinishedThreadsStamp;
        std::string _name;
        virtual ~TaskService();
    };

    /**
     * Manages a pool of TaskService objects, automatically allocating
     * threads among them based on a weighting metric.
     */
    class OSGEARTH_EXPORT TaskServiceManager : public osg::Referenced
    {
    public:
        /**
         * Creates a new manager, and sets the target number of threads to 
         * allocate across all managed task services.
         */
        TaskServiceManager( int numThreads =4 );

        /**
         * Sets a new total target thread count to allocate across all task
         * services under management. (The actual thread count may be higher since
         * each service is guaranteed at least one thread.)
         */
        void setNumThreads( int numThreads );

        /**
         * Gets the total number of threads allocated across all managed
         * task services.
         */
        int getNumThreads() const { return _numThreads; }

        /**
         * Adds a new task service to the manager and reallocates the thread pool
         * across the remaining services.
         */
        TaskService* add( UID uid, float weight =1.0f );

        /**
         * Gets the names task service.
         */
        TaskService* get( UID uid ) const;

        /**
         * Gets a task service by name, creating it if it does not yet exist
         */
        TaskService* getOrAdd( UID uid, float weight =1.0f );

        /**
         * Removes a task service from management, and reallocates the thread pool
         * across the remaining services.
         */
        void remove( TaskService* service );
        void remove( UID uid );

        /**
         * Assigned a weight value to a particular task service. The manager will
         * re-distribute available threads for all registered task services.
         */
        void setWeight( TaskService* service, float weight );

    private:
        typedef std::pair< osg::ref_ptr<TaskService>, float > WeightedTaskService;
        typedef std::map< UID, WeightedTaskService > TaskServiceMap;
        TaskServiceMap _services;
        int _numThreads, _targetNumThreads;
        OpenThreads::Mutex _taskServiceMgrMutex;

        void reallocate( int targetNumThreads );
    };
}

#endif // OSGEARTH_TASK_SERVICE