Cleaned up thread-related code.

.github/workflows/post-static-analysis.yml

@@ -16,6 +16,9 @@ jobs:
       # OPTIONAL: auto-closing conversations requires the `contents` permission
       contents: write
     steps:
+    - name: Sleep for 30 seconds
+      run: sleep 30s
+      shell: bash
     - name: Download analysis results
       uses: actions/github-script@v7
       with:

src/control.cc

@@ -142,9 +142,10 @@ Control::handle_shutdown() {
   rpc::commands.call_catch("event.system.shutdown", rpc::make_target(), "shutdown", "System shutdown event action failed: ");
 
   if (!m_shutdownQuick) {
-    // Temporary hack:
     if (worker_thread->is_active())
-      worker_thread->queue_item(&ThreadBase::stop_thread);
+      worker_thread->callback(nullptr, []() {
+          worker_thread->queue_stop_thread();
+        });
 
     torrent::connection_manager()->listen_close();
     m_directory_events->close();
@@ -154,9 +155,10 @@ Control::handle_shutdown() {
       priority_queue_insert(&taskScheduler, &m_taskShutdown, cachedTime + rak::timer::from_seconds(5));
 
   } else {
-    // Temporary hack:
     if (worker_thread->is_active())
-      worker_thread->queue_item(&ThreadBase::stop_thread);
+      worker_thread->callback(nullptr, []() {
+          worker_thread->queue_stop_thread();
+        });
 
     m_core->shutdown(true);
   }

src/control.h

@@ -1,39 +1,3 @@
-// rTorrent - BitTorrent client
-// Copyright (C) 2005-2011, Jari Sundell
-//
-// This program is free software; you can redistribute it and/or modify
-// it under the terms of the GNU General Public License as published by
-// the Free Software Foundation; either version 2 of the License, or
-// (at your option) any later version.
-// 
-// This program is distributed in the hope that it will be useful,
-// but WITHOUT ANY WARRANTY; without even the implied warranty of
-// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
-// GNU General Public License for more details.
-// 
-// You should have received a copy of the GNU General Public License
-// along with this program; if not, write to the Free Software
-// Foundation, Inc., 59 Temple Place, Suite 330, Boston, MA  02111-1307  USA
-//
-// In addition, as a special exception, the copyright holders give
-// permission to link the code of portions of this program with the
-// OpenSSL library under certain conditions as described in each
-// individual source file, and distribute linked combinations
-// including the two.
-//
-// You must obey the GNU General Public License in all respects for
-// all of the code used other than OpenSSL.  If you modify file(s)
-// with this exception, you may extend this exception to your version
-// of the file(s), but you are not obligated to do so.  If you do not
-// wish to do so, delete this exception statement from your version.
-// If you delete this exception statement from all source files in the
-// program, then also delete it here.
-//
-// Contact:  Jari Sundell <jaris@ifi.uio.no>
-//
-//           Skomakerveien 33
-//           3185 Skoppum, NORWAY
-
 #ifndef RTORRENT_CONTROL_H
 #define RTORRENT_CONTROL_H
 
@@ -61,7 +25,7 @@ namespace display {
 namespace input {
   class InputEvent;
   class Manager;
-}  
+}
 
 namespace rpc {
   class CommandScheduler;
@@ -78,7 +42,7 @@ class Control {
 public:
   Control();
   ~Control();
-  
+
   bool                is_shutdown_completed();
   bool                is_shutdown_received()        { return m_shutdownReceived; }
   bool                is_shutdown_started()         { return m_shutdownQuick; }

src/core/download_list.cc

@@ -90,11 +90,11 @@ DownloadList::create(torrent::Object* obj, uint32_t tracker_key, bool printLog)
     download = torrent::download_add(obj, tracker_key);
 
   } catch (torrent::local_error& e) {
-    delete obj;
-
     if (printLog)
       lt_log_print(torrent::LOG_TORRENT_ERROR, "Could not create download: %s", e.what());
 
+    delete obj;
+
     return NULL;
   }
 

src/rpc/scgi_task.cc

@@ -39,9 +39,9 @@ SCgiTask::open(SCgi* parent, int fd) {
   m_position    = m_buffer;
   m_body        = NULL;
 
-  torrent::thread_self->poll()->open(this);
+  torrent::thread_self()->poll()->open(this);
-  torrent::thread_self->poll()->insert_read(this);
+  torrent::thread_self()->poll()->insert_read(this);
-  torrent::thread_self->poll()->insert_error(this);
+  torrent::thread_self()->poll()->insert_error(this);
 }
 
 void
@@ -50,12 +50,12 @@ SCgiTask::close() {
     return;
 
   torrent::main_thread()->cancel_callback_and_wait(this);
-  torrent::thread_self->cancel_callback(this);
+  torrent::thread_self()->cancel_callback(this);
 
-  torrent::thread_self->poll()->remove_read(this);
+  torrent::thread_self()->poll()->remove_read(this);
-  torrent::thread_self->poll()->remove_write(this);
+  torrent::thread_self()->poll()->remove_write(this);
-  torrent::thread_self->poll()->remove_error(this);
+  torrent::thread_self()->poll()->remove_error(this);
-  torrent::thread_self->poll()->close(this);
+  torrent::thread_self()->poll()->close(this);
 
   get_fd().close();
   get_fd().clear();
@@ -184,7 +184,7 @@ SCgiTask::event_read() {
   if ((unsigned int)std::distance(m_buffer, m_position) != m_buffer_size)
     return;
 
-  torrent::thread_self->poll()->remove_read(this);
+  torrent::thread_self()->poll()->remove_read(this);
 
   if (m_parent->log_fd() >= 0) {
     int __UNUSED result;
@@ -238,7 +238,7 @@ void
 SCgiTask::receive_call(const char* buffer, uint32_t length) {
   // TODO: Rewrite RpcManager.process to pass the result buffer instead of having to copy it.
 
-  auto scgi_thread = torrent::thread_self;
+  auto scgi_thread = torrent::thread_self();
 
   auto result_callback = [this, scgi_thread](const char* b, uint32_t l) {
       receive_write(b, l);
@@ -248,7 +248,7 @@ SCgiTask::receive_call(const char* buffer, uint32_t length) {
           m_result_mutex.lock();
           m_result_mutex.unlock();
 
-          torrent::thread_self->poll()->insert_write(this);
+          torrent::thread_self()->poll()->insert_write(this);
         });
     };
 

src/thread_base.cc

@@ -14,44 +14,11 @@
 
 #include "globals.h"
 
-class thread_queue_hack : private std::vector<ThreadBase::thread_base_func> {
-public:
-  static constexpr unsigned int max_size = 32;
-
-  using value_type = ThreadBase::thread_base_func;
-  using base_type  = std::vector<value_type>;
-
-  thread_queue_hack() { reserve(max_size); };
-
-  bool empty() {
-    std::lock_guard<std::mutex> lguard(m_lock);
-    return base_type::empty();
-  }
-
-  void push_back(value_type v) {
-    const std::lock_guard<std::mutex> lguard(m_lock);
-    if (base_type::size() >= max_size)
-      throw torrent::internal_error("Overflowed thread_queue max size of " + std::to_string(max_size) + ".");
-    base_type::push_back(v);
-  }
-
-  void copy_and_clear(base_type& target) {
-    std::lock_guard<std::mutex> lguard(m_lock);
-    target.assign(begin(), end());
-    base_type::clear();
-  }
-
-private:
-  std::mutex m_lock;
-};
-
 void
 throw_shutdown_exception() { throw torrent::shutdown_exception(); }
 
 ThreadBase::ThreadBase() {
   m_taskShutdown.slot() = std::bind(&throw_shutdown_exception);
-
-  m_threadQueue = std::make_unique<thread_queue_hack>();
 }
 
 // Defined in here and not the header so that the default destructor
@@ -60,9 +27,9 @@ ThreadBase::~ThreadBase() = default;
 
 // Move to libtorrent...
 void
-ThreadBase::stop_thread(ThreadBase* thread) {
+ThreadBase::queue_stop_thread() {
-  if (!thread->m_taskShutdown.is_queued())
+  if (!m_taskShutdown.is_queued())
-    priority_queue_insert(&thread->m_taskScheduler, &thread->m_taskShutdown, cachedTime);
+    priority_queue_insert(&m_taskScheduler, &m_taskShutdown, cachedTime);
 }
 
 int64_t
@@ -73,33 +40,18 @@ ThreadBase::next_timeout_usec() {
     return 0;
   else
     return (m_taskScheduler.top()->time() - cachedTime).usec();
-}
 
-void
+  // TODO: Thread-specific cachedTime.
-ThreadBase::call_queued_items() {
+
-  std::vector<thread_queue_hack::value_type> queue;
+  // if (!taskScheduler.empty())
-  m_threadQueue->copy_and_clear(queue);
+  //   return std::max(taskScheduler.top()->time() - cachedTime, rak::timer()).usec();
-  for (auto itr : queue) {
+  // else
-    itr(this);
+  //   return rak::timer::from_seconds(600).usec();
-  }
 }
 
 void
 ThreadBase::call_events() {
-  // Check for new queued items set by other threads.
-  if (!m_threadQueue->empty())
-    call_queued_items();
-
   rak::priority_queue_perform(&m_taskScheduler, cachedTime);
 
   process_callbacks();
 }
-
-void
-ThreadBase::queue_item(thread_base_func newFunc) {
-  m_threadQueue->push_back(newFunc);
-
-  // Make it also restart inactive threads?
-  if (m_state == STATE_ACTIVE)
-    interrupt();
-}

src/thread_base.h

@@ -24,17 +24,17 @@ public:
   priority_queue&     task_scheduler() { return m_taskScheduler; }
 
   // Throw torrent::shutdown_exception to stop the thread.
-  static void         stop_thread(ThreadBase* thread);
+  void                queue_stop_thread();
 
   // ATM, only interaction with a thread's allowed by other threads is
   // through the queue_item call.
 
-  void                queue_item(thread_base_func newFunc);
+  // void                queue_item(thread_base_func newFunc);
 
 protected:
   int64_t             next_timeout_usec();
 
-  void                call_queued_items();
+  // void                call_queued_items();
   virtual void        call_events();
 
   // TODO: Add thread name.
@@ -47,7 +47,7 @@ protected:
 
   // Temporary hack to pass messages to a thread. This really needs to
   // be cleaned up and/or integrated into the priority queue itself.
-  std::unique_ptr<thread_queue_hack>  m_threadQueue;
+  // std::unique_ptr<thread_queue_hack>  m_threadQueue;
 };
 
 #endif

src/thread_worker.cc

@@ -34,7 +34,10 @@ ThreadWorker::set_scgi(rpc::SCgi* scgi) {
 
   change_rpc_log();
 
-  queue_item((thread_base_func)&start_scgi);
+  callback(nullptr, [this]() {
+      start_scgi(this);
+    });
+
   return true;
 }
 
@@ -42,7 +45,9 @@ void
 ThreadWorker::set_rpc_log(const std::string& filename) {
   m_rpcLog = filename;
 
-  queue_item((thread_base_func)&msg_change_rpc_log);
+  callback(nullptr, [this]() {
+      msg_change_rpc_log(this);
+    });
 }
 
 void