threads.hpp

浏览该文件的文档.

#ifndef __THREADS_HPP__
#define __THREADS_HPP__

namespace winux
{

//=========================================================================================
class ThreadSysError : public Error
{
public:
    ThreadSysError( int errCode, AnsiString const & errStr ) throw() : Error( errCode, errStr ) {  }
    int getErrCode() const throw() { return this->getErrType(); }
};

//=========================================================================================
// 线程相关错误返回值为errno.h定义的错误码
//=========================================================================================

class WINUX_DLL ThreadAttr
{
public:
    explicit ThreadAttr( bool isCreate = true );
    ~ThreadAttr();

    ThreadAttr( ThreadAttr && other );
    ThreadAttr & operator = ( ThreadAttr && other );

    int create();

    int destroy();

    operator bool() const;

public:
    //attributes:

    enum DetachStateType
    {
        threadCreateJoinable = 0,  
        threadCreateDetached = 1,  
    };
    void setDetachState( DetachStateType detachState = threadCreateJoinable );
    DetachStateType getDetachState() const;

    void setStackSize( size_t stackSize );
    size_t getStackSize() const;

private:
    MembersWrapper<struct ThreadAttr_Data> _self;
    friend class Thread;

    DISABLE_OBJECT_COPY(ThreadAttr)
};

class WINUX_DLL ThreadId
{
public:
    ThreadId();
    ~ThreadId();
    ThreadId( ThreadId const & other );
    ThreadId & operator = ( ThreadId const & other );
    ThreadId( ThreadId && other );
    ThreadId & operator = ( ThreadId && other );

    bool operator == ( ThreadId const & other ) const;
    bool operator != ( ThreadId const & other ) const { return !this->operator == (other); }

private:
    MembersWrapper<struct ThreadId_Data> _self;

    friend class Thread;
};

class ThreadGroup;
class WINUX_DLL Thread
{
public:

    typedef void * (* ThreadFuncPtr)( void * param );

public:

    static ThreadId Self();

    static void Exit( void * exitVal = NULL );

    static void * Join( Thread & otherThread );

private:
    // 默认线程处理函数, param被传入Thread对象, 调用thObj->run()
    static void * _ThreadDefaultFunc( void * param );

public:
    explicit Thread( bool isStartup = false ) : _attr(false), _exitVal(NULL), _deleter(NULL), _group(NULL), _isRunning(false)
    {
        if ( isStartup ) this->startup();
    }

    template < typename _Fx, typename... _ArgType >
    explicit Thread( bool isStartup, _Fx routine, _ArgType&&... arg ) : _attr(false), _exitVal(NULL), _deleter(NULL), _group(NULL), _isRunning(false)
    {
        this->setRunable( routine, std::forward<_ArgType>(arg)... );
        if ( isStartup ) this->startup();
    }

    virtual ~Thread() { }

    Thread( Thread && other );

    Thread & operator = ( Thread && other );

    int startup();

    template < typename _Fx, typename... _ArgType >
    int startup( _Fx routine, _ArgType&&... arg )
    {
        this->setRunable( routine, std::forward<_ArgType>(arg)... );
        return this->startup();
    }

    int startupEx( ThreadFuncPtr startRoutine, void * param );

    void * joined();

    int detach();

public:
    //attributes:

    ThreadAttr & attr();

    template < typename _Fx, typename... _ArgType >
    Thread & setRunable( _Fx routine, _ArgType&&... arg )
    {
        this->_runable.attachNew( NewRunable( routine, std::forward<_ArgType>(arg)... ) );
        return *this;
    }

    Thread & setDeleter( SimpleDeleterContext * deleter = NULL )
    {
        _deleter = deleter;
        return *this;
    }
    Thread & setDefaultDeleter() { return this->setDeleter( new SimpleDefaultDeleterContext<Thread*>(this) ); }
    template < typename _Dt >
    Thread & setCustomDeleter( _Dt fn ) { return this->setDeleter( new SimpleCustomDeleterContext< Thread*, _Dt >( this, fn ) ); }

    void * getExitVal() const { return _exitVal; }
    Thread & setExitVal( void * exitVal )
    {
        _exitVal = exitVal;
        return *this;
    }

    ThreadId get() const
    {
        return _threadId;
    }

protected:
    virtual void run();

private:
    ThreadAttr _attr; 
    void * _exitVal; 
    SimpleDeleterContext * _deleter; 
    ThreadGroup * _group; 
    bool _isRunning; 
    ThreadId _threadId; 
    SimplePointer<Runable> _runable; 

