Skip to content

Thread group #590

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
merged 3 commits into from
Jan 31, 2016
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
3 changes: 3 additions & 0 deletions .gitmodules
Original file line number Diff line number Diff line change
Expand Up @@ -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
3 changes: 1 addition & 2 deletions CMakeLists.txt
Original file line number Diff line number Diff line change
Expand Up @@ -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 )
Expand Down
78 changes: 78 additions & 0 deletions boost/network/utils/thread_group.hpp
Original file line number Diff line number Diff line change
@@ -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
11 changes: 6 additions & 5 deletions boost/network/utils/thread_pool.hpp
Original file line number Diff line number Diff line change
Expand Up @@ -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>
Expand All @@ -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();
Expand All @@ -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()) {
Expand Down
1 change: 1 addition & 0 deletions deps/cxxopts
Submodule cxxopts added at aec97a
1 change: 1 addition & 0 deletions libs/network/example/CMakeLists.txt
Original file line number Diff line number Diff line change
Expand Up @@ -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)
Expand Down
19 changes: 8 additions & 11 deletions libs/network/example/http/fileserver.cpp
Original file line number Diff line number Diff line change
Expand Up @@ -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>
Expand All @@ -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)) {}

Expand All @@ -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
Expand All @@ -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"},
Expand All @@ -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;
Expand All @@ -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);
Expand Down
Loading