threadwork.h

Go to the documentation of this file.

#pragma once

/*+
________________________________________________________________________

 (C) dGB Beheer B.V.; (LICENSE) http://opendtect.org/OpendTect_license.txt
 Author:        Kristofer Tingdahl
 Date:          4-11-2002
________________________________________________________________________


-*/

#include "basicmod.h"
#include "task.h"
#include "genc.h"
#include "objectset.h"
#include "notify.h"
#include "bufstringset.h"

typedef bool (*StaticTaskFunction)();
typedef bool (CallBacker::*TaskFunction)();

namespace Threads
{
class Thread;
class ConditionVar;
class WorkThread;
class Work;


mExpClass(Basic) WorkManager : public CallBacker
{
public:

                                //Interface from outside world
                                WorkManager(int nrthreads=-1);
                                ~WorkManager();

    enum QueueType              { MultiThread, SingleThread, Manual };
    int                         addQueue(QueueType type,const char* name);
    void                        setQueueName(int queueid,const char*);
    int                         queueSize(int queueid) const;
    void                        emptyQueue(int queueid,bool finishall);
    void                        removeQueue(int queueid,bool finishall);
    static int                  cDefaultQueueID() { return 0; }
    bool                        executeQueue(int queueid);
    void                        addWork(const Work&,CallBack* finished =0,
                                        int queueid=cDefaultQueueID(),
                                        bool putfirstinline=false,
                                        bool discardduplicates=false,
                                        bool forcedifferentthread=false);
    bool                        addWork(TypeSet<Work>&, int queueid = -1,
                                        bool firstinline = false);
    bool                        executeWork( Work*, int sz, int queueid = -1,
                                        bool firstinline = false );
    bool                        removeWork(const Work&);
    const Work*                 getWork(CallBacker*) const;
    bool                        getWorkExitStatus(CallBacker*) const;
    int                         nrThreads() const { return threads_.size(); }
    int                         nrFreeThreads() const;
    bool                        isWorkThread() const;

    Notifier<WorkManager>       isidle;

    static Threads::WorkManager& twm();

    Notifier<WorkManager>       isShuttingDown;
    void                        shutdown();

protected:

    int                         queueSizeNoLock(int queueid) const;
    int                         reportFinishedAndAskForMore(WorkThread*,
                                                            int oldqueueid );
    inline void                 reduceWorkload(int queueidx);

    friend class                WorkThread;

    //Linked (one entry per que-entry)
    TypeSet<Work>               workload_;
    TypeSet<int>                workqueueid_;
    TypeSet<CallBack>           callbacks_;

    ObjectSet<WorkThread>       threads_;
    ObjectSet<const void>       threadids_;
    ObjectSet<WorkThread>       freethreads_;

    //Linked (one entry per queue)
    TypeSet<int>                queueids_;
    TypeSet<int>                queueworkload_; //Nr threads working on it
    TypeSet<QueueType>          queuetypes_;
    BufferStringSet             queuenames_;
    BoolTypeSet                 queueisclosing_;

    ConditionVar&               workloadcond_;

    int                         freeid_;
    const int                   twmid_; 
};


mExpClass(Basic) Work
{
public:
    inline              Work();
    inline              Work(const CallBack&);
    inline              Work(CallBacker* o,TaskFunction f);
    inline              Work(StaticTaskFunction f);
    inline              Work(Task& t,bool takeover);
    bool                operator==(const Work&) const;

    inline bool         isOK() const;
    inline bool doRun();

protected:

    friend class        WorkThread;
    friend class        WorkManager;
    void                destroy();
    CallBacker*         obj_;
    CallBackFunction    cbf_;
    TaskFunction        tf_;
    StaticTaskFunction  stf_;
    bool                takeover_;
};

#define mSTFN(clss,fn) ((::TaskFunction)(&clss::fn))
#define mWMT(obj,clss,fn) ::Threads::Work( obj, mSTFN(clss,fn) )


}; // Namespace

inline Threads::Work::Work()
    : obj_( 0 ), cbf_( 0 ), tf_( 0 ), stf_( 0 )                         {}


inline Threads::Work::Work( const CallBack& cb )
    : obj_( const_cast<CallBacker*>(cb.cbObj()) )
    , cbf_( cb.cbFn() ), tf_( 0 ), stf_( 0 )                            {}


inline Threads::Work::Work( CallBacker* o, TaskFunction f )
    : obj_( o ), cbf_( 0 ), tf_( f ), stf_( 0 ), takeover_( false )     {}


inline Threads::Work::Work( StaticTaskFunction f )
    : obj_( 0 ), cbf_( 0 ), tf_( 0 ), stf_( f )                         {}


inline Threads::Work::Work( Task& t, bool takeover )
    : obj_( &t ), cbf_( 0 ), tf_( mSTFN(Task,execute) ), stf_( 0 )
    , takeover_( takeover )                                             {}


inline bool Threads::Work::isOK() const
{ return stf_ || (obj_ && (tf_ || cbf_ ) ); }


inline bool Threads::Work::doRun()
{
    if ( stf_ )     return stf_();
    if ( tf_ )
    {
        const bool res = (obj_->*tf_)();
        if ( takeover_ ) delete obj_;
        return res;
    }

    (obj_->*cbf_)( 0 );

    return true;
}