Пожалуйста покритикуйте мою имплементацию самого самого распостраненного паттерна многопоточного программирования Boss Worker!

Что интересует: концептульные просчёты, ненужные плюшки и есть ли аналогичкая имплементация в майнстримных библиотеках С++? Т.к., например, оный паттерн становиться ненужен с С#, там можно оч. удобно закодить через PLINQ

#include <boost/bind.hpp>
#include <boost/thread/mutex.hpp>
#include <boost/thread/thread.hpp>

namespace util {
  
  template<class QueueItemType, class ProductType>
    struct Worker
    {
      virtual ProductType proceed(QueueItemType) = 0;
      virtual ~Worker() {}
    };

  template<class QueueItemType,
           class ProductType, 
           class WorkerIterator,
           class QueueItemIterator,
           class ProductOutputIterator
           >
    /**
     * The Boss is motivating workers to act in parallel
     */
    class Boss 
    {
      typedef Worker<QueueItemType, ProductType> TheWorker;
    
      WorkerIterator workersBegin_; 
      WorkerIterator workersEnd_;

      QueueItemIterator queueBegin_; 
      QueueItemIterator queueEnd_;

      ProductOutputIterator productOutputIterator_;
    public:
      
      Boss(WorkerIterator workersBegin, 
           WorkerIterator workersEnd,
           
           QueueItemIterator queueBegin, 
           QueueItemIterator queueEnd,
           
           ProductOutputIterator productOutputIterator)
      
        : workersBegin_(workersBegin)
        , workersEnd_(workersEnd)
        , queueBegin_(queueBegin)
        , queueEnd_(queueEnd)
        , productOutputIterator_(productOutputIterator)
      {
      }
      
    void getThingsDone()
    {
      if( workersBegin_ == workersEnd_ )
        return; // as the input is, such is the output :)
      
      boost::thread_group threads;
      for(WorkerIterator workerIt = workersBegin_; workerIt != workersEnd_; ++workerIt)
      {
        TheWorker &worker = *workerIt;
        
        threads.create_thread( boost::bind(&Boss::loadWorker, this, boost::ref(worker)) );
      }
      
      threads.join_all();
    }
      
    private:
      boost::mutex popQueueItemMutex_;
      boost::mutex collectProductMutex_;
      
      void loadWorker(TheWorker& worker)
      {
        while(QueueItemType *item = popQueueItem())
        {
          ProductType product = worker.proceed(*item);
          
          collectProduct( product );
        }
      }
      
      /**
       * Thread-safe dequeue 
       * 
       * @return null if the end of queue
       */
      QueueItemType* popQueueItem()
      {
        boost::mutex::scoped_lock scopedLock(popQueueItemMutex_);
        
        if (queueBegin_ == queueEnd_)
          return NULL;
        else
          return &* (queueBegin_++);
      }
      
      /**
       * Thread-safe product collect
       */
      void collectProduct(ProductType product)
      {
        boost::mutex::scoped_lock scopedLock(collectProductMutex_);
        
        *(productOutputIterator_++) = product;
      }
      
    }; // class Boss
    
} // namespace util

И спасибо!

Здравствуйте, Olksy, Вы писали:

O>Пожалуйста покритикуйте мою имплементацию самого самого распостраненного паттерна многопоточного программирования Boss Worker!

На вскидку:

Количестно тасок worker'ами не должно быть в общем случае известно заранее,
т.е. боссу нужно передавать не queueBegin и queueEnd, а например объект, предоставляющий
поток тасок на выполнение.

Зачем создавать все worker'ы, лучше создавать по необходимости. Например, зачем создавать
сто потоков, когда с решением конкретной задачи справятся два-три.

С формированием результата неочевидно. Может лучше сделать объект,
принимающий поток результатов обработки.
tasks >> Boss(workers) >> results.

Здравствуйте, Olksy, Вы писали:

O>Пожалуйста покритикуйте мою имплементацию самого самого распостраненного паттерна многопоточного программирования Boss Worker!

O>Что интересует: концептульные просчёты, ненужные плюшки и есть ли аналогичкая имплементация в майнстримных библиотеках С++? Т.к., например, оный паттерн становиться ненужен с С#, там можно оч. удобно закодить через PLINQ

по-моему лучше взять boost::asio::io_service в качестве исполнителя задач (а возможно и вообще в качестве босса), и ограничить максимальное количество потоков, обычно их не требуется больше, чем ядер в системе.

Здравствуйте, XuMuK, Вы писали:

XMK>по-моему лучше взять boost::asio::io_service в качестве исполнителя задач (а возможно и вообще в качестве босса), и ограничить максимальное количество потоков, обычно их не требуется больше, чем ядер в системе.

Спасибо, попробую

Здравствуйте, 5er, Вы писали:

5er>Здравствуйте, Olksy, Вы писали:

O>>Пожалуйста покритикуйте мою имплементацию самого самого распостраненного паттерна многопоточного программирования Boss Worker!

5er>На вскидку:

5er>Количестно тасок worker'ами не должно быть в общем случае известно заранее,
5er>т.е. боссу нужно передавать не queueBegin и queueEnd, а например объект, предоставляющий
5er>поток тасок на выполнение.

По поводу очереди [queueBegin, queueEnd), через эти два можно передать очередь данных без фактических ограничеинй, т.е. закодить её в виде потоковых итераторов.

5er>Зачем создавать все worker'ы, лучше создавать по необходимости. Например, зачем создавать
5er>сто потоков, когда с решением конкретной задачи справятся два-три.

Согласен с замечанием по поводу воркеров. В самом деле, несообразно создавать over 9000 их если тасков всего пусть будет 2.

5er>С формированием результата неочевидно. Может лучше сделать объект,
5er>принимающий поток результатов обработки.
5er>tasks >> Boss(workers) >> results.

productOutputIterator тож может представлять собой что угодно, но использует всё ту же семантику stl-итераторов.

Спасибо за коменты!!!