How to add a Manager #107

Activity

1. Zzzzzya

   changed the title [-]How to add a[/-] [+]How to add a Manager[/+] on Oct 30, 2023
2. Zzzzzya commented on Oct 30, 2023

   

   Zzzzzya

   on Oct 30, 2023

   Author

   More actions

   in this demo ,the num of threads in the pool is fixed when the pool is created. Is there a more flexible algorithm or strategy ? How can i add a manager. I mean , in real project ,what should i add to the threadpool? Sorry that my English may be not very good. /(ㄒoㄒ)/~~
3. xiays146 commented on Apr 2, 2024

   

   xiays146

   on Apr 2, 2024

   More actions

   maybe you can fix the following Start func to your own:
   `
   #ifndef THREAD_POOL_H
   #define THREAD_POOL_H

   #include
   #include
   #include
   #include
   #include
   #include <condition_variable>
   #include
   #include
   #include

   class ThreadPool
   {
   public:
   ThreadPool(size_t);
   template <class F, class... Args>
   auto enqueue(F &&f, Args &&...args) -> std::future<typename std::result_of<F(Args...)>::type>;
   ~ThreadPool();
   ThreadPool(const ThreadPool &) = delete;
   ThreadPool(ThreadPool &&) = delete;
   ThreadPool &operator=(const ThreadPool &) = delete;
   ThreadPool &operator=(ThreadPool &&) = delete;
   void Stop()
   {
   if (stop)
   return;
   {
   std::unique_lockstd::mutex lock(queue_mutex);
   stop = true;
   }
   condition.notify_all();
   for (std::thread &worker : workers) {
   if (worker.joinable())
   worker.join();
   }
   }

   void Start(size_t threads)
   {
       if (!stop)
           return;
       {
           std::unique_lock<std::mutex> lock(queue_mutex);
           stop = false;
       }
       // TODO: clear workers
       for (size_t i = 0; i < threads; ++i) {
           workers.emplace_back([this] {
               for (;;) {
                   std::function<void()> task;
   
                   {
                       std::unique_lock<std::mutex> lock(this->queue_mutex);
                       this->condition.wait(
                           lock, [this] { return this->stop || !this->tasks.empty(); });
                       if (this->stop && this->tasks.empty()) {
                           return;
                       }
                       task = std::move(this->tasks.front());
                       this->tasks.pop();
                   }
   
                   task();
               }
           });
       }
   }
   

   private:
   // need to keep track of threads so we can join them
   std::vectorstd::thread workers;
   // the task queue
   std::queue<std::function<void()>> tasks;

   // synchronization
   std::mutex queue_mutex;
   std::condition_variable condition;
   bool stop;
   

   };

   // the constructor just launches some amount of workers
   inline ThreadPool::ThreadPool(size_t threads) : stop(true)
   {
   Start(threads);
   }

   // add new work item to the pool
   template <class F, class... Args>
   auto ThreadPool::enqueue(F &&f, Args &&...args)
   -> std::future<typename std::result_of<F(Args...)>::type>
   {
   using return_type = typename std::result_of<F(Args...)>::type;

   auto task = std::make_shared<std::packaged_task<return_type()>>(
       std::bind(std::forward<F>(f), std::forward<Args>(args)...));
   
   std::future<return_type> res = task->get_future();
   {
       std::unique_lock<std::mutex> lock(queue_mutex);
   
       // don't allow enqueueing after stopping the pool
       if (stop)
           throw std::runtime_error("enqueue on stopped ThreadPool");
   
       tasks.emplace([task]() { (*task)(); });
   }
   condition.notify_one();
   return res;
   

   }

   // the destructor joins all threads
   inline ThreadPool::~ThreadPool()
   {
   Stop();
   }

   #endif
   `