threadtask.hpp

浏览该文件的文档.

#pragma once
//
// threadtask 提供任务和线程池相关功能
//

namespace winux
{

template < typename _Ty >
class Task;
class ThreadPool;

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

    Mutex mtxTask; 
    Condition cdtTask; 
    TaskStatus status; 
    ThreadPool * pool; 
    bool posted; 
    bool aborted; 

    WeakPointer<TaskCtx> weakThis; 
    SimplePointer<Runable> routineForPool; 

    TaskCtx * prevTask; 
    SharedPointer<TaskCtx> nextTask; 

    TaskCtx * getStartTask()
    {
        TaskCtx * start = this;
        while ( start->prevTask != nullptr ) start = start->prevTask;
        return start;
    }

    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 post();

    void tryPostNext();
protected:
    TaskCtx() : mtxTask(true), cdtTask(true), status(taskPending), pool(nullptr), posted(false), aborted(false), prevTask(nullptr) { }
    virtual ~TaskCtx()
    {
        //auto start = getStartTask();
        //cout << "~TaskCtx(" << ( start == this ? "start-task" : "then-task" ) << ") " << this << "\n";
    }
};

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 ) noexcept
    {
        try
        {
            this->val = ivk->invoke();
        }
        catch ( winux::Error const & e )
        {
            std::cout << e.what() << std::endl;
            this->aborted = true;
        }
        catch ( ... )
        {
            std::cout << "unknown" << std::endl;
            this->aborted = true;
        }
    }
protected:
    TaskCtxT( ThreadPool * pool, TaskCtx::TaskStatus status = TaskCtx::taskPending )
    {
        this->pool = pool;
        this->status = status;
    }
};

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 ) noexcept
    {
        try
        {
            ivk->invoke();
        }
        catch ( winux::Error const & e )
        {
            std::cout << e.what() << std::endl;
            this->aborted = true;
        }
        catch ( ... )
        {
            std::cout << "unknown" << std::endl;
            this->aborted = true;
        }
    }
protected:
    TaskCtxT( ThreadPool * pool, TaskCtx::TaskStatus status = TaskCtx::taskPending )
    {
        this->pool = pool;
        this->status = status;
    }
};

class ThreadPool
{
public:
    explicit ThreadPool() : _mtxPool(true), _cdtPool(true), _poolStop(false), _taskChainCount(0)
    {
    }

    explicit ThreadPool( int threadCount ) : _mtxPool(true), _cdtPool(true), _poolStop(false), _taskChainCount(0)
    {
        this->startup(threadCount);
    }

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

    ThreadPool & startup( int threadCount )
    {
        _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->routineForPool->run();

                    taskCtx->posted = false;
                }
            }
        } );
        _group.startup(); // 启动线程组的线程
        return *this;
    }

    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()
    {
        while ( true )
        {
            ScopeGuard guard(_mtxPool);
            // 判断任务队列为空
            _cdtPool.waitUntil( [this] () { return _queueTask.empty(); }, _mtxPool );
            // 判断任务链数量为0
            if ( this->_taskChainCount <= 0 ) break;
        }

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

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

    int incTaskChainCount()
    {
        ScopeGuard guard(_mtxPool);
        return ++_taskChainCount;
    }

    int decTaskChainCount()
    {
        ScopeGuard guard(_mtxPool);
        return --_taskChainCount;
    }

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

    Mutex _mtxPool; // 互斥锁
    Condition _cdtPool; // 用于判断队列是否有数据
    bool _poolStop; // 线程池停止
    ThreadGroup _group; // 线程组
    std::queue< SharedPointer<TaskCtx> > _queueTask; // 任务队列
    int _taskChainCount; // 执行中的任务链数量

    friend struct TaskCtx;

    DISABLE_OBJECT_COPY(ThreadPool)
};

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

