Added SocketManager to handle reuse of uninterested fd's by the kernel.

src/rpc/scgi.cc

@@ -9,6 +9,7 @@
 #include <torrent/net/fd.h>
 #include <torrent/net/poll.h>
 #include <torrent/net/socket_address.h>
+#include <torrent/runtime/socket_manager.h>
 
 #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,
+  torrent::runtime::socket_manager()->open_event_or_throw(this, [&]() {
-                                   reinterpret_cast<sockaddr*>(sa)->sa_family);
+      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)
+      if (fd == -1)
-    throw torrent::resource_error("Could not open socket for listening: " + std::string(std::strerror(errno)));
+        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)) {
+      if (dont_route && !torrent::fd_set_dont_route(fd, true)) {
-    torrent::fd_close(fd);
+        torrent::fd_close(fd);
-    throw torrent::resource_error("Could not set socket option IP_DONTROUTE: " + std::string(std::strerror(errno)));
+        throw torrent::resource_error("Could not set socket option IP_DONTROUTE: " + std::string(std::strerror(errno)));
-  }
+      }
 
-  set_file_descriptor(fd);
+      set_file_descriptor(fd);
-  open(reinterpret_cast<sockaddr*>(sa), length);
+
+      open(reinterpret_cast<sockaddr*>(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)
+      if (fd == -1)
-    throw torrent::resource_error("Could not open socket for listening: " + std::string(std::strerror(errno)));
+        throw torrent::resource_error("Could not open socket for listening: " + std::string(std::strerror(errno)));
 
-  set_file_descriptor(fd);
+      set_file_descriptor(fd);
-  open(reinterpret_cast<sockaddr*>(sa), offsetof(struct sockaddr_un, sun_path) + filename.size() + 1);
+
+      open(reinterpret_cast<sockaddr*>(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;
-    throw e;
   }
 }
 
@@ -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());
+      torrent::fd_close(file_descriptor());
-  set_file_descriptor(-1);
+      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());
+    auto* task = std::find_if(m_task, m_task + max_tasks, std::mem_fn(&SCgiTask::is_available));
-
-    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));
 
     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;
   }
 }
 

src/rpc/scgi_task.cc

@@ -11,6 +11,7 @@
 #include <torrent/torrent.h>
 #include <torrent/net/fd.h>
 #include <torrent/net/poll.h>
+#include <torrent/runtime/socket_manager.h>
 #include <torrent/utils/log.h>
 #include <torrent/utils/thread.h>
 
@@ -23,11 +24,11 @@ namespace rpc {
 
 void
 SCgiTask::open(SCgi* parent, int fd) {
-  m_parent      = parent;
+  set_file_descriptor(fd);
-  m_fileDesc    = 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());
+      torrent::fd_close(file_descriptor());
-  set_file_descriptor(-1);
+      set_file_descriptor(-1);
+    });
 
   auto lock = std::lock_guard<std::mutex>(m_result_mutex);
 

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; }

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);
   }