Làm thế nào để làm tăng :: thread_group thực hiện một số cố định của đề song song

Question

Đây là mã để tạo ra một thread_group và thực hiện tất cả các chủ đề song song:

boost::thread_group group; 
for (int i = 0; i < 15; ++i) 
    group.create_thread(aFunctionToExecute); 
group.join_all(); 

Mã này sẽ thực hiện tất cả các chủ đề cùng một lúc. Những gì tôi muốn làm là để thực hiện tất cả, nhưng 4 tối đa song song. Khi kết thúc, một lệnh khác sẽ được thực thi cho đến khi không còn thực thi nữa.

Answer 1

Giải pháp khác, hiệu quả hơn sẽ là có mỗi lần gọi lại luồng đến chuỗi chính khi chúng được hoàn tất và trình xử lý trên chuỗi chính có thể khởi chạy một chuỗi mới mỗi lần. Điều này ngăn các cuộc gọi lặp lại đến timed_join, vì chuỗi chính sẽ không làm bất cứ điều gì cho đến khi cuộc gọi lại được kích hoạt.

Answer 2

Tôi có một cái gì đó như thế này:

boost::mutex mutex_; 
    boost::condition_variable condition_; 
    const size_t throttle_; 
    size_t size_; 
    bool wait_; 
    template <typename Env, class F> 
    void eval_(const Env &env, const F &f) { 
     { 
      boost::unique_lock<boost::mutex> lock(mutex_); 
      size_ = std::min(size_+1, throttle_); 
      while (throttle_ <= size_) condition_.wait(lock); 
     } 
     f.eval(env); 
     { 
      boost::lock_guard<boost::mutex> lock(mutex_); 
      --size_; 
     } 
     condition_.notify_one(); 
    } 

Answer 3

Tôi nghĩ rằng bạn đang tìm kiếm một thực hiện thread_pool, trong đó có sẵn here. Ngoài ra tôi đã nhận thấy rằng nếu bạn tạo một vectơ std :: tương lai và lưu trữ tương lai của nhiều std :: async_tasks trong đó và bạn không có bất kỳ mã chặn nào trong hàm được truyền cho chủ đề, VS2013 (ít nhất là từ những gì tôi có thể xác nhận) sẽ khởi chạy chính xác không có chủ đề thích hợp mà máy của bạn có thể xử lý. Nó reuses các chủ đề khi tạo ra.

Answer 4

Tôi tạo ra giao diện đơn giản hóa riêng của tôi về boost::thread_group để làm công việc này:

class ThreadGroup : public boost::noncopyable 
{ 
    private: 
     boost::thread_group  group; 
     std::size_t    maxSize; 
     float      sleepStart; 
     float      sleepCoef; 
     float      sleepMax; 
     std::set<boost::thread*> running; 

    public: 
     ThreadGroup(std::size_t max_size = 0, 
        float max_sleeping_time = 1.0f, 
        float sleeping_time_coef = 1.5f, 
        float sleeping_time_start = 0.001f) : 
      boost::noncopyable(), 
      group(), 
      maxSize(max_size), 
      sleepStart(sleeping_time_start), 
      sleepCoef(sleeping_time_coef), 
      sleepMax(max_sleeping_time), 
      running() 
     { 
      if(max_size == 0) 
       this->maxSize = (std::size_t)std::max(boost::thread::hardware_concurrency(), 1u); 
      assert(max_sleeping_time >= sleeping_time_start); 
      assert(sleeping_time_start > 0.0f); 
      assert(sleeping_time_coef > 1.0f); 
     } 

     ~ThreadGroup() 
     { 
      this->joinAll(); 
     } 

     template<typename F> boost::thread* createThread(F f) 
     { 
      float sleeping_time = this->sleepStart; 
      while(this->running.size() >= this->maxSize) 
      { 
       for(std::set<boost::thread*>::iterator it = running.begin(); it != running.end();) 
       { 
        const std::set<boost::thread*>::iterator jt = it++; 
        if((*jt)->timed_join(boost::posix_time::milliseconds((long int)(1000.0f * sleeping_time)))) 
         running.erase(jt); 
       } 
       if(sleeping_time < this->sleepMax) 
       { 
        sleeping_time *= this->sleepCoef; 
        if(sleeping_time > this->sleepMax) 
         sleeping_time = this->sleepMax; 
       } 
      } 
      return *this->running.insert(this->group.create_thread(f)).first; 
     } 

     void joinAll() 
     { 
      this->group.join_all(); 
     } 

     void interruptAll() 
     { 
#ifdef BOOST_THREAD_PROVIDES_INTERRUPTIONS 
      this->group.interrupt_all(); 
#endif 
     } 

     std::size_t size() const 
     { 
      return this->group.size(); 
     } 
    }; 

Dưới đây là một ví dụ về sử dụng, rất giống với boost::thread_group với sự khác biệt chính là việc tạo ra các chủ đề là một điểm chờ đợi:

{ 
    ThreadGroup group(4); 
    for(int i = 0; i < 15; ++i) 
    group.createThread(aFunctionToExecute); 
} // join all at destruction