    friend class ThreadGroup;
    DISABLE_OBJECT_COPY(Thread)
};

//=========================================================================================
class WINUX_DLL MutexAttr
{
public:
    explicit MutexAttr( bool isCreate = true );
    ~MutexAttr();

    MutexAttr( MutexAttr && other );
    MutexAttr & operator = ( MutexAttr && other );

    int create();

    int destroy();

    operator bool() const;

private:
    MembersWrapper<struct MutexAttr_Data> _self;
    friend class Mutex;

    DISABLE_OBJECT_COPY(MutexAttr)
};

class WINUX_DLL Mutex : public winux::ILockObj
{
public:
    explicit Mutex( bool isCreate = false );
    ~Mutex();

    Mutex( Mutex && other );
    Mutex & operator = ( Mutex && other );

    int create();

    int destroy();

    bool lock();
    bool tryLock();
    bool unlock();

    MutexAttr & attr();
private:
    MutexAttr _attr; 
    MembersWrapper<struct Mutex_Data> _self;
    friend class Condition;

    DISABLE_OBJECT_COPY(Mutex)
};

//=========================================================================================
class WINUX_DLL ConditionAttr
{
public:
    explicit ConditionAttr( bool isCreate = true );
    ~ConditionAttr();

    ConditionAttr( ConditionAttr && other );
    ConditionAttr & operator = ( ConditionAttr && other );

    int create();

    int destroy();

    operator bool() const;

private:
    MembersWrapper<struct ConditionAttr_Data> _self;
    friend class Condition;

    DISABLE_OBJECT_COPY(ConditionAttr)
};

class WINUX_DLL Condition
{
public:
    explicit Condition( bool isCreate = false );
    ~Condition();

    Condition( Condition && other );
    Condition & operator = ( Condition && other );

    int create();

    int destroy();

    bool wait( Mutex & mutex, double sec = -1 );

    template < typename _Predicate >
    bool waitUntil( _Predicate pred, Mutex & mutex, double sec = -1 )
    {
        while ( !pred() )
            if ( !this->wait( mutex, sec ) )
                return pred();
        return true;
    }

    int notify();

    int notifyAll();

    ConditionAttr & attr();

private:
    ConditionAttr _attr;
    MembersWrapper<struct Condition_Data> _self;
    DISABLE_OBJECT_COPY(Condition)
};

//===========================================================================================
class WINUX_DLL TlsKey
{
public:
    explicit TlsKey( void (*destructor)( void *pv ) = NULL );
    ~TlsKey();

    TlsKey( TlsKey && other );
    TlsKey & operator = ( TlsKey && other );

    int create( void (*destructor)( void *pv ) = NULL );

    int destroy();

    void * get() const;

private:
    MembersWrapper<struct TlsKey_Data> _self;

    friend class TlsVar;
    DISABLE_OBJECT_COPY(TlsKey)
};

class WINUX_DLL TlsVar
{
public:
    explicit TlsVar( TlsKey & tlsKey );
    ~TlsVar();

    TlsVar( TlsVar && other );
    TlsVar & operator = ( TlsVar && other );

    void * get();
    void * get() const;
    void set( void * v );

    template < typename _Ty >
    _Ty get() { return reinterpret_cast<_Ty>( this->get() ); }
    template < typename _Ty >
    _Ty get() const { return reinterpret_cast<_Ty>( this->get() ); }

    template < typename _Ty >
    void set( _Ty v ) { this->set( reinterpret_cast<void*>(v) ); }

    template < typename _Ty >
    _Ty & ref() { return *reinterpret_cast<_Ty*>( this->get() ); }
    template < typename _Ty >
    _Ty const & ref() const { return *reinterpret_cast<_Ty*>( this->get() ); }

private:
    TlsKey * _tlsKey;
    DISABLE_OBJECT_COPY(TlsVar)
};

//===========================================================================================

class WINUX_DLL ThreadGroup
{
public:

    ThreadGroup() : _mtxGroup(true), _cdtGroup(true)
    {
    }

    template < typename _Fx, typename... _ArgType >
    ThreadGroup( int count, _Fx fn, _ArgType&&... arg ) : _mtxGroup(true), _cdtGroup(true)
    {
        int i;
        for ( i = 0; i < count; i++ )
        {
            Thread * p = new Thread( false, fn, std::forward<_ArgType>(arg)... );
            p->_group = this;
            _threads.emplace_back(p);
        }
    }

