Revert "Use ReusableThread from the utilities package"

include/datahandlinglibs/FakeCardReaderBase.hpp

@@ -26,7 +26,7 @@
 
 #include "appfwk/ModuleConfiguration.hpp"
 
-#include "utilities/ReusableThread.hpp"
+#include "datahandlinglibs/utils/ReusableThread.hpp"
 #include "datahandlinglibs/utils/FileSourceBuffer.hpp"
 
 #include "nlohmann/json.hpp"

include/datahandlinglibs/concepts/RecorderConcept.hpp

@@ -12,7 +12,7 @@
 #include "datahandlinglibs/ReadoutTypes.hpp"
 //#include "datahandlinglibs/recorderconfig/Structs.hpp"
 #include "datahandlinglibs/utils/BufferedFileWriter.hpp"
-#include "utilities/ReusableThread.hpp"
+#include "datahandlinglibs/utils/ReusableThread.hpp"
 #include "appmodel/DataRecorderModule.hpp"
 
 #include <atomic>

include/datahandlinglibs/models/DataHandlingModel.hpp

@@ -43,7 +43,7 @@
 #include "datahandlinglibs/concepts/RequestHandlerConcept.hpp"
 
 #include "datahandlinglibs/DataHandlingIssues.hpp"
-#include "utilities/ReusableThread.hpp"
+#include "datahandlinglibs/utils/ReusableThread.hpp"
 
 #include <functional>
 #include <memory>
@@ -176,7 +176,7 @@ class DataHandlingModel : public DataHandlingConcept
   std::atomic<int> m_num_payloads_overwritten{ 0 };
 
   // CONSUMER
-  utilities::ReusableThread m_consumer_thread;
+  ReusableThread m_consumer_thread;
 
   // RAW RECEIVER
   std::chrono::milliseconds m_raw_receiver_timeout_ms;
@@ -197,7 +197,7 @@ class DataHandlingModel : public DataHandlingConcept
   // TIME-SYNC
   using timesync_sender_ct = iomanager::SenderConcept<dfmessages::TimeSync>; // no timeout -> published
   std::shared_ptr<timesync_sender_ct> m_timesync_sender;
-  utilities::ReusableThread m_timesync_thread;
+  ReusableThread m_timesync_thread;
   std::string m_timesync_connection_name;
   uint32_t m_pid_of_current_process;
 

include/datahandlinglibs/models/DataSubscriberModel.hpp

@@ -15,7 +15,7 @@
 #include "iomanager/Sender.hpp"
 #include "iomanager/Receiver.hpp"
 #include "logging/Logging.hpp"
-//#include "utilities/ReusableThread.hpp"
+//#include "datahandlinglibs/utils/ReusableThread.hpp"
 
 #include "confmodel/Connection.hpp"
 

include/datahandlinglibs/models/DefaultRequestHandlerModel.hpp

@@ -12,7 +12,7 @@
 #include "datahandlinglibs/DataHandlingIssues.hpp"
 #include "datahandlinglibs/concepts/RequestHandlerConcept.hpp"
 #include "datahandlinglibs/utils/BufferedFileWriter.hpp"
-#include "utilities/ReusableThread.hpp"
+#include "datahandlinglibs/utils/ReusableThread.hpp"
 
 #include "datahandlinglibs/opmon/datahandling_info.pb.h"
 
@@ -227,10 +227,10 @@ class DefaultRequestHandlerModel : public RequestHandlerConcept<ReadoutType, Lat
 
   // Data recording
   BufferedFileWriter<> m_buffered_writer;
-  utilities::ReusableThread m_recording_thread;
+  ReusableThread m_recording_thread;
 
-  utilities::ReusableThread m_cleanup_thread;
-  utilities::ReusableThread m_periodic_transmission_thread;
+  ReusableThread m_cleanup_thread;
+  ReusableThread m_periodic_transmission_thread;
 
   // Bookkeeping of OOB requests
   std::map<dfmessages::DataRequest, int> m_request_counter;

