Std::thread приостановка потока

// < асинхронный запуск потоков потоков 
std::future<void> ft1 = std::async(std::launch::async, ...);
std::future<void> ft2 = std::async(std::launch::async, ...);
// < ждем завершения потоков
ft1.get();
ft2.get();
// < идем дальше

ft1.get();
ft2.get();

#include "thread_pool/atomic_stream.h"
#include "thread_pool/thread_pool.h"
 
// --- в качестве задачи просто будем выводить в консоль данные
void Worker()
{
    const auto id 
        = std::this_thread::get_id();
 
    for(size_t n = 0; n<10; ++n)
        // --- атомик-стрим используется исключительно в учебных целях
        // его использование сводит на нет весь перфоманс
        // от использования потоков
        // поэтому, в боевых условиях его не нужно использовать
        service::AStream() << "[" << id << "]" << n << "\n";
}
 
// --- эту функцию будем вызывать для каждого потока
// в момент его завершения
void Stop()
{
    const auto id
        = std::this_thread::get_id();
    service::AStream() << "[" << id << "] finished\n";
}
 
// --- последняя задача для тред-пула
// она должна просигналить пулу о том,
// что нужно остановиться
void Finish()
{
    service::AStream() << "[ALL TASK FINISHED]\n";
 
    // --- иногда по факту завершению работы потока
    // необходимо выполнить дополнительную функцию-зачистки
    // например, если требуется доступ к TLS (thread local storage) данным
    // в этом случае можно указать дополнительную функцию
    // которая будет вызвана для каждого остановившегося потока
 
    // если такая зачистка не нужна
    // то можно вызвать Stop() без аргументов
    service::ThreadPool::Get().Stop(::Stop);
}
 
int main()
{
    std::cout<<"hello\n";
 
    // --- получаем доступ к статическому тред-пулу
    auto& pool = service::ThreadPool::Get();
 
    // --- забиваем в тред-пул 100 задач
    for(size_t n=0; n<100; ++n)
        pool.AddTask(Worker);
 
    // тред пул будет работать хоть до бесконечности,
    // пока какой либо из тредов не просигналит об остановке
    // поэтому, добавляем последнею задачу,
    // которая пошлет сигнал остановки
    pool.AddTask(Finish);
 
    // запуск и ожидание завершения работы
    // всех тредов
    pool.Start();
 
    std::cout<<"success\n";
}

#pragma once
 
#include <iostream>
#include <sstream>
#include <mutex>
 
//-----------------------------------------------------------------------
// атомарный вывод данных в консоль
//-----------------------------------------------------------------------
 
namespace service{
 
    template<class T>
    class BasicAtomicStream
    {
        typedef std::mutex 
            mutex_type;
 
        typedef typename T::char_type
            char_type;
 
        typedef std::basic_ostringstream<char_type>
            buffer_type;
    public:
        BasicAtomicStream(T& s)
            : mMutex()
            , mStream()
            , mDst(s)
        {}
 
       ~BasicAtomicStream()
        {
            mMutex.lock();
            mDst << mStream.str();
            mMutex.unlock();
        }
 
        template<class U> 
        buffer_type& operator << (U&& v)
        { 
            mStream << v;
            return mStream;
        }
 
    private:
        mutex_type  
            mMutex;
 
        buffer_type 
            mStream;
 
        T& mDst;
    };
 
    template<class T> 
    class AtomicStream 
        : public BasicAtomicStream<T>
    {
        typedef BasicAtomicStream<T>
            base;
 
        AtomicStream(T& os)
            :base(os)
        {}
 
    };
 
    template<>
    class AtomicStream<std::ostream>
        : public BasicAtomicStream<std::ostream>
    {
        typedef BasicAtomicStream<std::ostream>
            base;
    public:
        AtomicStream(std::ostream& os = std::cout)
            :base(os)
        {}
    };
 
    template<>
    class AtomicStream<std::wostream>
        : public BasicAtomicStream<std::wostream>
    {
        typedef BasicAtomicStream<std::wostream>
            base;
    public:
        AtomicStream(std::wostream& os = std::wcout)
            :base(os)
        {}
    };
 
    typedef AtomicStream<std::ostream>
        AStream;
    typedef AtomicStream<std::wostream>
        WAStream;
 
 
}//namespace service

#pragma once
 
#include <functional>
#include <atomic>
#include <mutex>
#include <queue>
 
 
struct TaskQueue final 
{
    typedef ::std::function<void()> 
        Task;
 
   ~TaskQueue(){}
    TaskQueue();
 
    // добавляйте std::function/функцию/лямбду/функтор
    template<class T>
    void AddTask(T&& task)
    {
        ::std::lock_guard<decltype(mGuard)>
            lock(mGuard);
 
        mTasks.push(task);
    }
 
    // запускает 1 задачу
    // вернёт true, если ещё остались задачи
    bool RunOne(); 
 
