diff --git a/src/rpc/scgi.cc b/src/rpc/scgi.cc index db3ccf6c..64e9c56c 100644 --- a/src/rpc/scgi.cc +++ b/src/rpc/scgi.cc @@ -9,6 +9,7 @@ #include #include #include +#include #include "control.h" #include "globals.h" @@ -25,19 +26,22 @@ SCgi::~SCgi() { void SCgi::open_port(sockaddr* sa, unsigned int length, bool dont_route) { - int fd = torrent::fd_open_family(torrent::fd_flag_stream | torrent::fd_flag_nonblock | torrent::fd_flag_reuse_address, - reinterpret_cast(sa)->sa_family); + torrent::runtime::socket_manager()->open_event_or_throw(this, [&]() { + int fd = torrent::fd_open_family(torrent::fd_flag_stream | torrent::fd_flag_nonblock | torrent::fd_flag_reuse_address, + sa->sa_family); - if (fd == -1) - throw torrent::resource_error("Could not open socket for listening: " + std::string(std::strerror(errno))); + if (fd == -1) + throw torrent::resource_error("Could not open socket for listening: " + std::string(std::strerror(errno))); - if (dont_route && !torrent::fd_set_dont_route(fd, true)) { - torrent::fd_close(fd); - throw torrent::resource_error("Could not set socket option IP_DONTROUTE: " + std::string(std::strerror(errno))); - } + if (dont_route && !torrent::fd_set_dont_route(fd, true)) { + torrent::fd_close(fd); + throw torrent::resource_error("Could not set socket option IP_DONTROUTE: " + std::string(std::strerror(errno))); + } - set_file_descriptor(fd); - open(reinterpret_cast(sa), length); + set_file_descriptor(fd); + + open(reinterpret_cast(sa), length); + }); torrent::connection_manager()->inc_socket_count(); } @@ -53,13 +57,16 @@ SCgi::open_named(const std::string& filename) { sa->sun_family = AF_LOCAL; std::memcpy(sa->sun_path, filename.c_str(), filename.size() + 1); - int fd = torrent::fd_open_local(torrent::fd_flag_stream | torrent::fd_flag_nonblock | torrent::fd_flag_reuse_address); + torrent::runtime::socket_manager()->open_event_or_throw(this, [&]() { + int fd = torrent::fd_open_local(torrent::fd_flag_stream | torrent::fd_flag_nonblock | torrent::fd_flag_reuse_address); - if (fd == -1) - throw torrent::resource_error("Could not open socket for listening: " + std::string(std::strerror(errno))); + if (fd == -1) + throw torrent::resource_error("Could not open socket for listening: " + std::string(std::strerror(errno))); - set_file_descriptor(fd); - open(reinterpret_cast(sa), offsetof(struct sockaddr_un, sun_path) + filename.size() + 1); + set_file_descriptor(fd); + + open(reinterpret_cast(sa), offsetof(struct sockaddr_un, sun_path) + filename.size() + 1); + }); torrent::connection_manager()->inc_socket_count(); @@ -78,8 +85,7 @@ SCgi::open(sockaddr* sa, unsigned int length) { } catch (torrent::resource_error& e) { torrent::fd_close(file_descriptor()); set_file_descriptor(-1); - - throw e; + throw; } } @@ -92,6 +98,7 @@ SCgi::activate() { torrent::this_thread::poll()->insert_error(this); } +// TODO: This should close the fd to avoid reuse. void SCgi::stop() { assert(torrent::this_thread::thread() == scgi_thread::thread()); @@ -103,10 +110,12 @@ SCgi::stop() { if (itr->is_open()) itr->close(); - torrent::this_thread::poll()->remove_and_close(this); + torrent::runtime::socket_manager()->close_event_or_throw(this, [this]() { + torrent::this_thread::poll()->remove_and_close(this); - torrent::fd_close(file_descriptor()); - set_file_descriptor(-1); + torrent::fd_close(file_descriptor()); + set_file_descriptor(-1); + }); torrent::connection_manager()->dec_socket_count(); @@ -117,23 +126,39 @@ SCgi::stop() { void SCgi::event_read() { while (true) { - int fd = torrent::fd_accept(file_descriptor()); - - if (fd == -1) { - if (errno == EAGAIN || errno == EWOULDBLOCK) - break; - - throw torrent::resource_error("Listener port accept() failed: " + std::string(std::strerror(errno))); - } - - SCgiTask* task = std::find_if(m_task, m_task + max_tasks, std::mem_fn(&SCgiTask::is_available)); + auto* task = std::find_if(m_task, m_task + max_tasks, std::mem_fn(&SCgiTask::is_available)); if (task == m_task + max_tasks) { - torrent::fd_close(fd); + // TODO: Currently just close, although we should remove ourselves from read. + int fd = torrent::fd_accept(file_descriptor()); + + if (fd != -1) + torrent::fd_close(fd); + continue; } - task->open(this, fd); + auto open_func = [this, task]() { + int fd = torrent::fd_accept(file_descriptor()); + + if (fd == -1) { + if (errno == EAGAIN || errno == EWOULDBLOCK) + return; + + throw torrent::resource_error("Listener port accept() failed: " + std::string(std::strerror(errno))); + } + + task->open(this, fd); + }; + + auto cleanup_func = [task]() { + task->cancel_open(); + }; + + bool result = torrent::runtime::socket_manager()->open_event_or_cleanup(task, open_func, cleanup_func); + + if (!result) + break; } } diff --git a/src/rpc/scgi_task.cc b/src/rpc/scgi_task.cc index 131fe720..2439d4dc 100644 --- a/src/rpc/scgi_task.cc +++ b/src/rpc/scgi_task.cc @@ -11,6 +11,7 @@ #include #include #include +#include #include #include @@ -23,11 +24,11 @@ namespace rpc { void SCgiTask::open(SCgi* parent, int fd) { - m_parent = parent; - m_fileDesc = fd; + set_file_descriptor(fd); m_buffer.reset(new char[default_buffer_size + 1]); + m_parent = parent; m_buffer_size = default_buffer_size; m_position = m_buffer.get(); m_body = nullptr; @@ -37,6 +38,19 @@ SCgiTask::open(SCgi* parent, int fd) { torrent::this_thread::poll()->insert_error(this); } +void +SCgiTask::cancel_open() { + if (!is_open()) + return; + + torrent::runtime::socket_manager()->close_event_or_throw(this, [this]() { + torrent::this_thread::poll()->remove_and_close(this); + + torrent::fd_close(file_descriptor()); + set_file_descriptor(-1); + }); +}; + void SCgiTask::close() { if (!is_open()) @@ -45,10 +59,12 @@ SCgiTask::close() { torrent::main_thread::thread()->cancel_callback_and_wait(this); torrent::utils::Thread::self()->cancel_callback(this); - torrent::this_thread::poll()->remove_and_close(this); + torrent::runtime::socket_manager()->close_event_or_throw(this, [this]() { + torrent::this_thread::poll()->remove_and_close(this); - torrent::fd_close(file_descriptor()); - set_file_descriptor(-1); + torrent::fd_close(file_descriptor()); + set_file_descriptor(-1); + }); auto lock = std::lock_guard(m_result_mutex); diff --git a/src/rpc/scgi_task.h b/src/rpc/scgi_task.h index b1ce5ce7..db21732c 100644 --- a/src/rpc/scgi_task.h +++ b/src/rpc/scgi_task.h @@ -25,6 +25,8 @@ public: bool is_available() const { return m_fileDesc == -1; } void open(SCgi* parent, int fd); + void cancel_open(); + void close(); ContentType content_type() const { return m_content_type; } diff --git a/src/scgi/thread_scgi.cc b/src/scgi/thread_scgi.cc index 7e09d373..e44ce473 100644 --- a/src/scgi/thread_scgi.cc +++ b/src/scgi/thread_scgi.cc @@ -86,6 +86,7 @@ ThreadScgi::change_rpc_log() { if (scgi()->log_fd() != -1) { ::close(scgi()->log_fd()); scgi()->set_log_fd(-1); + lt_log_print(torrent::LOG_NOTICE, "Closed RPC log.", 0); }