    virtual ~ThreadGroup()
    {
    }

    ThreadGroup( ThreadGroup && other ) : _mtxGroup( std::move(other._mtxGroup) ), _cdtGroup( std::move(other._cdtGroup) ), _threads( std::move(other._threads) )
    {
    }

    ThreadGroup & operator = ( ThreadGroup && other )
    {
        if ( this != &other )
        {
            // 先释放自身
            this->wait();

            _mtxGroup = std::move(other._mtxGroup);
            _cdtGroup = std::move(other._cdtGroup);
            _threads = std::move(other._threads);
        }
        return *this;
    }

    ThreadGroup & destroy();

    template < typename _Fx, typename... _ArgType >
    ThreadGroup & create( int count, _Fx fn, _ArgType&&... arg )
    {
        this->destroy();

        int i;
        for ( i = 0; i < count; i++ )
        {
            Thread * p = new Thread( false, fn, std::forward<_ArgType>(arg)... );
            p->_group = this;
            _threads.emplace_back(p);
        }
        return *this;
    }

    template < class _ThreadCls >
    ThreadGroup & create( int count )
    {
        this->destroy();

        int i;
        for ( i = 0; i < count; i++ )
        {
            Thread * p = new _ThreadCls();
            p->_group = this;
            _threads.emplace_back(p);
        }
        return *this;
    }

    ThreadGroup & startup();

    bool wait( double sec = -1 );

private:
    static void * _ThreadGroupDefaultFunc( void * param );

    Mutex _mtxGroup; // 互斥量保护数据
    Condition _cdtGroup; // 用于判断组中线程是否全部运行完毕
    std::vector< SimplePointer<Thread> > _threads;

    DISABLE_OBJECT_COPY(ThreadGroup)
};


template < typename _Ty >
class Task;
class ThreadPool;

struct TaskCtx
{
    enum TaskStatus
    {
        taskPending,
        taskRunning,
        taskStop
    };

    Mutex mtxTask; // 互斥量保护数据
    Condition cdtTask; // 用于判断运行状态
    TaskStatus status; // 运行状态
    ThreadPool * pool; // 相关线程池
    WeakPointer<TaskCtx> weakThis;
    SimplePointer<Runable> poolRunable; // 投递到线程池的例程

    bool wait( double sec = -1 )
    {
        ScopeGuard guard(mtxTask);
        return cdtTask.waitUntil( [this] () { return this->status == TaskCtx::taskStop; }, mtxTask, sec );
    }

    void updateStatus( TaskStatus st, bool isNotifyAll = false )
    {
        ScopeGuard guard(mtxTask);
        this->status = st;
        if ( isNotifyAll ) cdtTask.notifyAll(); // 唤醒所有等待此任务的线程
    }

    void rePost();
protected:
    TaskCtx() : mtxTask(true), cdtTask(true), status(taskPending), pool(nullptr) { }
    virtual ~TaskCtx() { }
};

template < typename _Ty >
struct TaskCtxT : public TaskCtx
{
    _Ty val;

    template < typename... _ArgType >
    static SharedPointer<TaskCtxT> Create( _ArgType&& ... arg )
    {
        SharedPointer<TaskCtxT> p( new TaskCtxT( std::forward<_ArgType>(arg)... ) );
        p->weakThis = p;
        return p;
    }

    _Ty get()
    {
        this->wait();
        return val;
    }

    void exec( RunableInvoker<_Ty> * ivk )
    {
        this->val = ivk->invoke();
    }
protected:
    TaskCtxT() { }
};

template <>
struct TaskCtxT<void> : public TaskCtx
{
    template < typename... _ArgType >
    static SharedPointer<TaskCtxT> Create( _ArgType&& ... arg )
    {
        SharedPointer<TaskCtxT> p( new TaskCtxT( std::forward<_ArgType>(arg)... ) );
        p->weakThis = p;
        return p;
    }

    void get()
    {
        this->wait();
    }