    // запускает все задачи
    size_t Run() ; 
 
    // очищает очередь задач
    void Clear() ; 
 
    // true, если очередь задач пуста
    bool Empty()const; 
private:
    //удалены
    TaskQueue(const TaskQueue&) ;
    TaskQueue& operator=(const TaskQueue&);
    
private:
    ::std::queue<Task>   
        mTasks;
    mutable ::std::mutex 
        mGuard;
};

#include "thread_pool/task_queue.h"
 
 
TaskQueue::TaskQueue()
    :mTasks()
    ,mGuard()
{}
 
bool TaskQueue::RunOne()
{
    Task task;
    {
        ::std::lock_guard<decltype(mGuard)> 
            lock(mGuard);
        if(!mTasks.empty())
            task = mTasks.front(),
            mTasks.pop();
        else
            return false;
    }
    return task(), 
        true;
}
 
size_t TaskQueue::Run()
{
    size_t count = 0;
    while(RunOne())
        ++count;
    return count;
}
 
void TaskQueue::Clear()
{
    ::std::lock_guard<decltype(mGuard)>
        lock(mGuard);
    while(!mTasks.empty())
        mTasks.pop();
}
 
bool TaskQueue::Empty()const
{
    ::std::lock_guard<decltype(mGuard)> 
        lock(mGuard);
    return mTasks.empty();
}

#pragma once
 
#include "task_queue.h"
 
#include <condition_variable>
#include <thread>
#include <vector>
 
 
namespace service{
 
    struct ThreadPool
    {
        typedef std::function<void()>
            Task;
 
        //если 0 - количество потоков 
        //будет определено автоматически
        ThreadPool(const size_t num_threads = 0);
 
        // добавляйте std::function/функцию/лямбду/функтор
        template<class T>
        ThreadPool& AddTask(T&& task) 
        { 
            mTaskQueue.AddTask(task); 
 
            // --- если пул уже запущен
            // то при добавлении 1 задачи, будем 1 поток
            if(mStarted)
                mCondition.notify_one();
            return *this;
        }
 
        // эта задача будет выполнена для каждого треда
        // по завершению работы тред-пула
        // укажите std::function/функцию/лямбду/функтор
        template<class T>
        ThreadPool& SetFinished(T&& task)
        {
            ::std::lock_guard<decltype(mGuard)>
                lock(mGuard);
            mTaskStop = task;
            return *this;
        }
 
        // сигнал остановки работы тред пула
        // по завершению работы, тред запустит эту задачу
        // (std::function/функцию/лямбду/функтор)
        template<class T> void Stop(T&& task)
        {
            SetFinished( std::forward<T>(task)  );
            Stop();
        }
 
        // запуск обработки задач
        // и ожидание завершения всех потоков
        void Start();
 
        // потоки доделывают текущие задачи и завершаются
        void Stop();
 
        // тред-пул сингелтон
        static ThreadPool& 
            Get(const size_t num_threads=0);
 
    private:
        // ждать завершения работы всех потоков
        void Wait();
 
        std::atomic<bool>
            mStarted;
        std::atomic<bool>
            mStopped;
        ::std::mutex
            mGuard;
        Task
            mTaskStop;
        TaskQueue
            mTaskQueue;
        std::condition_variable 
            mCondition;
        std::vector<std::thread> 
            mThreads;
    };
 
}//namespace service

#include "thread_pool/thread_pool.h"
 
namespace service{
 
    ThreadPool::ThreadPool(const size_t num_threads)
        :mStarted(false)
        ,mStopped(false)
        ,mTaskStop()
        ,mTaskQueue()
        ,mCondition()
        ,mThreads()
    {
 
        // --- если количество тредов 0
        // тогда будет выбрано оптимальное количество
        // (равное количеству ядер процессора)
        const auto num = num_threads==0?
                std::thread::hardware_concurrency():
                num_threads;
 
        mThreads.reserve(num);
        for(size_t n=0; n<num; ++n)
            mThreads.emplace_back([this, n]()
            { 
                std::mutex mutex;
 
                // --- если тред-пул ещё не стартовал, 
                // --- то тред засыпает
                if(!mStarted)
                {
                    std::unique_lock<std::mutex>
                        locker(mutex);
                    mCondition.wait(locker);
                }
 
                // до тех пор, пока не будет получен сигнал о завершении
                while(!mStopped)
                {
                    std::unique_lock<std::mutex>
                        locker(mutex);
 
                    // --- тред извлекает и исполняет очередную задачу
                    const bool empty = !mTaskQueue.RunOne();
 
                    // --- если по окончанию выполнения очередной задачи
                    // был получен сигнал остановки,
                    // то тред завершается
                    if(mStopped)
                        break;
 
                    // --- если мы здесь, значит сигнала остановки ещё не было
                    if(empty)
                        // --- если задача была последней - тред засыпает
                        mCondition.wait(locker);
                }
 
                // --- если мы здесь, значит был получен сигнал остановки
                // если была указана функция зачистки
                // тред запускает её, и полностью завершается
                if(mTaskStop)
                    mTaskStop();
            });
    }
 
