title: How to make boost::thread_group smaller having boost::asio::io_service::run in its threads?
body: <p>Generally as I've seen it is <a href="http://think-async.com/Asio/Recipes" rel="nofollow">common way to create thread pools via "io_service + thread_group"</a>. Its really great for const size thread pools. Or pools that can only get bigger. But I wonder how to make such pool smaller with out stopping all io_service?</p>
  
  <p>So we have as <a href="http://think-async.com/Asio/Recipes" rel="nofollow">shown</a></p>
  
  <pre><code>// class variables
  asio::io_service io_service;
  boost::thread_group threads;
  asio::io_service::work *work;
  
  // some pool init function
  work = new asio::io_service::work(io_service);
  int cores_number = boost::thread::hardware_concurrency(); 
    for (std::size_t i = 0; i &lt; cores_number; ++i)
      threads.create_thread(boost::bind(&amp;asio::io_service::run, &amp;io_service));
  
  // and now we can simply post tasks
  io_service.post(boost::bind(&amp;class_name::an_expensive_calculation, this, 42));
  io_service.post(boost::bind(&amp;class_name::a_long_running_task, this, 123));
  
  // and it is really eazy to make pool biger - just call (mutexes may be required)
  threads.create_thread(boost::bind(&amp;asio::io_service::run, &amp;io_service));
  </code></pre>
  
  <p>But what if we want to remove threads from our threads pool? we can not just simply call <code>threads.remove_thread(thread* thrd);</code> because it will not stop running in it <code>&amp;asio::io_service::run</code> (IMHO) so I wonder - is it possible and how to really remove threads from such pool? (not just interrupting tham but waiting until current thread task is out of scope)? </p>
  
  <p><em><strong>Update:</em></strong></p>
  
  <p>Here is some simple compilable code: thread pool, with threads desired life time.</p>
  
  <pre><code>#include &lt;stdio.h&gt;
  #include &lt;iostream&gt;
  #include &lt;fstream&gt;
  
  //Boost
  #include &lt;boost/asio.hpp&gt;
  #include &lt;boost/thread.hpp&gt;
  #include &lt;boost/thread/locks.hpp&gt;
  
  boost::asio::io_service io_service;
  boost::asio::io_service::work *work;
  boost::thread_group threads;
  boost::mutex threads_creation;
  int time_limit;
  
  int calculate_the_answer_to_life_the_universe_and_everything(int i)
  {
      boost::this_thread::sleep(boost::posix_time::milliseconds(i));
      std::cout &lt;&lt; i &lt;&lt; std::endl;
      return i;
  }
  
  void run(boost::shared_ptr&lt;boost::thread&gt; thread_ptr)
  {
      try
      {
          io_service.run();
      }
      catch(std::exception &amp;e)
      {
          std::cout &lt;&lt; "exeption: " &lt;&lt; e.what() &lt;&lt; std::endl;
          boost::mutex::scoped_lock lock(threads_creation);
          threads.remove_thread(thread_ptr.get());
          lock.unlock();
          std::cout &lt;&lt; "thread removed from group" &lt;&lt; std::endl;
          return;
      }
  
  }
  
  void pool_item( int i)
  {
      boost::packaged_task&lt;int&gt; pt(boost::bind(calculate_the_answer_to_life_the_universe_and_everything, i));
      boost::unique_future&lt;int&gt; fi=pt.get_future();
  
      boost::thread *task = new boost::thread(std::move(pt)); // launch task on a thread
  
      if(fi.timed_wait(boost::posix_time::milliseconds(time_limit)))
      {
          std::cout &lt;&lt; "sucsess function returned: " &lt;&lt; fi.get() &lt;&lt; std::endl;
      }
      else
      {
          std::cout &lt;&lt; "request took way 2 long!" &lt;&lt; std::endl;
  
          std::cout &lt;&lt; "current group size:" &lt;&lt; threads.size() &lt;&lt; std::endl;
  
          boost::shared_ptr&lt;boost::thread&gt; thread;
          boost::packaged_task&lt;void&gt; pt(boost::bind(run, thread));
          thread = boost::shared_ptr&lt;boost::thread&gt;( new boost::thread(std::move(pt)));
  
          boost::mutex::scoped_lock lock(threads_creation);
          threads.add_thread(thread.get());
          lock.unlock();
  
          task-&gt;join();
  
          throw std::runtime_error("killed joined thread");
  
      }
  }
  
  int main()
  {
      time_limit = 500;
  
      work = new boost::asio::io_service::work(io_service);
      int cores_number = boost::thread::hardware_concurrency(); 
      for (std::size_t i = 0; i &lt; cores_number; ++i)
      {
  
          boost::shared_ptr&lt;boost::thread&gt; thread;
          boost::packaged_task&lt;void&gt; pt(boost::bind(run, thread));
          thread = boost::shared_ptr&lt;boost::thread&gt;( new boost::thread(std::move(pt)));
          threads.add_thread(thread.get());
      }
  
      int i = 800;
      io_service.post(boost::bind(pool_item, i));
  
      boost::this_thread::sleep(boost::posix_time::milliseconds(i*2));
      std::cout &lt;&lt; "thread should be removed by now." &lt;&lt; std::endl
          &lt;&lt; "group size:" &lt;&lt; threads.size() &lt;&lt; std::endl;
  
      std::cin.get();
      return 0;
  }
  </code></pre>
  
  <p>As you can see threads are not removed from threads pool even after <code>.remove_thread(ptr);</code> call.=( Why?</p>
  
  <p><em><strong>Update #2:</em></strong></p>
  
  <p>Well any way I ended up with costume thread group...</p>
  
  <pre><code>#include &lt;stdio.h&gt;
  #include &lt;iostream&gt;
  #include &lt;fstream&gt;
  #include &lt;set&gt;
  
  //Boost
  #include &lt;boost/asio.hpp&gt;
  #include &lt;boost/thread.hpp&gt;
  #include &lt;boost/thread/locks.hpp&gt;
  
  //cf service interface
  //#include &lt;service.hpp&gt;
  
  //cf-server
  //#include &lt;server.h&gt;
  
  #include &lt;boost/foreach.hpp&gt;
  
  class thread_group
  {
  public:
      void add( boost::shared_ptr&lt;boost::thread&gt; to_add)
      {
          boost::mutex::scoped_lock lock(m);
          ds_.insert(to_add);
      }
      void remove( boost::shared_ptr&lt;boost::thread&gt; to_remove)
      {
          boost::mutex::scoped_lock lock(m);
          ds_.erase(to_remove);
      }
  
      int size()
      {
          boost::mutex::scoped_lock lock(m);
          return ds_.size();
      }
  
      void join_all(boost::posix_time::milliseconds interuption_time=boost::posix_time::milliseconds(1000))
      {
          boost::mutex::scoped_lock lock(m);
          BOOST_FOREACH(boost::shared_ptr&lt;boost::thread&gt; t, ds_)
          {
              boost::thread interrupter(boost::bind(&amp;thread_group::interupt_thread, this, t, interuption_time));
          }
      }
  
  private:
      std::set&lt; boost::shared_ptr&lt;boost::thread&gt; &gt; ds_;
      boost::mutex m;
      void interupt_thread(boost::shared_ptr&lt;boost::thread&gt; t, boost::posix_time::milliseconds interuption_time)
      {
          try
          {
              if(!t-&gt;timed_join(interuption_time))
                  t-&gt;interrupt();
  
          }
          catch(std::exception &amp;e)
          {
          }
      }
  };
  
  boost::asio::io_service io_service;
  boost::asio::io_service::work *work;
  thread_group threads;
  int time_limit;
  
  
  
  int calculate_the_answer_to_life_the_universe_and_everything(int i)
  {
      boost::this_thread::sleep(boost::posix_time::milliseconds(i));
      std::cout &lt;&lt; i &lt;&lt; std::endl;
      return i;
  }
  
  void run(boost::shared_ptr&lt;boost::thread&gt; thread_ptr)
  {
      try
      {
          io_service.run();
      }
      catch(std::exception &amp;e)
      {
          std::cout &lt;&lt; "exeption: " &lt;&lt; e.what() &lt;&lt; std::endl;
          threads.remove(thread_ptr);
          std::cout &lt;&lt; "thread removed from group" &lt;&lt; std::endl;
          return;
      }
  
  }
  
  void pool_item( int i)
  {
      boost::packaged_task&lt;int&gt; pt(boost::bind(calculate_the_answer_to_life_the_universe_and_everything, i));
      boost::unique_future&lt;int&gt; fi=pt.get_future();
  
      boost::thread *task = new boost::thread(std::move(pt)); // launch task on a thread
  
      if(fi.timed_wait(boost::posix_time::milliseconds(time_limit)))
      {
          std::cout &lt;&lt; "sucsess function returned: " &lt;&lt; fi.get() &lt;&lt; std::endl;
      }
      else
      {
          std::cout &lt;&lt; "request took way 2 long!" &lt;&lt; std::endl;
  
          std::cout &lt;&lt; "current group size:" &lt;&lt; threads.size() &lt;&lt; std::endl;
          std::cout &lt;&lt; "we want to add thread!" &lt;&lt; std::endl;
          boost::shared_ptr&lt;boost::thread&gt; thread;
          boost::packaged_task&lt;void&gt; pt(boost::bind(run, thread));
          threads.add(thread);
          std::cout &lt;&lt; "thread added" &lt;&lt; std::endl
              &lt;&lt; "current group size:" &lt;&lt; threads.size() &lt;&lt; std::endl;
          task-&gt;join();
  
          throw std::runtime_error("killed joined thread");
  
      }
  }
  
  int main()
  {
      time_limit = 500;
  
      work = new boost::asio::io_service::work(io_service);
      int cores_number = boost::thread::hardware_concurrency(); 
      for (std::size_t i = 0; i &lt; cores_number; ++i)
      {
  
          boost::shared_ptr&lt;boost::thread&gt; thread;
          boost::packaged_task&lt;void&gt; pt(boost::bind(run, thread));
          thread = boost::shared_ptr&lt;boost::thread&gt;( new boost::thread(std::move(pt)));
          threads.add(thread);
      }
  
      int i = 800;
      io_service.post(boost::bind(pool_item, i));
  
      boost::this_thread::sleep(boost::posix_time::milliseconds(i*2));
      std::cout &lt;&lt; "thread should be removed by now." &lt;&lt; std::endl
          &lt;&lt; "group size:" &lt;&lt; threads.size() &lt;&lt; std::endl;
  
      std::cin.get();
      return 0;
  }
  </code></pre>