    void exec( RunableInvoker<void> * ivk )
    {
        ivk->invoke();
    }
protected:
    TaskCtxT() { }
};

class WINUX_DLL ThreadPool
{
public:
    // 线程池任务等待模式
    enum ThreadPoolMode
    {
        modeWaitKeep, 
        modeWaitTimeRePost 
    };
    explicit ThreadPool( int threadCount = 4, ThreadPoolMode mode = modeWaitKeep, double durationSec = 0.1 ) : _mtxPool(true), _cdtPool(true), _poolStop(false), _mode(mode), _durationSec(durationSec)
    {
        _group.create( threadCount, [this] () {
            while ( !_poolStop )
            {
                SharedPointer<TaskCtx> taskCtx;
                {
                    ScopeGuard guard(_mtxPool);
                    while ( _queueTask.empty() && !_poolStop )
                    {
                        _cdtPool.wait(_mtxPool);
                    }
                    if ( !_queueTask.empty() ) // 如果队列不空
                    {
                        taskCtx = _queueTask.front();
                        _queueTask.pop();

                        _cdtPool.notifyAll(); // 唤醒所有线程池线程
                    }
                }

                if ( taskCtx )
                {
                    taskCtx->updateStatus( TaskCtx::taskRunning, false );
                    taskCtx->poolRunable->run();
                }
            }
        } );
        _group.startup(); // 启动线程组的线程
    }

    virtual ~ThreadPool()
    {
        this->whenEmptyStopAndWait();
    }

    template < typename _Fx, typename... _ArgType >
    Task<typename FuncTraits<_Fx>::ReturnType> task( _Fx fn, _ArgType&&... arg )
    {
        return Task<typename FuncTraits<_Fx>::ReturnType>( this, fn, std::forward<_ArgType>(arg)... );
    }

    void stop()
    {
        ScopeGuard guard(_mtxPool);
        _poolStop = true;
        _cdtPool.notifyAll(); // 唤醒所有线程池线程
    }

    bool wait( double sec = -1 )
    {
        return _group.wait(sec); // 等待全部线程退出
    }

    void whenEmptyStopAndWait()
    {
        if ( true )
        {
            ScopeGuard guard(_mtxPool);
            _cdtPool.waitUntil( [this] () { return _queueTask.empty(); }, _mtxPool );
        }

        this->stop();
        this->wait();
    }

    size_t getTaskCount() const
    {
        ScopeGuard guard( const_cast<Mutex &>(_mtxPool) );
        return _queueTask.size();
    }

    double getDurationSec() const { return _durationSec; }

private:
    // 投递一个任务
    void _postTask( SharedPointer<TaskCtx> taskCtx )
    {
        ScopeGuard guard(_mtxPool);
        _queueTask.push(taskCtx);
        _cdtPool.notify(); // 通知一个等待中的线程
    }

    Mutex _mtxPool;
    Condition _cdtPool; // 用于判断队列是否有数据
    bool _poolStop; // 线程池停止
    ThreadPoolMode _mode; // 等待子任务的模式
    double _durationSec; // 等待的时间
    ThreadGroup _group;
    std::queue< SharedPointer<TaskCtx> > _queueTask;

    friend struct TaskCtx;
    template < typename _Ty0 >
    friend class Task;

    DISABLE_OBJECT_COPY(ThreadPool)
};

// partial TaskCtx
inline void TaskCtx::rePost()
{
    if ( !weakThis.expired() )
    {
        SharedPointer<TaskCtx> p = weakThis.lock();
        p->status = taskPending;
        pool->_postTask(p);
    }
}

template < typename _Ty >
class Task
{
public:
    using ReturnType = _Ty;

    template < typename _Fx, typename... _ArgType >
    Task( ThreadPool * pool, _Fx fnRoutine, _ArgType&& ... argRoutine )
    {
        static_assert( std::is_same< ReturnType, typename FuncTraits<_Fx>::ReturnType >::value , "FuncTraits<_Fx>::ReturnType is not match Task<_Ty>." );
        _taskCtx = TaskCtxT<ReturnType>::Create();
        _taskCtx->pool = pool;
        _taskCtx->status = TaskCtx::taskPending;

        auto routine = MakeSimple( NewRunable( fnRoutine, std::forward<_ArgType>(argRoutine)... ) );
        _taskCtx->poolRunable.attachNew( NewRunable( [routine] ( TaskCtxT<ReturnType> * taskCtx ) {
            taskCtx->exec( routine.get() );
            // 更新运行状态并通知唤醒等待在此任务的线程
            taskCtx->updateStatus( TaskCtx::taskStop, true );
        }, _taskCtx.get() ) );

        // 添加任务到线程池
        _taskCtx->pool->_postTask(_taskCtx);
    }

