diff --git a/doc/rtorrent.1.xml b/doc/rtorrent.1.xml index 0e4c2401..671fa138 100644 --- a/doc/rtorrent.1.xml +++ b/doc/rtorrent.1.xml @@ -845,8 +845,9 @@ be nessesary in case of filesystem bugs like NFS in linux ~2.6.13. max_open_files = value -Number of files to simultaneously keep open. Libtorrent dynamically -opens and closes files when mapping files to memory. Defaults to 128. +Number of files to simultaneously keep open. LibTorrent dynamically +opens and closes files as necessary when mapping files to +memory. Default is based on sysconf(_SC_OPEN_MAX). @@ -855,9 +856,18 @@ opens and closes files when mapping files to memory. Defaults to 128. max_open_sockets = value -Number of sockets to simultaneously keep open. This value is set to be -sysconf(_SC_OPEN_MAX) - 256 at startup. This -gives the client 128 sockets to use as it wishes. +Number of network sockets to simultaneously keep open. This value is +set to a reasonable value based on sysconf(_SC_OPEN_MAX). + + + + + + max_open_http = value + + +Number of sockets to simultaneously keep open. This value is set +to 32 by default. diff --git a/src/core/curl_get.cc b/src/core/curl_get.cc index 6d1465ba..1c0d4c3c 100644 --- a/src/core/curl_get.cc +++ b/src/core/curl_get.cc @@ -37,9 +37,9 @@ #include "config.h" #include -#include #include #include +#include #include "curl_get.h" #include "curl_stack.h" @@ -54,30 +54,17 @@ curl_get_receive_write(void* data, size_t size, size_t nmemb, void* handle) { return 0; } -CurlGet::CurlGet(CurlStack* s) : - m_handle(NULL), - m_stack(s) { - - if (m_stack == NULL) - throw std::logic_error("Tried to create CurlGet without a valid CurlStack"); -} - CurlGet::~CurlGet() { close(); } -CurlGet* -CurlGet::new_object(CurlStack* s) { - return new CurlGet(s); -} - void CurlGet::start() { if (is_busy()) - throw std::logic_error("Tried to call CurlGet::start on a busy object"); + throw torrent::internal_error("Tried to call CurlGet::start on a busy object."); if (m_stream == NULL) - throw std::logic_error("Tried to call CurlGet::start without a valid output stream"); + throw torrent::internal_error("Tried to call CurlGet::start without a valid output stream."); m_handle = curl_easy_init(); @@ -127,32 +114,4 @@ CurlGet::size_total() { return d; } -void -CurlGet::set_user_agent(const char* s) { - curl_easy_setopt(m_handle, CURLOPT_USERAGENT, s); -} - -void -CurlGet::set_http_proxy(const char* s) { - curl_easy_setopt(m_handle, CURLOPT_PROXY, s); -} - -void -CurlGet::set_bind_address(const char* s) { - curl_easy_setopt(m_handle, CURLOPT_INTERFACE, s); -} - -void -CurlGet::perform(CURLMsg* msg) { - if (msg->msg != CURLMSG_DONE) - throw std::logic_error("CurlGet::process got CURLMSG that isn't done"); - - if (msg->data.result == CURLE_OK) - m_signalDone.emit(); - else - m_signalFailed.emit(curl_easy_strerror(msg->data.result)); - - // Do nothing below after emitting the signals. -} - } diff --git a/src/core/curl_get.h b/src/core/curl_get.h index 0db760b1..517cc6f6 100644 --- a/src/core/curl_get.h +++ b/src/core/curl_get.h @@ -40,47 +40,38 @@ #include #include #include -#include #include - -struct CURLMsg; +#include namespace core { class CurlStack; class CurlGet : public torrent::Http { - public: - friend class CurlStack; - - CurlGet(CurlStack* s); +public: + CurlGet(CurlStack* s) : m_active(false), m_handle(NULL), m_stack(s) {} virtual ~CurlGet(); - static CurlGet* new_object(CurlStack* s); - void start(); void close(); - bool is_busy() { return m_handle; } + bool is_busy() const { return m_handle; } + bool is_active() const { return m_active; } + + void set_active(bool a) { m_active = a; } double size_done(); double size_total(); - void set_user_agent(const char* s); - void set_http_proxy(const char* s); - void set_bind_address(const char* s); + CURL* handle() { return m_handle; } - protected: - CURL* handle() { return m_handle; } - - void perform(CURLMsg* msg); - - private: +private: CurlGet(const CurlGet&); void operator = (const CurlGet&); - CURL* m_handle; + bool m_active; + CURL* m_handle; CurlStack* m_stack; }; diff --git a/src/core/curl_stack.cc b/src/core/curl_stack.cc index 22a61a08..ec1189d9 100644 --- a/src/core/curl_stack.cc +++ b/src/core/curl_stack.cc @@ -37,7 +37,6 @@ #include "config.h" #include -#include #include #include #include @@ -50,39 +49,51 @@ namespace core { CurlStack::CurlStack() : m_handle((void*)curl_multi_init()), - m_size(0) { + m_active(0), + m_maxActive(32) { } CurlStack::~CurlStack() { - while (!m_getList.empty()) - m_getList.front()->close(); + while (!empty()) + front()->close(); curl_multi_cleanup((CURLM*)m_handle); } +CurlGet* +CurlStack::new_object() { + return new CurlGet(this); +} + void CurlStack::perform() { - int s; CURLMcode code; do { - code = curl_multi_perform((CURLM*)m_handle, &s); + int count; + code = curl_multi_perform((CURLM*)m_handle, &count); if (code > 0) - throw std::runtime_error("Error calling curl_multi_perform"); + throw torrent::internal_error("Error calling curl_multi_perform."); - if (s != m_size) { + if ((unsigned int)count != size()) { // Done with some handles. int t; CURLMsg* msg; while ((msg = curl_multi_info_read((CURLM*)m_handle, &t)) != NULL) { - CurlGetList::iterator itr = std::find_if(m_getList.begin(), m_getList.end(), rak::equal(msg->easy_handle, std::mem_fun(&CurlGet::handle))); + if (msg->msg != CURLMSG_DONE) + throw torrent::internal_error("CurlStack::perform() msg->msg != CURLMSG_DONE."); - if (itr == m_getList.end()) - throw std::logic_error("Could not find CurlGet with the right easy_handle"); + iterator itr = std::find_if(begin(), end(), rak::equal(msg->easy_handle, std::mem_fun(&CurlGet::handle))); + + if (itr == end()) + throw torrent::internal_error("Could not find CurlGet with the right easy_handle."); - (*itr)->perform(msg); + if (msg->data.result == CURLE_OK) + (*itr)->signal_done().emit(); + else + (*itr)->signal_failed().emit(curl_easy_strerror(msg->data.result)); } } @@ -94,7 +105,7 @@ CurlStack::fdset(fd_set* readfds, fd_set* writefds, fd_set* exceptfds) { int maxFd = 0; if (curl_multi_fdset((CURLM*)m_handle, readfds, writefds, exceptfds, &maxFd) != 0) - throw std::runtime_error("Error calling curl_multi_fdset"); + throw torrent::internal_error("Error calling curl_multi_fdset."); return std::max(maxFd, 0); } @@ -102,38 +113,54 @@ CurlStack::fdset(fd_set* readfds, fd_set* writefds, fd_set* exceptfds) { void CurlStack::add_get(CurlGet* get) { if (!m_userAgent.empty()) - get->set_user_agent(m_userAgent.c_str()); + curl_easy_setopt(get->handle(), CURLOPT_USERAGENT, m_userAgent.c_str()); if (!m_httpProxy.empty()) - get->set_http_proxy(m_httpProxy.c_str()); + curl_easy_setopt(get->handle(), CURLOPT_PROXY, m_httpProxy.c_str()); if (!m_bindAddress.empty()) - get->set_bind_address(m_bindAddress.c_str()); + curl_easy_setopt(get->handle(), CURLOPT_INTERFACE, m_bindAddress.c_str()); - CURLMcode code; + base_type::push_back(get); - if ((code = curl_multi_add_handle((CURLM*)m_handle, get->handle())) > 0) - throw std::logic_error("curl_multi_add_handle \"" + std::string(curl_multi_strerror(code))); + if (m_active >= m_maxActive) + return; - m_size++; - m_getList.push_back(get); - - // Curl ML suggest we need to do perform after adding a handle. - //perform(); + m_active++; + get->set_active(true); + + if (curl_multi_add_handle((CURLM*)m_handle, get->handle()) > 0) + throw torrent::internal_error("Error calling curl_multi_add_handle."); } void CurlStack::remove_get(CurlGet* get) { + iterator itr = std::find(begin(), end(), get); + + if (itr == end()) + throw torrent::internal_error("Could not find CurlGet when calling CurlStack::remove."); + + base_type::erase(itr); + + // The CurlGet object was never activated, so we just skip this one. + if (!get->is_active()) + return; + + get->set_active(false); + if (curl_multi_remove_handle((CURLM*)m_handle, get->handle()) > 0) - throw std::logic_error("Error calling curl_multi_remove_handle"); + throw torrent::internal_error("Error calling curl_multi_remove_handle."); - CurlGetList::iterator itr = std::find(m_getList.begin(), m_getList.end(), get); + if (m_active == m_maxActive && + (itr = std::find_if(begin(), end(), std::not1(std::mem_fun(&CurlGet::is_active)))) != end()) { + (*itr)->set_active(true); - if (itr == m_getList.end()) - throw std::logic_error("Could not find CurlGet when calling CurlStack::remove"); + if (curl_multi_add_handle((CURLM*)m_handle, (*itr)->handle()) > 0) + throw torrent::internal_error("Error calling curl_multi_add_handle."); - m_size--; - m_getList.erase(itr); + } else { + m_active--; + } } void @@ -146,9 +173,4 @@ CurlStack::global_cleanup() { curl_global_cleanup(); } -CurlStack::SlotFactory -CurlStack::get_http_factory() { - return sigc::bind(sigc::ptr_fun(&CurlGet::new_object), this); -} - } diff --git a/src/core/curl_stack.h b/src/core/curl_stack.h index 6800b2f2..39bb15b5 100644 --- a/src/core/curl_stack.h +++ b/src/core/curl_stack.h @@ -37,7 +37,7 @@ #ifndef RTORRENT_CORE_CURL_STACK_H #define RTORRENT_CORE_CURL_STACK_H -#include +#include #include #include @@ -45,31 +45,56 @@ namespace core { class CurlGet; -class CurlStack { +// By using a deque instead of vector we allow for cheaper removal of +// the oldest elements, those that will be first in the in the +// deque. +// +// This should fit well with the use-case of a http stack, thus +// we get most of the cache locality benefits of a vector with fast +// removal of elements. + +class CurlStack : std::deque { public: friend class CurlGet; - typedef std::list CurlGetList; - typedef sigc::slot0 SlotFactory; + typedef std::deque base_type; + + using base_type::value_type; + using base_type::iterator; + using base_type::const_iterator; + using base_type::reverse_iterator; + using base_type::const_reverse_iterator; + + using base_type::begin; + using base_type::end; + using base_type::rbegin; + using base_type::rend; + + using base_type::back; + using base_type::front; + + using base_type::size; + using base_type::empty; CurlStack(); ~CurlStack(); - int get_size() const { return m_size; } - bool is_busy() const { return !m_getList.empty(); } + CurlGet* new_object(); void perform(); // TODO: Set fd_set's only once? unsigned int fdset(fd_set* readfds, fd_set* writefds, fd_set* exceptfds); - SlotFactory get_http_factory(); + unsigned int active() const { return m_active; } + unsigned int max_active() const { return m_maxActive; } + void set_max_active(unsigned int a) { m_maxActive = a; } - const std::string& user_agent() const { return m_userAgent; } - void set_user_agent(const std::string& s) { m_userAgent = s; } + const std::string& user_agent() const { return m_userAgent; } + void set_user_agent(const std::string& s) { m_userAgent = s; } - const std::string& http_proxy() const { return m_httpProxy; } - void set_http_proxy(const std::string& s) { m_httpProxy = s; } + const std::string& http_proxy() const { return m_httpProxy; } + void set_http_proxy(const std::string& s) { m_httpProxy = s; } const std::string& bind_address() const { return m_bindAddress; } void set_bind_address(const std::string& s) { m_bindAddress = s; } @@ -87,8 +112,8 @@ class CurlStack { void* m_handle; - int m_size; - CurlGetList m_getList; + unsigned int m_active; + unsigned int m_maxActive; std::string m_userAgent; std::string m_httpProxy; diff --git a/src/core/http_queue.cc b/src/core/http_queue.cc index 2dd70302..744ab4ea 100644 --- a/src/core/http_queue.cc +++ b/src/core/http_queue.cc @@ -51,12 +51,9 @@ namespace core { HttpQueue::iterator HttpQueue::insert(const std::string& url, std::iostream* s) { std::auto_ptr h(m_slotFactory()); - //std::auto_ptr s(new std::stringstream); h->set_url(url); - //h->set_stream(s.get()); h->set_stream(s); - h->set_user_agent("rtorrent/" VERSION); iterator itr = Base::insert(end(), h.get()); @@ -66,8 +63,6 @@ HttpQueue::insert(const std::string& url, std::iostream* s) { (*itr)->start(); h.release(); - //s.release(); - m_signalInsert.emit(*itr); return itr; @@ -77,9 +72,7 @@ void HttpQueue::erase(iterator itr) { m_signalErase.emit(*itr); - //delete (*itr)->get_stream(); delete *itr; - Base::erase(itr); } diff --git a/src/core/manager.cc b/src/core/manager.cc index 5c0fe48b..c08e9ffe 100644 --- a/src/core/manager.cc +++ b/src/core/manager.cc @@ -36,7 +36,6 @@ #include "config.h" -#include #include #include #include @@ -231,8 +230,8 @@ Manager::initialize_first() { // Most of this should be possible to move out. void Manager::initialize_second() { - torrent::Http::set_factory(m_pollManager->get_http_stack()->get_http_factory()); - m_httpQueue->slot_factory(m_pollManager->get_http_stack()->get_http_factory()); + torrent::Http::set_factory(sigc::mem_fun(m_pollManager->get_http_stack(), &CurlStack::new_object)); + m_httpQueue->slot_factory(sigc::mem_fun(m_pollManager->get_http_stack(), &CurlStack::new_object)); CurlStack::global_init(); diff --git a/src/core/poll_manager_epoll.cc b/src/core/poll_manager_epoll.cc index 9d45d1a5..ccf645cd 100644 --- a/src/core/poll_manager_epoll.cc +++ b/src/core/poll_manager_epoll.cc @@ -67,7 +67,7 @@ PollManagerEPoll::poll(rak::timer timeout) { torrent::perform(); timeout = std::min(timeout, rak::timer(torrent::next_timeout())) + 1000; - if (m_httpStack.is_busy()) { + if (!m_httpStack.empty()) { // When we're using libcurl we need to use select, but as this is // inefficient we try avoiding it whenever possible. #if defined USE_VARIABLE_FDSET diff --git a/src/core/poll_manager_kqueue.cc b/src/core/poll_manager_kqueue.cc index 2baa2c07..f391a6c0 100644 --- a/src/core/poll_manager_kqueue.cc +++ b/src/core/poll_manager_kqueue.cc @@ -68,7 +68,7 @@ PollManagerKQueue::poll(rak::timer timeout) { torrent::perform(); timeout = std::min(timeout, rak::timer(torrent::next_timeout())) + 1000; - if (m_httpStack.is_busy()) { + if (!m_httpStack.empty()) { // When we're using libcurl we need to use select, but as this is // inefficient we try avoiding it whenever possible. #if defined USE_VARIABLE_FDSET diff --git a/src/core/poll_manager_select.cc b/src/core/poll_manager_select.cc index bf98def8..913a4545 100644 --- a/src/core/poll_manager_select.cc +++ b/src/core/poll_manager_select.cc @@ -77,7 +77,7 @@ PollManagerSelect::poll(rak::timer timeout) { unsigned int maxFd = static_cast(m_poll)->fdset(m_readSet, m_writeSet, m_errorSet); - if (m_httpStack.is_busy()) + if (!m_httpStack.empty()) maxFd = std::max(maxFd, m_httpStack.fdset(m_readSet, m_writeSet, m_errorSet)); timeval t = timeout.tval(); @@ -85,7 +85,7 @@ PollManagerSelect::poll(rak::timer timeout) { if (select(maxFd + 1, m_readSet, m_writeSet, m_errorSet, &t) == -1) return check_error(); - if (m_httpStack.is_busy()) + if (!m_httpStack.empty()) m_httpStack.perform(); torrent::perform(); diff --git a/src/display/utils.cc b/src/display/utils.cc index badd8da3..239b1596 100644 --- a/src/display/utils.cc +++ b/src/display/utils.cc @@ -51,7 +51,9 @@ #include #include "core/download.h" +#include "core/manager.h" +#include "globals.h" #include "utils.h" namespace display { @@ -279,11 +281,15 @@ print_status_info(char* first, char* last) { } char* -print_status_extra(char* first, char* last, __UNUSED Control* c) { +print_status_extra(char* first, char* last) { first = print_buffer(first, last, " [U %i/%i]", torrent::currently_unchoked(), torrent::max_unchoked()); + first = print_buffer(first, last, " [H %u/%u]", + control->core()->get_poll_manager()->get_http_stack()->active(), + control->core()->get_poll_manager()->get_http_stack()->max_active()); + first = print_buffer(first, last, " [S %i/%i/%i]", torrent::total_handshakes(), torrent::connection_manager()->size(), diff --git a/src/display/utils.h b/src/display/utils.h index a79969af..48779eac 100644 --- a/src/display/utils.h +++ b/src/display/utils.h @@ -76,7 +76,7 @@ char* print_entry_tags(char* first, char* last); char* print_entry_file(char* first, char* last, const torrent::Entry& entry); char* print_status_info(char* first, char* last); -char* print_status_extra(char* first, char* last, Control* c); +char* print_status_extra(char* first, char* last); inline char* print_buffer(char* first, char* last, const char* format) { diff --git a/src/display/window_statusbar.cc b/src/display/window_statusbar.cc index 9ed91340..a7b058e7 100644 --- a/src/display/window_statusbar.cc +++ b/src/display/window_statusbar.cc @@ -95,7 +95,7 @@ WindowStatusbar::redraw() { last = last - (position - buffer); if (last > buffer) { - position = print_status_extra(buffer, last, control); + position = print_status_extra(buffer, last); m_canvas->print(m_canvas->width() - (position - buffer), 0, "%s", buffer); } diff --git a/src/option_handler_rules.cc b/src/option_handler_rules.cc index 05cf14cc..04ef6554 100644 --- a/src/option_handler_rules.cc +++ b/src/option_handler_rules.cc @@ -471,6 +471,8 @@ initialize_option_handler(Control* c) { variables->insert("max_open_files", new utils::VariableValueSlot(rak::ptr_fn(&torrent::max_open_files), rak::ptr_fn(&torrent::set_max_open_files))); variables->insert("max_open_sockets", new utils::VariableValueSlot(rak::mem_fn(torrent::connection_manager(), &torrent::ConnectionManager::max_size), rak::mem_fn(torrent::connection_manager(), &torrent::ConnectionManager::set_max_size))); + variables->insert("max_open_http", new utils::VariableValueSlot(rak::mem_fn(c->core()->get_poll_manager()->get_http_stack(), &core::CurlStack::max_active), + rak::mem_fn(c->core()->get_poll_manager()->get_http_stack(), &core::CurlStack::set_max_active))); variables->insert("print", new utils::VariableStringSlot(rak::value_fn(std::string()), rak::mem_fn(control->core(), &core::Manager::push_log))); variables->insert("import", new utils::VariableStringSlot(rak::value_fn(std::string()), rak::ptr_fn(&apply_import)));