include/datahandlinglibs/models/RecorderModel.hpp

@@ -14,7 +14,7 @@
 #include "datahandlinglibs/ReadoutTypes.hpp"
 #include "datahandlinglibs/concepts/RecorderConcept.hpp"
 #include "datahandlinglibs/utils/BufferedFileWriter.hpp"
-#include "utilities/ReusableThread.hpp"
+#include "datahandlinglibs/utils/ReusableThread.hpp"
 
 #include "datahandlinglibs/opmon/datahandling_info.pb.h"
 
@@ -68,7 +68,7 @@ class RecorderModel : public RecorderConcept
   BufferedFileWriter<> m_buffered_writer;
 
   // Threading
-  utilities::ReusableThread m_work_thread;
+  ReusableThread m_work_thread;
   std::atomic<bool> m_run_marker;
 
   // Stats

include/datahandlinglibs/models/SourceEmulatorModel.hpp

@@ -21,7 +21,7 @@
 #include "datahandlinglibs/utils/ErrorBitGenerator.hpp"
 #include "datahandlinglibs/utils/FileSourceBuffer.hpp"
 #include "datahandlinglibs/utils/RateLimiter.hpp"
-#include "utilities/ReusableThread.hpp"
+#include "datahandlinglibs/utils/ReusableThread.hpp"
 
 #include "datahandlinglibs/opmon/datahandling_info.pb.h"
 
@@ -140,7 +140,7 @@ class SourceEmulatorModel : public SourceEmulatorConcept
   std::unique_ptr<FileSourceBuffer> m_file_source;
   ErrorBitGenerator m_error_bit_generator;
 
-  utilities::ReusableThread m_producer_thread;
+  ReusableThread m_producer_thread;
 
   std::string m_name;
   bool m_is_configured = false;

include/datahandlinglibs/models/TaskRawDataProcessorModel.hpp

@@ -24,7 +24,7 @@
 #include "appmodel/DataHandlerConf.hpp"
 #include "appmodel/DataProcessor.hpp"
 
-#include "utilities/ReusableThread.hpp"
+#include "datahandlinglibs/utils/ReusableThread.hpp"
 
 #include <folly/ProducerConsumerQueue.h>
 
@@ -113,7 +113,7 @@ class TaskRawDataProcessorModel : public RawDataProcessorConcept<ReadoutType>
   // Post-processing functions and their corresponding threads
   std::vector<std::function<void(const ReadoutType*)>> m_post_process_functions;
   std::vector<std::unique_ptr<folly::ProducerConsumerQueue<const ReadoutType*>>> m_items_to_postprocess_queues;
-  std::vector<std::unique_ptr<utilities::ReusableThread>> m_post_process_threads;
+  std::vector<std::unique_ptr<ReusableThread>> m_post_process_threads;
 
   // Internals
   size_t m_postprocess_queue_sizes;

include/datahandlinglibs/models/detail/TaskRawDataProcessorModel.hxx

@@ -80,7 +80,7 @@ template<typename Task>
 void 
 TaskRawDataProcessorModel<ReadoutType>::add_postprocess_task(Task&& task)
 {
-  m_post_process_threads.emplace_back(std::make_unique<utilities::ReusableThread>(0));
+  m_post_process_threads.emplace_back(std::make_unique<ReusableThread>(0));
   m_post_process_functions.push_back(std::forward<Task>(task));
 }
 

include/datahandlinglibs/utils/ReusableThread.hpp