inline void TaskCtx::tryPostNext()
{
    // 检测是否有下个任务且任务不中止，则投递到线程池
    if ( this->nextTask && !this->aborted )
    {
        this->nextTask->post();
    }
    else
    {
        this->pool->decTaskChainCount();
    }
}

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(pool, TaskCtx::taskPending);
        //cout << "start-task: " << _taskCtx.get() << endl;

        auto routine = MakeSimple( NewRunable( fnRoutine, std::forward<_ArgType>(argRoutine)... ) );
        _taskCtx->routineForPool.attachNew( NewRunable( [routine] ( TaskCtxT<ReturnType> * taskCtx ) {
            // 执行任务例程
            taskCtx->exec( routine.get() );

            // 检测是否有下个任务，投递到线程池
            taskCtx->tryPostNext();

            // 更新运行状态并通知唤醒等待在此任务的线程
            taskCtx->updateStatus( TaskCtx::taskStop, true );

        }, _taskCtx.get() ) );
    }

    template < typename _Fx, typename... _ArgType >
    Task( SharedPointer< TaskCtxT<void> > prevTaskCtx, _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(prevTaskCtx->pool, TaskCtx::taskPending);
        _taskCtx->prevTask = prevTaskCtx.get();
        //cout << "then 2-1 " << endl;

        auto routine = MakeSimple( NewRunable( fnRoutine, std::forward<_ArgType>(argRoutine)... ) );
        _taskCtx->routineForPool.attachNew( NewRunable( [routine] ( SharedPointer< TaskCtxT<void> > prevTaskCtx, TaskCtxT<ReturnType> * taskCtx ) {
            // 执行任务例程
            taskCtx->exec( routine.get() );

            // 后续任务的例程执行完成后没有必要还留有上一个任务的nextTask，必须清空，否则内存泄露。
            prevTaskCtx->nextTask.reset();

            // 检测是否有下个任务，投递到线程池
            taskCtx->tryPostNext();

            // 更新运行状态并通知唤醒等待在此任务的线程
            taskCtx->updateStatus( TaskCtx::taskStop, true );

        }, prevTaskCtx, _taskCtx.get() ) );

    }

    template < typename _Fx, typename... _ArgType >
    Task( SharedPointer< TaskCtxT<void> > prevTaskCtx, _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(prevTaskCtx->pool, TaskCtx::taskPending);
        _taskCtx->prevTask = prevTaskCtx.get();
        //cout << "then 2-2 " << endl;

        auto routine = MakeSimple( NewRunable( fnRoutine, obj, std::forward<_ArgType>(argRoutine)... ) );
        _taskCtx->routineForPool.attachNew( NewRunable( [routine] ( SharedPointer< TaskCtxT<void> > prevTaskCtx, TaskCtxT<ReturnType> * taskCtx ) {
            // 执行任务例程
            taskCtx->exec( routine.get() );

            // 后续任务的例程执行完成后没有必要还留有上一个任务的nextTask，必须清空，否则内存泄露。
            prevTaskCtx->nextTask.reset();

            // 检测是否有下个任务，投递到线程池
            taskCtx->tryPostNext();

            // 更新运行状态并通知唤醒等待在此任务的线程
            taskCtx->updateStatus( TaskCtx::taskStop, true );

        }, prevTaskCtx, _taskCtx.get() ) );

    }

    template < typename _Ty2, typename _Fx, typename... _ArgType >
    Task( SharedPointer< TaskCtxT<_Ty2> > prevTaskCtx, _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(prevTaskCtx->pool, TaskCtx::taskPending);
        _taskCtx->prevTask = prevTaskCtx.get();
        //cout << "then 3-1 " << endl;

        auto routine = MakeSimple( NewRunable( fnRoutine, _Ty2(), std::forward<_ArgType>(argRoutine)... ) );
        _taskCtx->routineForPool.attachNew( NewRunable( [routine] ( SharedPointer< TaskCtxT<_Ty2> > prevTaskCtx, TaskCtxT<ReturnType> * taskCtx ) {
            // 执行任务例程
            std::get<0>(routine->_tuple) = std::move(prevTaskCtx->val);
            taskCtx->exec( routine.get() );

            // 后续任务的例程执行完成后没有必要还留有上一个任务的nextTask，必须清空，否则内存泄露。
            prevTaskCtx->nextTask.reset();

            // 检测是否有下个任务，投递到线程池
            taskCtx->tryPostNext();

            // 更新运行状态并通知唤醒等待在此任务的线程
            taskCtx->updateStatus( TaskCtx::taskStop, true );

        }, prevTaskCtx, _taskCtx.get() ) );

    }

    template < typename _Ty2, typename _Fx, typename... _ArgType >
    Task( SharedPointer< TaskCtxT<_Ty2> > prevTaskCtx, _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(prevTaskCtx->pool, TaskCtx::taskPending);
        _taskCtx->prevTask = prevTaskCtx.get();
        //cout << "then 3-2 " << endl;

        auto routine = MakeSimple( NewRunable( fnRoutine, obj, _Ty2(), std::forward<_ArgType>(argRoutine)... ) );
        _taskCtx->routineForPool.attachNew( NewRunable( [routine] ( SharedPointer< TaskCtxT<_Ty2> > prevTaskCtx, TaskCtxT<ReturnType> * taskCtx ) {
            // 执行任务例程
            std::get<1>(routine->_tuple) = std::move(prevTaskCtx->val);
            taskCtx->exec( routine.get() );

            // 后续任务的例程执行完成后没有必要还留有上一个任务的nextTask，必须清空，否则内存泄露。
            prevTaskCtx->nextTask.reset();

            // 检测是否有下个任务，投递到线程池
            taskCtx->tryPostNext();

            // 更新运行状态并通知唤醒等待在此任务的线程
            taskCtx->updateStatus( TaskCtx::taskStop, true );

        }, prevTaskCtx, _taskCtx.get() ) );

    }

    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)... );
    }

    Task & post()
    {
        // 寻到起始任务
        TaskCtx * startTask = _taskCtx->getStartTask();
        if ( !startTask->posted )
        {
            TaskCtx * cur = nullptr;

            // 修改任务状态为taskPending
            cur = _taskCtx.get();
            while ( cur != nullptr )
            {
                ScopeGuard guard(cur->mtxTask);
                cur->status = TaskCtx::taskPending;
                cur = cur->prevTask;
            }

            // 用prev链构建next链
            cur = _taskCtx.get();
            while ( cur->prevTask != nullptr )
            {
                auto p = cur;
                cur = cur->prevTask;
                cur->nextTask = p->weakThis.lock();
            }

            // 投递起始任务
            startTask->post();

            startTask->posted = true;

            // 增加任务链数目
            this->_taskCtx->pool->incTaskChainCount();
        }

        return *this;
    }

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

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

private:
    SharedPointer< TaskCtxT<ReturnType> > _taskCtx; 

    template < typename _Ty0 >
    friend class Task;
};

} // namespace winux