    template < typename _Fx, typename... _ArgType >
    Task( SharedPointer< TaskCtxT<void> > waitTaskCtx, _Fx fnRoutine, _ArgType&& ... argRoutine )
    {
        static_assert( std::is_same< ReturnType, typename FuncTraits<_Fx>::ReturnType >::value , "FuncTraits<_Fx>::ReturnType is not match Task<_Ty>." );
        _taskCtx = TaskCtxT<ReturnType>::Create();
        _taskCtx->pool = waitTaskCtx->pool;
        _taskCtx->status = TaskCtx::taskPending;

        auto routine = MakeSimple( NewRunable( fnRoutine, std::forward<_ArgType>(argRoutine)... ) );
        _taskCtx->poolRunable.attachNew( NewRunable( [routine] ( SharedPointer< TaskCtxT<void> > waitTaskCtx, TaskCtxT<ReturnType> * taskCtx ) {
            ThreadPool * pool = taskCtx->pool;
            if ( pool->_mode == ThreadPool::modeWaitKeep )
            {
                waitTaskCtx->wait();
                taskCtx->exec( routine.get() );
                // 更新运行状态并通知唤醒等待在此任务的线程
                taskCtx->updateStatus( TaskCtx::taskStop, true );
            }
            else if ( pool->_mode == ThreadPool::modeWaitTimeRePost )
            {
                if ( waitTaskCtx->wait(pool->_durationSec) )
                {
                    taskCtx->exec( routine.get() );
                    // 更新运行状态并通知唤醒等待在此任务的线程
                    taskCtx->updateStatus( TaskCtx::taskStop, true );
                }
                else
                {
                    // 再投递
                    taskCtx->rePost();
                }
            }
        }, waitTaskCtx, _taskCtx.get() ) );

        // 添加任务到线程池
        _taskCtx->pool->_postTask(_taskCtx);
    }

    template < typename _Fx, typename... _ArgType >
    Task( SharedPointer< TaskCtxT<void> > waitTaskCtx, _Fx fnRoutine, typename FuncTraits<_Fx>::ClassType * obj, _ArgType&& ... argRoutine )
    {
        static_assert( std::is_same< ReturnType, typename FuncTraits<_Fx>::ReturnType >::value , "FuncTraits<_Fx>::ReturnType is not match Task<_Ty>." );
        _taskCtx = TaskCtxT<ReturnType>::Create();
        _taskCtx->pool = waitTaskCtx->pool;
        _taskCtx->status = TaskCtx::taskPending;

        auto routine = MakeSimple( NewRunable( fnRoutine, obj, std::forward<_ArgType>(argRoutine)... ) );
        _taskCtx->poolRunable.attachNew( NewRunable( [routine] ( SharedPointer< TaskCtxT<void> > waitTaskCtx, TaskCtxT<ReturnType> * taskCtx ) {
            ThreadPool * pool = taskCtx->pool;
            if ( pool->_mode == ThreadPool::modeWaitKeep )
            {
                waitTaskCtx->wait();
                taskCtx->exec( routine.get() );
                // 更新运行状态并通知唤醒等待在此任务的线程
                taskCtx->updateStatus( TaskCtx::taskStop, true );
            }
            else if ( pool->_mode == ThreadPool::modeWaitTimeRePost )
            {
                if ( waitTaskCtx->wait(pool->_durationSec) )
                {
                    taskCtx->exec( routine.get() );
                    // 更新运行状态并通知唤醒等待在此任务的线程
                    taskCtx->updateStatus( TaskCtx::taskStop, true );
                }
                else
                {
                    // 再投递
                    taskCtx->rePost();
                }
            }
        }, waitTaskCtx, _taskCtx.get() ) );

        // 添加任务到线程池
        _taskCtx->pool->_postTask(_taskCtx);
    }

