premiere
/
premiere-libtorrent
2
2
Fork 1
Code Issues Pull Requests Releases Wiki Activity

remove network thread pool feature (#688) 

remove network thread pool feature (it was messy and mostly useful for bittorrent over SSL). The proper solution is to allow every peer run in its own asio strand and have multiple threads running the io service
This commit is contained in:
Arvid Norberg 2016-05-02 02:18:51 -04:00
parent 8ae1e9a21d
commit 0d2c04cb2e
10 changed files with 25 additions and 412 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

2
include/libtorrent/Makefile.am
View File

@ -76,7 +76,6 @@ nobase_include_HEADERS = \
magnet_uri.hpp \
max.hpp \
natpmp.hpp \
network_thread_pool.hpp \
operations.hpp \
packet_buffer.hpp \
parse_url.hpp \
@ -128,7 +127,6 @@ nobase_include_HEADERS = \
storage_defs.hpp \
tailqueue.hpp \
string_util.hpp \
thread_pool.hpp \
time.hpp \
timestamp_history.hpp \
torrent_handle.hpp \

8
include/libtorrent/aux_/session_impl.hpp
View File

@ -86,7 +86,6 @@ POSSIBILITY OF SUCH DAMAGE.
#include "libtorrent/bloom_filter.hpp"
#include "libtorrent/peer_class.hpp"
#include "libtorrent/disk_io_job.hpp" // block_cache_reference
#include "libtorrent/network_thread_pool.hpp"
#include "libtorrent/peer_class_type_filter.hpp"
#include "libtorrent/kademlia/dht_observer.hpp"
#include "libtorrent/resolver.hpp"
@ -622,8 +621,6 @@ namespace libtorrent
// implements uncork_interface
virtual void do_delayed_uncork() override;
void post_socket_job(socket_job& j) override;
// implements session_interface
virtual tcp::endpoint bind_outgoing_socket(socket_type& s, address
const& remote_address, error_code& ec) const override;
@ -661,7 +658,6 @@ namespace libtorrent
void update_queued_disk_bytes();
void update_alert_queue_size();
void update_disk_threads();
void update_network_threads();
void update_cache_buffer_chunk_size();
void update_report_web_seed_downloads();
void update_outgoing_interfaces();
@ -758,10 +754,6 @@ namespace libtorrent
// constructed after it.
disk_io_thread m_disk_thread;
// a thread pool used for async_write_some calls,
// to distribute its cost to multiple threads
std::vector<std::unique_ptr<network_thread_pool>> m_net_thread_pool;
// the bandwidth manager is responsible for
// handing out bandwidth to connections that
// asks for it, it can also throttle the

6
include/libtorrent/aux_/session_interface.hpp
View File

@ -64,7 +64,6 @@ namespace libtorrent
{
class peer_connection;
class torrent;
struct socket_job;
#ifndef TORRENT_NO_DEPRECATE
struct pe_settings;
#endif
@ -215,11 +214,6 @@ namespace libtorrent { namespace aux
virtual boost::uint16_t listen_port() const = 0;
virtual boost::uint16_t ssl_listen_port() const = 0;
// TODO: 2 factor out the thread pool for socket jobs into a separate
// class
// used to (potentially) issue socket write calls onto multiple threads
virtual void post_socket_job(socket_job& j) = 0;
// load the specified torrent. also evict one torrent, except
// for the one specified, if we are at the limit of loaded torrents
virtual bool load_torrent(torrent* t) = 0;

83
include/libtorrent/network_thread_pool.hpp
View File

@ -1,83 +0,0 @@
/*
Copyright (c) 2011-2016, Arvid Norberg
All rights reserved.
Redistribution and use in source and binary forms, with or without
modification, are permitted provided that the following conditions
are met:
* Redistributions of source code must retain the above copyright
notice, this list of conditions and the following disclaimer.
* Redistributions in binary form must reproduce the above copyright
notice, this list of conditions and the following disclaimer in
the documentation and/or other materials provided with the distribution.
* Neither the name of the author nor the names of its
contributors may be used to endorse or promote products derived
from this software without specific prior written permission.
THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS "AS IS"
AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE
IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR PURPOSE
ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT OWNER OR CONTRIBUTORS BE
LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL, SPECIAL, EXEMPLARY, OR
CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT LIMITED TO, PROCUREMENT OF
SUBSTITUTE GOODS OR SERVICES; LOSS OF USE, DATA, OR PROFITS; OR BUSINESS
INTERRUPTION) HOWEVER CAUSED AND ON ANY THEORY OF LIABILITY, WHETHER IN
CONTRACT, STRICT LIABILITY, OR TORT (INCLUDING NEGLIGENCE OR OTHERWISE)
ARISING IN ANY WAY OUT OF THE USE OF THIS SOFTWARE, EVEN IF ADVISED OF THE
POSSIBILITY OF SUCH DAMAGE.
*/
#ifndef TORRENT_NETWORK_THREAD_POOL_HPP_INCLUDED
#define TORRENT_NETWORK_THREAD_POOL_HPP_INCLUDED
#include "libtorrent/thread_pool.hpp"
#include <boost/shared_ptr.hpp>
#include <vector>
namespace libtorrent
{
class peer_connection;
class buffer;
struct socket_job
{
socket_job() : type(none), vec(NULL), recv_buf(NULL), buf_size(0) {}
#if __cplusplus >= 201103L
socket_job(socket_job const&) = default;
socket_job& operator=(socket_job const&) = default;
#endif
enum job_type_t
{
read_job = 0,
write_job,
none
};
job_type_t type;
// used for write jobs
std::vector<boost::asio::const_buffer> const* vec;
// used for read jobs
char* recv_buf;
int buf_size;
std::array<boost::asio::mutable_buffer, 2> read_vec;
boost::shared_ptr<peer_connection> peer;
// defined in session_impl.cpp
~socket_job();
};
// defined in session_impl.cpp
struct network_thread_pool : thread_pool<socket_job>
{
void process_job(socket_job const& j, bool post);
};
}
#endif // TORRENT_NETWORK_THREAD_POOL_HPP_INCLUDED

2
include/libtorrent/peer_connection.hpp
View File

@ -76,7 +76,6 @@ POSSIBILITY OF SUCH DAMAGE.
#include <array>
#include <boost/optional.hpp>
#include <boost/cstdint.hpp>
#include <boost/pool/pool.hpp>
#include "libtorrent/aux_/disable_warnings_pop.hpp"
@ -276,7 +275,6 @@ namespace libtorrent
, public boost::enable_shared_from_this<peer_connection>
{
friend class invariant_access;
friend struct network_thread_pool;
friend class torrent;
public:

2
include/libtorrent/settings_pack.hpp
View File

@ -1410,6 +1410,8 @@ namespace libtorrent
aio_threads,
aio_max,
// .. note:: This is not implemented
//
// ``network_threads`` is the number of threads to use to call
// ``async_write_some`` (i.e. send) on peer connection sockets. When
// seeding at extremely high rates, this may become a bottleneck, and

169
include/libtorrent/thread_pool.hpp
View File

@ -1,169 +0,0 @@
/*
Copyright (c) 2011-2016, Arvid Norberg
All rights reserved.
Redistribution and use in source and binary forms, with or without
modification, are permitted provided that the following conditions
are met:
* Redistributions of source code must retain the above copyright
notice, this list of conditions and the following disclaimer.
* Redistributions in binary form must reproduce the above copyright
notice, this list of conditions and the following disclaimer in
the documentation and/or other materials provided with the distribution.
* Neither the name of the author nor the names of its
contributors may be used to endorse or promote products derived
from this software without specific prior written permission.
THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS "AS IS"
AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE
IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR PURPOSE
ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT OWNER OR CONTRIBUTORS BE
LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL, SPECIAL, EXEMPLARY, OR
CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT LIMITED TO, PROCUREMENT OF
SUBSTITUTE GOODS OR SERVICES; LOSS OF USE, DATA, OR PROFITS; OR BUSINESS
INTERRUPTION) HOWEVER CAUSED AND ON ANY THEORY OF LIABILITY, WHETHER IN
CONTRACT, STRICT LIABILITY, OR TORT (INCLUDING NEGLIGENCE OR OTHERWISE)
ARISING IN ANY WAY OUT OF THE USE OF THIS SOFTWARE, EVEN IF ADVISED OF THE
POSSIBILITY OF SUCH DAMAGE.
*/
#ifndef TORRENT_THREAD_POOL
#define TORRENT_THREAD_POOL
#include "libtorrent/config.hpp"
#include <thread>
#include <atomic>
#include <condition_variable>
#include <deque>
#include <vector>
#include <boost/shared_ptr.hpp>
#include <boost/bind.hpp>
namespace libtorrent
{
template <class T>
struct thread_pool
{
thread_pool() : m_num_threads(0) {}
virtual ~thread_pool() {}
void stop() { set_num_threads(0, true); }
void set_num_threads(int n, bool wait = true)
{
if (n == m_num_threads) return;
if (n > m_num_threads)
{
while (m_num_threads < n)
{
++m_num_threads;
m_threads.emplace_back(&thread_pool::thread_fun
, this, int(m_num_threads)-1);
}
}
else
{
while (m_num_threads > n) { --m_num_threads; }
{
std::lock_guard<std::mutex> l(m_mutex);
m_cond.notify_all();
}
if (wait)
{
for (int i = m_num_threads; i < int(m_threads.size()); ++i)
{
m_threads[i].join();
}
}
// this will detach the threads
m_threads.resize(m_num_threads);
}
}
// returns true if the job was posted, and false if it was
// processed immediately
bool post_job(T& e)
{
if (m_num_threads == 0)
{
// if we don't have any worker threads
// just do the work immediately
process_job(e, false);
return false;
}
else
{
retain_job(e);
std::lock_guard<std::mutex> l(m_mutex);
m_queue.push_back(e);
// we only need to signal if the threads
// may have been put to sleep. If the size
// previous to adding the new job was > 0
// they don't need waking up.
if (m_queue.size() == 1)
m_cond.notify_one();
return true;
}
}
protected:
virtual void process_job(T const& j, bool post) = 0;
virtual void retain_job(T&) {}
private:
void thread_fun(int thread_id)
{
for (;;)
{
std::unique_lock<std::mutex> l(m_mutex);
while (m_queue.empty() && thread_id < m_num_threads) m_cond.wait(l);
// if the number of wanted thread is decreased,
// we may stop this thread
// when we're terminating the last hasher thread (id=0), make sure
// we finish up all queud jobs first
if ((thread_id != 0 || m_queue.empty()) && thread_id >= m_num_threads) break;
TORRENT_ASSERT(!m_queue.empty());
T e = m_queue.front();
m_queue.pop_front();
l.unlock();
process_job(e, true);
}
#ifdef TORRENT_DEBUG
if (thread_id == 0)
{
// when we're terminating the last hasher thread, make sure
// there are no more scheduled jobs
std::lock_guard<std::mutex> l(m_mutex);
TORRENT_ASSERT(m_queue.empty());
}
#endif
}
// the std::mutex only protects m_cond and m_queue
// all other members are only used from a single
// thread (the user of this class, i.e. the disk
// thread).
std::mutex m_mutex;
std::condition_variable m_cond;
std::deque<T> m_queue;
std::vector<std::thread> m_threads;
// this is a counter which is atomically incremented
// by each thread as it's started up, in order to
// assign a unique id to each thread
std::atomic<int> m_num_threads;
};
}
#endif // TORRENT_THREAD_POOL

94
src/peer_connection.cpp
View File

@ -71,7 +71,6 @@ POSSIBILITY OF SUCH DAMAGE.
#include "libtorrent/torrent.hpp"
#include "libtorrent/peer_info.hpp"
#include "libtorrent/bt_peer_connection.hpp"
#include "libtorrent/network_thread_pool.hpp"
#include "libtorrent/error.hpp"
#include "libtorrent/alloca.hpp"
#include "libtorrent/disk_interface.hpp"
@ -5661,20 +5660,8 @@ namespace libtorrent
m_socket_is_writing = true;
#endif
// uTP sockets aren't thread safe...
if (is_utp(*m_socket))
{
m_socket->async_write_some(vec, make_write_handler(boost::bind(
&peer_connection::on_send_data, self(), _1, _2)));
}
else
{
socket_job j;
j.type = socket_job::write_job;
j.vec = &vec;
j.peer = self();
m_ses.post_socket_job(j);
}
m_socket->async_write_some(vec, make_write_handler(boost::bind(
&peer_connection::on_send_data, self(), _1, _2)));
m_channel_state[upload_channel] |= peer_info::bw_network;
}
@ -5803,42 +5790,20 @@ namespace libtorrent
// utp sockets aren't thread safe...
ADD_OUTSTANDING_ASYNC("peer_connection::on_receive_data");
if (is_utp(*m_socket))
if (num_bufs == 1)
{
if (num_bufs == 1)
{
TORRENT_ASSERT(boost::asio::buffer_size(vec[0]) > 0);
m_socket->async_read_some(
boost::asio::mutable_buffers_1(vec[0]), make_read_handler(
boost::bind(&peer_connection::on_receive_data, self(), _1, _2)));
}
else
{
TORRENT_ASSERT(boost::asio::buffer_size(vec[0])
+ boost::asio::buffer_size(vec[1])> 0);
m_socket->async_read_some(
vec, make_read_handler(
boost::bind(&peer_connection::on_receive_data, self(), _1, _2)));
}
TORRENT_ASSERT(boost::asio::buffer_size(vec[0]) > 0);
m_socket->async_read_some(
boost::asio::mutable_buffers_1(vec[0]), make_read_handler(
boost::bind(&peer_connection::on_receive_data, self(), _1, _2)));
}
else
{
socket_job j;
j.type = socket_job::read_job;
j.peer = self();
if (num_bufs == 1)
{
TORRENT_ASSERT(boost::asio::buffer_size(vec[0]) > 0);
j.recv_buf = boost::asio::buffer_cast<char*>(vec[0]);
j.buf_size = int(boost::asio::buffer_size(vec[0]));
}
else
{
TORRENT_ASSERT(boost::asio::buffer_size(vec[0])
+ boost::asio::buffer_size(vec[1])> 0);
j.read_vec = vec;
}
m_ses.post_socket_job(j);
TORRENT_ASSERT(boost::asio::buffer_size(vec[0])
+ boost::asio::buffer_size(vec[1])> 0);
m_socket->async_read_some(
vec, make_read_handler(
boost::bind(&peer_connection::on_receive_data, self(), _1, _2)));
}
return 0;
}
@ -6027,35 +5992,20 @@ namespace libtorrent
boost::asio::mutable_buffer buffer = m_recv_buffer.reserve(int(buffer_size));
TORRENT_ASSERT(m_recv_buffer.normalized());
// utp sockets aren't thread safe...
if (is_utp(*m_socket))
{
bytes_transferred = m_socket->read_some(boost::asio::mutable_buffers_1(buffer), ec);
bytes_transferred = m_socket->read_some(boost::asio::mutable_buffers_1(buffer), ec);
if (ec)
if (ec)
{
if (ec == boost::asio::error::try_again
|| ec == boost::asio::error::would_block)
{
if (ec == boost::asio::error::try_again
|| ec == boost::asio::error::would_block)
{
// allow reading from the socket again
TORRENT_ASSERT(m_channel_state[download_channel] & peer_info::bw_network);
m_channel_state[download_channel] &= ~peer_info::bw_network;
setup_receive();
return;
}
disconnect(ec, op_sock_read);
// allow reading from the socket again
TORRENT_ASSERT(m_channel_state[download_channel] & peer_info::bw_network);
m_channel_state[download_channel] &= ~peer_info::bw_network;
setup_receive();
return;
}
}
else
{
ADD_OUTSTANDING_ASYNC("peer_connection::on_receive_data");
socket_job j;
j.type = socket_job::read_job;
j.recv_buf = boost::asio::buffer_cast<char*>(buffer);
j.buf_size = int(boost::asio::buffer_size(buffer));
j.peer = self();
m_ses.post_socket_job(j);
disconnect(ec, op_sock_read);
return;
}

69
src/session_impl.cpp
View File

@ -201,35 +201,6 @@ namespace libtorrent {
std::mutex _async_ops_mutex;
#endif
socket_job::~socket_job() {}
void network_thread_pool::process_job(socket_job const& j, bool post)
{
TORRENT_UNUSED(post);
if (j.type == socket_job::write_job)
{
TORRENT_ASSERT(j.peer->m_socket_is_writing);
j.peer->get_socket()->async_write_some(
*j.vec, j.peer->make_write_handler(boost::bind(
&peer_connection::on_send_data, j.peer, _1, _2)));
}
else
{
if (j.recv_buf)
{
j.peer->get_socket()->async_read_some(boost::asio::buffer(j.recv_buf, j.buf_size)
, j.peer->make_read_handler(boost::bind(
&peer_connection::on_receive_data, j.peer, _1, _2)));
}
else
{
j.peer->get_socket()->async_read_some(j.read_vec
, j.peer->make_read_handler(boost::bind(
&peer_connection::on_receive_data, j.peer, _1, _2)));
}
}
}
namespace aux {
void session_impl::init_peer_class_filter(bool unlimited_local)
@ -596,7 +567,6 @@ namespace aux {
update_connections_limit();
update_unchoke_limit();
update_disk_threads();
update_network_threads();
update_upnp();
update_natpmp();
update_lsd();
@ -6076,45 +6046,6 @@ namespace aux {
m_disk_thread.set_num_threads(m_settings.get_int(settings_pack::aio_threads));
}
void session_impl::update_network_threads()
{
// TODO: 3 this is a mess
int num_threads = m_settings.get_int(settings_pack::network_threads);
int num_pools = num_threads > 0 ? num_threads : 1;
while (num_pools > m_net_thread_pool.size())
{
m_net_thread_pool.emplace_back(new network_thread_pool());
m_net_thread_pool.back()->set_num_threads(num_threads > 0 ? 1 : 0);
}
while (num_pools < m_net_thread_pool.size())
{
m_net_thread_pool.erase(m_net_thread_pool.end() - 1);
}
if (num_threads == 0 && m_net_thread_pool.size() > 0)
{
m_net_thread_pool[0]->set_num_threads(0);
}
}
void session_impl::post_socket_job(socket_job& j)
{
uintptr_t idx = 0;
if (m_net_thread_pool.size() > 1)
{
// each peer needs to be pinned to a specific thread
// since reading and writing simultaneously on the same
// socket from different threads is not supported by asio.
// as long as a specific socket is consistently used from
// the same thread, it's safe
idx = uintptr_t(j.peer.get());
idx ^= idx >> 8;
idx %= m_net_thread_pool.size();
}
m_net_thread_pool[idx]->post_job(j);
}
void session_impl::update_cache_buffer_chunk_size()
{
if (m_settings.get_int(settings_pack::cache_buffer_chunk_size) <= 0)

2
src/settings_pack.cpp
View File

@ -323,7 +323,7 @@ namespace libtorrent
SET(predictive_piece_announce, 0, 0),
SET(aio_threads, 4, &session_impl::update_disk_threads),
SET(aio_max, 300, 0),
SET(network_threads, 0, &session_impl::update_network_threads),
SET(network_threads, 0, 0),
DEPRECATED_SET(ssl_listen, 0, &session_impl::update_ssl_listen),
SET(tracker_backoff, 250, 0),
SET_NOPREV(share_ratio_limit, 200, 0),