同步任务

Question

我正在开发一个使用线程池的应用程序，向它提交任务并同步它们。主线程必须等到单个循环迭代中所有提交的任务完成，然后再提交另一组任务（因为下一次迭代中的任务操作相同的数据，它们将相互依赖）。

我的问题是，最好的方法是什么？

到目前为止，我想出的是，每个线程在完成一项任务后，都会增加一个原子无符号整数。当整数等于提交的任务数时，主线程继续工作并提交另一轮任务。

这是我的第一个多线程应用程序。 这是处理这个问题的最佳和明智的方法吗？

我正在使用从一本优秀的书“C++ Concurrency in Action: by Anthony Williams”中复制的线程池类。

以下是课程：

class thread_pool
{
    std::atomic_bool done;
    thread_safe_queue<std::function<void()> > work_queue;
    std::vector<std::thread> threads;
    join_threads joiner;

    void worker_thread()
    {
        while(!done)
        {
            std::function<void()> task;
            if(work_queue.try_pop(task))
            {
                task();
            }
            else
            {
                std::this_thread::yield();
            }
        }
    }
public:
    thread_pool():
        done(false),joiner(threads)
    {
        unsigned const thread_count=std::thread::hardware_concurrency();
        try
        {
            for(unsigned i=0;i<thread_count;++i)
            {
                threads.push_back(
                    std::thread(&thread_pool::worker_thread,this));
            }
        }
        catch(...)
        {
            done=true;
            throw;
        }
    }

    ~thread_pool()
    {
        done=true;
    }

    template<typename FunctionType>
    void submit(FunctionType f)
    {
        work_queue.push(std::function<void()>(f));
    }
};

template<typename T>
class threadsafe_queue
{
private:
    mutable std::mutex mut;
    std::queue<T> data_queue;
    std::condition_variable data_cond;
public:
    threadsafe_queue()
    {}

    void push(T new_value)
    {
        std::lock_guard<std::mutex> lk(mut);
        data_queue.push(std::move(new_value));
        data_cond.notify_one();
    }

    void wait_and_pop(T& value)
    {
        std::unique_lock<std::mutex> lk(mut);
        data_cond.wait(lk, [this]{return !data_queue.empty(); });
        value = std::move(data_queue.front());
        data_queue.pop();
    }

    std::shared_ptr<T> wait_and_pop()
    {
        std::unique_lock<std::mutex> lk(mut);
        data_cond.wait(lk, [this]{return !data_queue.empty(); });
        std::shared_ptr<T> res(
            std::make_shared<T>(std::move(data_queue.front())));
        data_queue.pop();
        return res;
    }

    bool try_pop(T& value)
    {
        std::lock_guard<std::mutex> lk(mut);
        if (data_queue.empty())
            return false;
        value = std::move(data_queue.front());
        data_queue.pop();
    }

    std::shared_ptr<T> try_pop()
    {
        std::lock_guard<std::mutex> lk(mut);
        if (data_queue.empty())
            return std::shared_ptr<T>();
        std::shared_ptr<T> res(
            std::make_shared<T>(std::move(data_queue.front())));
        data_queue.pop();
        return res;
    }

    bool empty() const
    {
        std::lock_guard<std::mutex> lk(mut);
        return data_queue.empty();
    }
};

main() 函数：

std::condition_variable waitForThreads;
std::mutex mut;

std::atomic<unsigned> doneCount = 0;

unsigned threadCount = 4; // sample concurrent thread count that I use for testing

 void synchronizeWork()
   {
    doneCount++;
    if (doneCount.load() == threadCount)
    {
        doneCount = 0;
        std::lock_guard<std::mutex> lock(mut);
        waitForThreads.notify_one();
    }
   }

   void Task_A()
   {
    std::cout << "Task A, thread id: " << std::this_thread::get_id() << std::endl;
    std::this_thread::sleep_for(std::chrono::milliseconds(3000));
    synchronizeWork();
   }

int main()
{   
    unsigned const thread_count = std::thread::hardware_concurrency();
    thread_pool threadPool;

    for (int i = 0; i < 1000; ++i)
    {
        for (unsigned j = 0; j < thread_count; j++)
            threadPool.submit(Task_A);

// Below is my way of synchronizing the tasks

        {
            std::unique_lock<std::mutex> lock(mut);
            waitForThreads.wait(lock);
        }

    }

Answer 1

我不熟悉你使用的线程池类。

不使用这样的类，通常的做法是这样的：

  std::cout << "Spawning 3 threads...\n";
  std::thread t1 (pause_thread,1);
  std::thread t2 (pause_thread,2);
  std::thread t3 (pause_thread,3);
  std::cout << "Done spawning threads. Now waiting for them to join:\n";
  t1.join();
  t2.join();
  t3.join();
  std::cout << "All threads joined!\n";

我想任何像样的线程池类都可以让你做同样的事情，更简单的是，给你一个阻塞的方法，直到所有线程都完成。我建议您仔细检查文档。