Merge pull request #590 from glynos/thread_group

deanberris · deanberris · commit 853550b1ebec · 2016-01-31T16:54:03.000+11:00
Thread group
diff --git a/.gitmodules b/.gitmodules
@@ -7,3 +7,6 @@
 [submodule "libs/network/doc/_ext/breathe"]
 	path = libs/network/doc/_ext/breathe
 	url = https://github.com/michaeljones/breathe.git
+[submodule "deps/cxxopts"]
+	path = deps/cxxopts
+	url = https://github.com/jarro2783/cxxopts.git
diff --git a/CMakeLists.txt b/CMakeLists.txt
@@ -46,8 +46,7 @@ add_definitions(-DBOOST_TEST_DYN_LINK)
 # Always use multi-threaded Boost libraries.
 set(Boost_USE_MULTI_THREADED ON)
 
-find_package(Boost 1.57.0
-  REQUIRED system thread filesystem program_options)
+find_package(Boost 1.57.0 REQUIRED system filesystem)
 
 if (CPP-NETLIB_ENABLE_HTTPS)
     find_package( OpenSSL )
diff --git a/boost/network/utils/thread_group.hpp b/boost/network/utils/thread_group.hpp
@@ -0,0 +1,78 @@
+//            Copyright (c) Glyn Matthews 2016.
+//         (C) Copyright 2007-9 Anthony Williams
+// Distributed under the Boost Software License, Version 1.0.
+//    (See accompanying file LICENSE_1_0.txt or copy at
+//          http://www.boost.org/LICENSE_1_0.txt)
+
+#ifndef BOOST_NETWORK_UTILS_THREAD_GROUP_INC
+#define BOOST_NETWORK_UTILS_THREAD_GROUP_INC
+
+#include <thread>
+#include <mutex>
+#include <memory>
+#include <list>
+#include <algorithm>
+
+namespace boost {
+namespace network {
+namespace utils {
+class thread_group {
+ private:
+  thread_group(thread_group const&);
+  thread_group& operator=(thread_group const&);
+
+ public:
+  thread_group() {}
+  ~thread_group() {}
+
+  template <typename F>
+  std::thread* create_thread(F threadfunc) {
+    std::lock_guard<std::mutex> guard(m);
+    std::unique_ptr<std::thread> new_thread(new std::thread(threadfunc));
+    threads.push_back(std::move(new_thread));
+    return threads.back().get();
+  }
+
+  void add_thread(std::thread* thrd) {
+    if (thrd) {
+      std::lock_guard<std::mutex> guard(m);
+      threads.push_back(std::unique_ptr<std::thread>(thrd));
+    }
+  }
+
+  void remove_thread(std::thread* thrd) {
+    std::lock_guard<std::mutex> guard(m);
+    auto const it = std::find_if(threads.begin(), threads.end(),
+                                 [&thrd] (std::unique_ptr<std::thread> &arg) {
+                                   return arg.get() == thrd;
+                                 });
+    if (it != threads.end()) {
+      threads.erase(it);
+    }
+  }
+
+  void join_all() {
+    std::unique_lock<std::mutex> guard(m);
+
+    for (auto &thread : threads) {
+      if (thread->joinable()) {
+        thread->join();
+      }
+    }
+  }
+
+  size_t size() const {
+    std::unique_lock<std::mutex> guard(m);
+    return threads.size();
+  }
+
+ private:
+  std::list<std::unique_ptr<std::thread>> threads;
+  mutable std::mutex m;
+};
+
+}  // namespace utils
+}  // namespace network
+}  // namespace boost
+
+#endif  // BOOST_NETWORK_UTILS_THREAD_GROUP_INC
diff --git a/boost/network/utils/thread_pool.hpp b/boost/network/utils/thread_pool.hpp
@@ -6,21 +6,22 @@
 // (See accompanying file LICENSE_1_0.txt or copy at
 // http://www.boost.org/LICENSE_1_0.txt)
 
+#include <cstddef>
 #include <memory>
 #include <functional>
 #include <boost/asio/io_service.hpp>
 #include <boost/function.hpp>
 #include <boost/network/tags.hpp>
 #include <boost/scope_exit.hpp>