    template < typename _Ty2, typename _Fx, typename... _ArgType >
    Task( SharedPointer< TaskCtxT<_Ty2> > waitTaskCtx, _Fx fnRoutine, _ArgType&& ... argRoutine )
    {
        static_assert( std::is_same< ReturnType, typename FuncTraits<_Fx>::ReturnType >::value , "FuncTraits<_Fx>::ReturnType is not match Task<_Ty>." );
        _taskCtx = TaskCtxT<ReturnType>::Create();
        _taskCtx->pool = waitTaskCtx->pool;
        _taskCtx->status = TaskCtx::taskPending;

        auto routine = MakeSimple( NewRunable( fnRoutine, _Ty2(), std::forward<_ArgType>(argRoutine)... ) );
        _taskCtx->poolRunable.attachNew( NewRunable( [routine] ( SharedPointer< TaskCtxT<_Ty2> > waitTaskCtx, TaskCtxT<ReturnType> * taskCtx ) {
            ThreadPool * pool = taskCtx->pool;
            if ( pool->_mode == ThreadPool::modeWaitKeep )
            {
                waitTaskCtx->wait();
                std::get<0>(routine->_tuple) = std::move(waitTaskCtx->val);
                taskCtx->exec( routine.get() );
                // 更新运行状态并通知唤醒等待在此任务的线程
                taskCtx->updateStatus( TaskCtx::taskStop, true );
            }
            else if ( pool->_mode == ThreadPool::modeWaitTimeRePost )
            {
                if ( waitTaskCtx->wait(pool->_durationSec) )
                {
                    std::get<0>(routine->_tuple) = std::move(waitTaskCtx->val);
                    taskCtx->exec( routine.get() );
                    // 更新运行状态并通知唤醒等待在此任务的线程
                    taskCtx->updateStatus( TaskCtx::taskStop, true );
                }
                else
                {
                    // 再投递
                    taskCtx->rePost();
                }
            }
        }, waitTaskCtx, _taskCtx.get() ) );

        // 添加任务到线程池
        _taskCtx->pool->_postTask(_taskCtx);
    }

    template < typename _Ty2, typename _Fx, typename... _ArgType >
    Task( SharedPointer< TaskCtxT<_Ty2> > waitTaskCtx, _Fx fnRoutine, typename FuncTraits<_Fx>::ClassType * obj, _ArgType&& ... argRoutine )
    {
        static_assert( std::is_same< ReturnType, typename FuncTraits<_Fx>::ReturnType >::value , "FuncTraits<_Fx>::ReturnType is not match Task<_Ty>." );
        _taskCtx = TaskCtxT<ReturnType>::Create();
        _taskCtx->pool = waitTaskCtx->pool;
        _taskCtx->status = TaskCtx::taskPending;

        auto routine = MakeSimple( NewRunable( fnRoutine, obj, _Ty2(), std::forward<_ArgType>(argRoutine)... ) );
        _taskCtx->poolRunable.attachNew( NewRunable( [routine] ( SharedPointer< TaskCtxT<_Ty2> > waitTaskCtx, TaskCtxT<ReturnType> * taskCtx ) {
            ThreadPool * pool = taskCtx->pool;
            if ( pool->_mode == ThreadPool::modeWaitKeep )
            {
                waitTaskCtx->wait();
                std::get<1>(routine->_tuple) = std::move(waitTaskCtx->val);
                taskCtx->exec( routine.get() );
                // 更新运行状态并通知唤醒等待在此任务的线程
                taskCtx->updateStatus( TaskCtx::taskStop, true );
            }
            else if ( pool->_mode == ThreadPool::modeWaitTimeRePost )
            {
                if ( waitTaskCtx->wait(pool->_durationSec) )
                {
                    std::get<1>(routine->_tuple) = std::move(waitTaskCtx->val);
                    taskCtx->exec( routine.get() );
                    // 更新运行状态并通知唤醒等待在此任务的线程
                    taskCtx->updateStatus( TaskCtx::taskStop, true );
                }
                else
                {
                    // 再投递
                    taskCtx->rePost();
                }
            }
        }, waitTaskCtx, _taskCtx.get() ) );

        // 添加任务到线程池
        _taskCtx->pool->_postTask(_taskCtx);
    }

    virtual ~Task()
    {
    }

    template < typename _Fx, typename... _ArgType >
    Task<typename FuncTraits<_Fx>::ReturnType> then( _Fx fn, _ArgType&& ... arg )
    {
        return Task<typename FuncTraits<_Fx>::ReturnType>( _taskCtx, fn, std::forward<_ArgType>(arg)... );
    }

    void wait( double sec = -1 )
    {
        _taskCtx->wait(sec);
    }

    ReturnType get()
    {
        return _taskCtx->get();
    }

private:
    SharedPointer< TaskCtxT<ReturnType> > _taskCtx;

    template < typename _Ty0 >
    friend class Task;
};


}

#endif // __THREADS_HPP__