@@ -0,0 +1,123 @@
+/**
+ * @file ReusableThread.hpp Reusable thread wrapper
+ * The same thread instance can be used with different tasks to be executed
+ * Inspired by:
+ * https://codereview.stackexchange.com/questions/134214/reuseable-c11-thread
+ *
+ * This is part of the DUNE DAQ , copyright 2020.
+ * Licensing/copyright details are in the COPYING file that you should have
+ * received with this code.
+ */
+#ifndef DATAHANDLINGLIBS_INCLUDE_DATAHANDLINGLIBS_UTILS_REUSABLETHREAD_HPP_
+#define DATAHANDLINGLIBS_INCLUDE_DATAHANDLINGLIBS_UTILS_REUSABLETHREAD_HPP_
+
+#include <atomic>
+#include <chrono>
+#include <condition_variable>
+#include <functional>
+#include <mutex>
+#include <string>
+#include <thread>
+
+namespace dunedaq {
+namespace datahandlinglibs {
+
+class ReusableThread
+{
+public:
+  explicit ReusableThread(int threadid=0)
+    : m_thread_id(threadid)
+    , m_task_executed(true)
+    , m_task_assigned(false)
+    , m_thread_quit(false)
+    , m_worker_done(false)
+    , m_thread(&ReusableThread::thread_worker, this)
+  {}
+
+  ~ReusableThread()
+  {
+    while (m_task_assigned) {
+      std::this_thread::sleep_for(std::chrono::milliseconds(1));
+    }
+    m_thread_quit = true;
+    while (!m_worker_done) {
+      std::this_thread::sleep_for(std::chrono::milliseconds(1));
+      m_cv.notify_all();
+    }
+    m_thread.join();
+  }
+
+  ReusableThread(const ReusableThread&) = delete;            ///< ReusableThread is not copy-constructible
+  ReusableThread& operator=(const ReusableThread&) = delete; ///< ReusableThread is not copy-assginable
+  ReusableThread(ReusableThread&&) = delete;                 ///< ReusableThread is not move-constructible
+  ReusableThread& operator=(ReusableThread&&) = delete;      ///< ReusableThread is not move-assignable
+
+  // Set thread ID
+  void set_thread_id(int tid) { m_thread_id = tid; }
+
+  // Get thread ID
+  int get_thread_id() const { return m_thread_id; }
+
+  // Set name for pthread handle
+  void set_name(const std::string& name, int tid)
+  {
+    set_thread_id(tid);
+    char tname[16];
+    snprintf(tname, 16, "%s-%d", name.c_str(), tid); // NOLINT
+    auto handle = m_thread.native_handle();
+    pthread_setname_np(handle, tname);
+  }
+
+  // Check for completed task execution
+  bool get_readiness() const { return m_task_executed; }
+
+  // Set task to be executed
+  template<typename Function, typename... Args>
+  bool set_work(Function&& f, Args&&... args)
+  {
+    if (!m_task_assigned && m_task_executed.exchange(false)) {
+      m_task = std::bind(f, args...);
+      m_task_assigned = true;
+      m_cv.notify_all();
+      return true;
+    }
+    return false;
+  }
+
+private:
+  // Internals
+  int m_thread_id;
+  std::atomic<bool> m_task_executed;
+  std::atomic<bool> m_task_assigned;
+  std::atomic<bool> m_thread_quit;
+  std::atomic<bool> m_worker_done;
+  std::function<void()> m_task;
+
+  // Locks
+  std::mutex m_mtx;
+  std::condition_variable m_cv;
+  std::thread m_thread;
+
+  // Actual worker thread
+  void thread_worker()
+  {
+    std::unique_lock<std::mutex> lock(m_mtx);
+
+    while (!m_thread_quit) {
+      if (!m_task_executed && m_task_assigned) {
+        m_task();
+        m_task_executed = true;
+        m_task_assigned = false;
+      } else {
+        m_cv.wait(lock);
+      }
+    }
+
+    m_worker_done = true;
+  }
+};
+
+} // namespace datahandlinglibs
+} // namespace dunedaq
+
+#endif // DATAHANDLINGLIBS_INCLUDE_DATAHANDLINGLIBS_UTILS_REUSABLETHREAD_HPP_