    void ThreadPool::Start()
    { 
        mStarted = true;
        // --- запускает все треды на выполнение
        // и ожидает их завершения
        mCondition.notify_all(); 
        Wait();
    }
 
    void ThreadPool::Stop()
    {
        // --- выставляет флаг остановки,
        // --- и пробуждает все спящие треды
        // --- что бы проснувшись, они корректно завершились
        mStopped  = true;
        mCondition.notify_all();
    }
 
    void ThreadPool::Wait()
    {
        // --- ждем завершения всех тредов
        for(auto& i: mThreads)
            i.join();
    }
 
    ThreadPool& ThreadPool::Get(const size_t num_threads)
    {
        // --- thread-safe singelton
        static ThreadPool p(num_threads);
        return p;
    }
 
 
}//namespace service

private:
    //удалены
    TaskQueue(const TaskQueue&) ;
    TaskQueue& operator=(const TaskQueue&);

#include <iostream>
#include <thread>
#include <mutex>
#include <condition_variable>
#include <vector>
 
using namespace std;
mutex m0,m1,m2,m3,m4;
condition_variable c1,c2;
vector <int> vInt;
bool b1 = false;
bool b2 = false;
int b0;
 
void gen() {
    unique_lock<mutex> lk1(m3);
    for (int i=1; i<6; ++i) {
        if (i>1) { // в перовой итерации цикла пропускаем ожидание окончания работы потоков ex1, ex2
            while (b0 < 2) {
                c2.wait(lk1); //ожидание окончания работы потоков ex1, ex2
            };
        };
        m0.lock();
            vInt.push_back(1);
        m0.unlock();
 
        m4.lock();
            b0 = 0;
            b1=true;
            b2=true;
            c1.notify_all(); //разблокируем потоки ex1, ex2
        m4.unlock();
 
    };
};
 
void ex1() {
    unique_lock<mutex> lk1(m1);
    for (int i = 1; i<6; ++i) {
        while (!b1) {
            c1.wait(lk1); //ожидание окончания работы потока gen
        };
        for (int i = 1; i<10; ++i) {
            m0.lock();
                vInt.push_back(2);//тут должна быть функция1 выполняемая в этом потоке
            m0.unlock();
        };
        m4.lock();
            b1=false;
            b0++;
            if (b0 == 2) //проверяем что этот поток последний завершает работу
                c2.notify_one(); //разблокируем поток gen
        m4.unlock();
    };
};
 
void ex2() {
    unique_lock<mutex> lk1(m2);
    for (int i = 1; i<6; ++i) {
        while (!b2) {
            c1.wait(lk1); //ожидание окончания работы потока gen
        };
        for (int i = 1; i<10; ++i) {
            m0.lock();
                vInt.push_back(3); //тут должна быть функция (не такая как в потоке ex1) выполняемая в этом потоке
            m0.unlock();
        };
        m4.lock();
            b2=false;
            b0++;
            if (b0 == 2) //проверяем что этот поток последний завершает работу
                c2.notify_one(); //разблокируем поток gen
        m4.unlock();
    };
};
 
int main()
{
    thread tgen(gen);
    thread t1(ex1);
    thread t2(ex2);
    tgen.join();
    t1.join();
    t2.join();
 
    for (vector <int>::iterator it=vInt.begin(); it != vInt.end(); ++it){
        cout << *it;
    };
 
    return 0;
}

template<class T>
    void AddTask(T&& task)
    {
        ::std::lock_guard<decltype(mGuard)>
            lock(mGuard);
 
        mTasks.push(task);
    }

template<class T>
    void AddTask(T&& task)
    {
        ::std::lock_guard<decltype(mGuard)>
            lock(mGuard);
 
        mTasks.push(  std::forward<T>(task) );
    }

void push (const value_type& val);

// --- если тред-пул ещё не стартовал, 
// --- то тред засыпает
if(!mStarted)
{
    // error C2664: 'void std::condition_variable::wait(std::unique_lock<std::mutex> &)' : 
    // cannot convert argument 1 from 'std::lock_guard<std::mutex>' 
    // to 'std::unique_lock<std::mutex> &'
 
    //std::unique_lock<std::mutex>
    ::std::lock_guard<std::mutex>
        locker(mutex);
    mCondition.wait(locker);
}

pool.RunOne();
pool.Run();

RunOne(pool);  
Run(pool);

pool.RunOne();  
Run(pool);

#pragma once
 
#include "external/task_queue.h"
 
namespace sample
{
    typedef service::TaskQueue 
        TaskQueue;
}

sample::TaskQueue taskqueue;