Converted curl/http to be thread-safe.

Author: rakshasa

src/manager.cc

@@ -31,7 +31,7 @@ namespace this_thread {
 void event_open(Event* event)             { utils::ThreadInternal::poll()->open(event); }
 void event_open_and_count(Event* event)   { utils::ThreadInternal::poll()->open(event); manager->connection_manager()->inc_socket_count(); }
 void event_close_and_count(Event* event)  { utils::ThreadInternal::poll()->close(event); manager->connection_manager()->dec_socket_count(); }
-void event_closed_and_count(Event* event) { utils::ThreadInternal::poll()->closed(event); manager->connection_manager()->dec_socket_count(); }
+void event_closed_and_count(Event* event) { utils::ThreadInternal::poll()->cleanup_closed(event); manager->connection_manager()->dec_socket_count(); }
 void event_insert_read(Event* event)      { utils::ThreadInternal::poll()->insert_read(event); }
 void event_insert_write(Event* event)     { utils::ThreadInternal::poll()->insert_write(event); }
 void event_insert_error(Event* event)     { utils::ThreadInternal::poll()->insert_error(event); }

src/net/curl_get.cc

@@ -2,6 +2,7 @@
 
 #include "net/curl_get.h"
 
+#include <cassert>
 #include <iostream>
 #include <curl/easy.h>
 
@@ -11,6 +12,8 @@
 
 namespace torrent::net {
 
+// TODO: Seperate the thread-owned and public variables in different cachelines.
+
 static size_t
 curl_get_receive_write(void* data, size_t size, size_t nmemb, void* handle) {
   if (!((CurlGet*)handle)->stream()->write((const char*)data, size * nmemb).fail())
@@ -19,70 +22,124 @@ curl_get_receive_write(void* data, size_t size, size_t nmemb, void* handle) {
     return 0;
 }
 
-CurlGet::CurlGet(CurlStack* s)
-  : m_stack(s) {
-
+CurlGet::CurlGet() {
   m_task_timeout.slot() = [this]() { receive_timeout(); };
 }
 
 CurlGet::~CurlGet() {
-  close();
+  // CurlStack keeps a shared_ptr to this object, so it will only be destroyed once it is removed
+  // from the stack.
+  assert(!is_busy() && "CurlGet::~CurlGet called while still busy.");
+}
+
+void
+CurlGet::set_url(const std::string& url) {
+  auto guard = lock_guard();
+
+  if (m_handle != nullptr)
+    throw torrent::internal_error("CurlGet::set_url(...) called on a busy object.");
+
+  m_url = url;
 }
 
+// Make sure the output stream does not have any bad/failed bits set.
+//
+// TODO: Make the stream into something you pass to CurlGet, as a unique_ptr, and then have a way to
+// receive it in a thread-safe way on success.
 void
-CurlGet::start() {
-  if (is_busy())
-    throw torrent::internal_error("Tried to call CurlGet::start on a busy object.");
+CurlGet::set_stream(std::iostream* str) {
+  auto guard = lock_guard();
+
+  if (m_handle != nullptr)
+    throw torrent::internal_error("CurlGet::set_stream(...) called on a busy object.");
+
+  m_stream = str;
+}
+
+void
+CurlGet::set_timeout(uint32_t seconds) {
+  auto guard = lock_guard();
+
+  if (m_handle != nullptr)
+    throw torrent::internal_error("CurlGet::set_timeout(...) called on a busy object.");
+
+  m_timeout = seconds;
+}
+
+// TODO: When we add callback for start/close add an atomic_bool to indicate we've queued the
+// action, and use that to tell the user that the http_get is busy or not.
 
-  if (m_stream == NULL)
-    throw torrent::internal_error("Tried to call CurlGet::start without a valid output stream.");
+void
+CurlGet::prepare_start(CurlStack* stack) {
+  if (m_handle != nullptr)
+    throw torrent::internal_error("CurlGet::prepare_start(...) called on a busy object.");
 
-  if (!m_stack->is_running())
-    return;
+  if (m_stream == nullptr)
+    throw torrent::internal_error("CurlGet::prepare_start(...) called with a null stream.");
 
   m_handle = curl_easy_init();
+  m_stack = stack;
 
-  if (m_handle == NULL)
+  if (m_handle == nullptr)
     throw torrent::internal_error("Call to curl_easy_init() failed.");
 
   curl_easy_setopt(m_handle, CURLOPT_URL,            m_url.c_str());
   curl_easy_setopt(m_handle, CURLOPT_WRITEFUNCTION,  &curl_get_receive_write);
   curl_easy_setopt(m_handle, CURLOPT_WRITEDATA,      this);
 
   if (m_timeout != 0) {
-    curl_easy_setopt(m_handle, CURLOPT_CONNECTTIMEOUT, (long)60);
-    curl_easy_setopt(m_handle, CURLOPT_TIMEOUT,        (long)m_timeout);
-
-    // Normally libcurl should handle the timeout. But sometimes that doesn't
-    // work right so we do a fallback timeout that just aborts the transfer.
-    torrent::this_thread::scheduler()->update_wait_for_ceil_seconds(&m_task_timeout, 5s + 1s*m_timeout);
+    curl_easy_setopt(m_handle, CURLOPT_CONNECTTIMEOUT, 60l);
+    curl_easy_setopt(m_handle, CURLOPT_TIMEOUT,        static_cast<long>(m_timeout));
   }
 
-  curl_easy_setopt(m_handle, CURLOPT_FORBID_REUSE,   (long)1);
-  curl_easy_setopt(m_handle, CURLOPT_NOSIGNAL,       (long)1);
-  curl_easy_setopt(m_handle, CURLOPT_FOLLOWLOCATION, (long)1);
-  curl_easy_setopt(m_handle, CURLOPT_MAXREDIRS,      (long)5);
+  curl_easy_setopt(m_handle, CURLOPT_FORBID_REUSE,   1l);
+  curl_easy_setopt(m_handle, CURLOPT_NOSIGNAL,       1l);
+  curl_easy_setopt(m_handle, CURLOPT_FOLLOWLOCATION, 1l);
+  curl_easy_setopt(m_handle, CURLOPT_MAXREDIRS,      5l);
 
   curl_easy_setopt(m_handle, CURLOPT_IPRESOLVE,      CURL_IPRESOLVE_WHATEVER);
-
   curl_easy_setopt(m_handle, CURLOPT_ENCODING,       "");
 
   m_ipv6 = false;
+}
+
+void
+CurlGet::activate() {
+  CURLMcode code = curl_multi_add_handle(m_stack->handle(), m_handle);
+
+  if (code != CURLM_OK)
+    throw torrent::internal_error("CurlGet::activate() error calling curl_multi_add_handle: " + std::string(curl_multi_strerror(code)));
 
-  m_stack->add_get(this);
+  // Normally libcurl should handle the timeout. But sometimes that doesn't
+  // work right so we do a fallback timeout that just aborts the transfer.
+  //
+  // TODO: Verify this is still needed, as it was added to work around during early libcurl
+  // versions.
+  if (m_timeout != 0)
+    torrent::this_thread::scheduler()->update_wait_for_ceil_seconds(&m_task_timeout, 1min + 1s*m_timeout);
+
+  m_active = true;
 }
 
 void
-CurlGet::close() {
-  torrent::this_thread::scheduler()->erase(&m_task_timeout);
+CurlGet::cleanup() {
+  if (m_handle == nullptr)
+    throw torrent::internal_error("CurlGet::cleanup() called on a null m_handle.");
 
-  if (!is_busy())
-    return;
+  if (m_active) {
+    CURLMcode code = curl_multi_remove_handle(m_stack->handle(), m_handle);
 
-  m_stack->remove_get(this);
+    if (code != CURLM_OK)
+      throw torrent::internal_error("CurlGet::cleanup() error calling curl_multi_remove_handle: " + std::string(curl_multi_strerror(code)));
+
+    torrent::this_thread::scheduler()->erase(&m_task_timeout);
+    m_active = false;
+  }
 
   curl_easy_cleanup(m_handle);
-  m_handle = NULL;
+
+  m_handle = nullptr;
+  m_stack = nullptr;
 }
 
 void
@@ -114,9 +171,12 @@ CurlGet::size_total() {
 
 void
 CurlGet::receive_timeout() {
-  return m_stack->transfer_done(m_handle, "Timed out");
+  // return m_stack->transfer_done(m_handle, "Timed out");
+  throw internal_error("CurlGet::receive_timeout() called, however this was a hack to work around libcurl not handling timeouts correctly.");
 }
 
+// TODO: Verify slots are handled while CurlGet and CurlStack are unlocked.
+
 void
 CurlGet::trigger_done() {
   ::utils::slot_list_call(m_signal_done);

src/net/curl_get.h

@@ -3,6 +3,7 @@
 
 #include <iosfwd>
 #include <list>
+#include <mutex>
 #include <string>
 #include <curl/curl.h>
 
@@ -14,68 +15,124 @@ class CurlStack;
 
 class CurlGet {
 public:
-  CurlGet(CurlStack* s);
-  virtual ~CurlGet();
+  CurlGet();
+  ~CurlGet();
 
-  bool               is_active() const               { return m_active; }
-  bool               is_busy() const                 { return m_handle; }
-  bool               is_using_ipv6()                 { return m_ipv6; }
+  bool               is_active() const;
+  bool               is_busy() const;
+  bool               is_using_ipv6() const;
 
-  void               start();
-  void               close();
-
-  const std::string& url() const                     { return m_url; }
-  std::iostream*     stream()                        { return m_stream; }
-  uint32_t           timeout() const                 { return m_timeout; }
+  // TODO: Don't allow getting the stream.
+  const std::string& url() const;
+  std::iostream*     stream();
+  uint32_t           timeout() const;
 
   // Make sure the output stream does not have any bad/failed bits set.
-  void               set_active(bool a)              { m_active = a; }
-  void               set_url(const std::string& url) { m_url = url; }
-  void               set_stream(std::iostream* str)  { m_stream = str; }
-  void               set_timeout(uint32_t seconds)   { m_timeout = seconds; }
+  void               set_url(const std::string& url);
+  void               set_stream(std::iostream* str);
+  // TODO: Change to std::chrono.
+  void               set_timeout(uint32_t seconds);
 
   void               retry_ipv6();
 
   int64_t            size_done();
   int64_t            size_total();
 
+  // TODO: This isn't thread-safe, and HttpGet needs to be fixed.
   CURL*              handle()                        { return m_handle; }
+  CurlStack*         curl_stack()                    { return m_stack; }
 
   // The owner of the Http object must close it as soon as possible
   // after receiving the signal, as the implementation may allocate
   // limited resources during its lifetime.
   auto&              signal_done()                   { return m_signal_done; }
   auto&              signal_failed()                 { return m_signal_failed; }
 
-private:
+protected:
   friend class CurlStack;
 
-  CurlGet(const CurlGet&) = delete;
-  void operator = (const CurlGet&) = delete;
+  // CurlStack is responsible for locking.
+  bool               is_active_no_locking() const    { return m_active; }
+
+  void               prepare_start(CurlStack* stack);
+  void               activate();
+  void               cleanup();
+
+  // We need to lock when changing any of the values publically accessible. This means we don't need
+  // to lock when changing the underlying vector.
+  void               lock() const                    { m_mutex.lock(); }
+  auto               lock_guard() const              { return std::scoped_lock(m_mutex); }
+  void               unlock() const                  { m_mutex.unlock(); }
+  auto&              mutex() const                   { return m_mutex; }
 
   void               receive_timeout();
 
   void               trigger_done();
   void               trigger_failed(const std::string& message);
 
-  bool               m_active{};
+private:
+  CurlGet(const CurlGet&) = delete;
+  void operator = (const CurlGet&) = delete;
+
+  mutable std::mutex  m_mutex;
+
+  // TODO: Consider locking requirements.
+  CURL*              m_handle{};
+  CurlStack*         m_stack;
+
   bool               m_ipv6;
 
-  // TODO: Use shared_ptr, or replace with std::function.
-  std::string        m_url;
-  std::iostream*     m_stream{};
+  torrent::utils::SchedulerEntry m_task_timeout;
+
+  // When you change timeout to a different type, update curl_get.cc where it multiplies 1s*m_timeout.
 
-  // When you change this to a different type, update curl_get.cc where it multiplies 1s*m_timeout.
+  bool               m_active{};
   uint32_t           m_timeout{};
 
-  CURL*              m_handle{};
-  CurlStack*         m_stack;
+  std::string        m_url;
+  // TODO: Use shared_ptr, or replace with std::function.
+  std::iostream*     m_stream{};
 
-  torrent::utils::SchedulerEntry                     m_task_timeout;
   std::list<std::function<void()>>                   m_signal_done;
   std::list<std::function<void(const std::string&)>> m_signal_failed;
 };
 
+inline bool
+CurlGet::is_active() const {
+  auto guard = lock_guard();
+  return m_active;
+}
+
+inline bool
+CurlGet::is_busy() const {
+  auto guard = lock_guard();
+  return m_handle != nullptr;
+}
+
+inline bool
+CurlGet::is_using_ipv6() const {
+  auto guard = lock_guard();
+  return m_ipv6;
+}
+
+inline const std::string&
+CurlGet::url() const {
+  auto guard = lock_guard();
+  return m_url;
+}
+
+inline std::iostream*
+CurlGet::stream() {
+  auto guard = lock_guard();
+  return m_stream;
+}
+
+inline uint32_t
+CurlGet::timeout() const {
+  auto guard = lock_guard();
+  return m_timeout;
+}
+
 }
 
 #endif