-#include <boost/thread/thread.hpp>
-#include <cstddef>
+//#include <boost/thread/thread.hpp>
+#include <boost/network/utils/thread_group.hpp>
 
 namespace boost {
 namespace network {
 namespace utils {
 
 typedef std::shared_ptr<boost::asio::io_service> io_service_ptr;
-typedef std::shared_ptr<boost::thread_group> worker_threads_ptr;
+typedef std::shared_ptr<utils::thread_group> worker_threads_ptr;
 typedef std::shared_ptr<boost::asio::io_service::work> sentinel_ptr;
 
 template <class Tag>
@@ -46,7 +47,7 @@ struct basic_thread_pool {
         sentinel_.reset();
         io_service_.reset();
         if (worker_threads_.get()) {
-          worker_threads_->interrupt_all();
+          // worker_threads_->interrupt_all();
           worker_threads_->join_all();
         }
         worker_threads_.reset();
@@ -59,7 +60,7 @@ struct basic_thread_pool {
     }
 
     if (!worker_threads_.get()) {
-      worker_threads_.reset(new boost::thread_group);
+      worker_threads_.reset(new utils::thread_group);
     }
 
     if (!sentinel_.get()) {
diff --git a/deps/cxxopts b/deps/cxxopts
@@ -0,0 +1 @@
+Subproject commit aec97a6f53c3486fc51e0d9857f10b683180d668
diff --git a/libs/network/example/CMakeLists.txt b/libs/network/example/CMakeLists.txt
@@ -4,6 +4,7 @@
 #          http://www.boost.org/LICENSE_1_0.txt)
 
 include_directories(${CPP-NETLIB_SOURCE_DIR})
+include_directories(${CPP-NETLIB_SOURCE_DIR}/deps/cxxopts/src)
 if (OPENSSL_FOUND)
   include_directories(${OPENSSL_INCLUDE_DIR})
 endif (OPENSSL_FOUND)
diff --git a/libs/network/example/http/fileserver.cpp b/libs/network/example/http/fileserver.cpp
@@ -4,8 +4,8 @@
 // http://www.boost.org/LICENSE_1_0.txt)
 
 #include <memory>
+#include <thread>
 #include <boost/network/include/http/server.hpp>
-#include <boost/thread.hpp>
 #include <sys/mman.h>
 #include <sys/types.h>
 #include <sys/stat.h>
@@ -28,7 +28,7 @@ struct file_cache {
   std::string doc_root_;
   region_map regions;
   meta_map file_headers;
-  boost::shared_mutex cache_mutex;
+  std::mutex cache_mutex;
 
   explicit file_cache(std::string doc_root) : doc_root_(std::move(doc_root)) {}
 
@@ -39,12 +39,12 @@ struct file_cache {
   }
 
   bool has(std::string const &path) {
-    boost::shared_lock<boost::shared_mutex> lock(cache_mutex);
+    std::unique_lock<std::mutex> lock(cache_mutex);
     return regions.find(doc_root_ + path) != regions.end();
   }
 
   bool add(std::string const &path) {
-    boost::upgrade_lock<boost::shared_mutex> lock(cache_mutex);
+    std::unique_lock<std::mutex> lock(cache_mutex);
     std::string real_filename = doc_root_ + path;
     if (regions.find(real_filename) != regions.end()) return true;
 #ifdef O_NOATIME
@@ -60,7 +60,6 @@ struct file_cache {
       return false;
     }
 
-    boost::upgrade_to_unique_lock<boost::shared_mutex> unique_lock(lock);
     regions.insert(std::make_pair(real_filename, std::make_pair(region, size)));
     static server::response_header common_headers[] = {
         {"Connection", "close"}, {"Content-Type", "x-application/octet-stream"},
@@ -73,7 +72,7 @@ struct file_cache {
   }
 
   std::pair<void *, std::size_t> get(std::string const &path) {
-    boost::shared_lock<boost::shared_mutex> lock(cache_mutex);
+    std::unique_lock<std::mutex> lock(cache_mutex);
     region_map::const_iterator region = regions.find(doc_root_ + path);
     if (region != regions.end())
       return region->second;
@@ -83,14 +82,12 @@ struct file_cache {
 
   boost::iterator_range<std::vector<server::response_header>::iterator> meta(
       std::string const &path) {
-    boost::shared_lock<boost::shared_mutex> lock(cache_mutex);
+    std::unique_lock<std::mutex> lock(cache_mutex);
     static std::vector<server::response_header> empty_vector;
     auto headers = file_headers.find(doc_root_ + path);
     if (headers != file_headers.end()) {
-      auto begin = headers->second
-                                                                 .begin(),
-                                                     end =
-                                                         headers->second.end();
+      auto begin = headers->second.begin(),
+           end = headers->second.end();
       return boost::make_iterator_range(begin, end);
     } else
       return boost::make_iterator_range(empty_vector);
diff --git a/libs/network/example/http/hello_world_async_server_with_work_queue.cpp b/libs/network/example/http/hello_world_async_server_with_work_queue.cpp
diff --git a/libs/network/example/http_client.cpp b/libs/network/example/http_client.cpp
