[beast: 42/95] BSE: port engine master thread to std::thread

[Tim Janik <timj src gnome org>]

commit ccde70ab12b309e2a555582ae6d43cf9b6867875
Author: Tim Janik <timj gnu org>
Date:   Tue Mar 19 01:35:11 2013 +0100

    BSE: port engine master thread to std::thread

 bse/bseengine.cc       |   45 ++++++++++++---------------------------------
 bse/bseenginemaster.cc |   46 ++++++++++++++++++++++++++++------------------
 bse/bseenginemaster.hh |   23 ++++++++++++++++++-----
 3 files changed, 58 insertions(+), 56 deletions(-)
---
diff --git a/bse/bseengine.cc b/bse/bseengine.cc
index 1dd191a..b1c65f3 100644
--- a/bse/bseengine.cc
+++ b/bse/bseengine.cc
@@ -1,6 +1,7 @@
 // Licensed GNU LGPL v2.1 or later: http://www.gnu.org/licenses/lgpl.html
 #include "bseengine.hh"
 #include "bsecore.hh"
+#include "bsemain.hh"
 #include "gslcommon.hh"
 #include "bseengineutils.hh"
 #include "bseenginemaster.hh"
@@ -9,7 +10,9 @@
 #include <unistd.h>
 
 static SFI_MSG_TYPE_DEFINE (debug_engine, "engine", SFI_MSG_DEBUG, NULL);
+#undef DEBUG // FIXME
 #define DEBUG(...)      sfi_debug (debug_engine, __VA_ARGS__)
+
 /* some systems don't have ERESTART (which is what linux returns for system
  * calls on pipes which are being interrupted). most probably just use EINTR,
  * and maybe some can return both. so we check for both in the below code,
@@ -18,8 +21,10 @@ static SFI_MSG_TYPE_DEFINE (debug_engine, "engine", SFI_MSG_DEBUG, NULL);
 #ifndef ERESTART
 #define ERESTART        EINTR
 #endif
+
 /* --- prototypes --- */
 static void wakeup_master (void);
+
 /* --- UserThread --- */
 /**
  * @param klass        the BseModuleClass which determines the module's behaviour
@@ -1096,8 +1101,7 @@ slave (gpointer data)
 /* --- setup & trigger --- */
 static gboolean                bse_engine_initialized = FALSE;
 static gboolean                bse_engine_threaded = FALSE;
-static BirnetThread    *master_thread = NULL;
-static EngineMasterData master_data;
+static Bse::MasterThread *master_thread = NULL;
 guint                  bse_engine_exvar_block_size = 0;
 guint                  bse_engine_exvar_sample_freq = 0;
 guint                  bse_engine_exvar_control_mask = 0;
@@ -1275,43 +1279,18 @@ bse_engine_init (gboolean run_threaded)
   bse_engine_threaded = run_threaded;
   if (bse_engine_threaded)
     {
-      gint err = pipe (master_data.wakeup_pipe);
-      master_data.user_thread = sfi_thread_self ();
-      if (!err)
-       {
-         glong d_long = fcntl (master_data.wakeup_pipe[0], F_GETFL, 0);
-         /* DEBUG ("master_wpipe-readfd, blocking=%ld", d_long & O_NONBLOCK); */
-         d_long |= O_NONBLOCK;
-         err = fcntl (master_data.wakeup_pipe[0], F_SETFL, d_long);
-       }
-      if (!err)
-       {
-         glong d_long = fcntl (master_data.wakeup_pipe[1], F_GETFL, 0);
-         /* DEBUG ("master_wpipe-writefd, blocking=%ld", d_long & O_NONBLOCK); */
-         d_long |= O_NONBLOCK;
-         err = fcntl (master_data.wakeup_pipe[1], F_SETFL, d_long);
-       }
-      if (err)
-       g_error ("failed to create wakeup pipe: %s", g_strerror (errno));
-      master_thread = sfi_thread_run ("DSP #1", (BirnetThreadFunc) bse_engine_master_thread, &master_data);
-      if (!master_thread)
-       g_error ("failed to create master thread");
-      if (0)
-       sfi_thread_run ("DSP #2", slave, NULL);
+      master_thread = new Bse::MasterThread (bse_main_wakeup);
+      (void) slave; // FIXME: start slave ("DSP #2")
     }
 }
+
 static void
 wakeup_master (void)
 {
-  if (master_thread)
-    {
-      guint8 data = 'W';
-      gint l;
-      do
-       l = write (master_data.wakeup_pipe[1], &data, 1);
-      while (l < 0 && (errno == EINTR || errno == ERESTART));
-    }
+  g_return_if_fail (master_thread != NULL);
+  master_thread->wakeup();
 }
+
 gboolean
 bse_engine_prepare (BseEngineLoop *loop)
 {
diff --git a/bse/bseenginemaster.cc b/bse/bseenginemaster.cc
index 917123d..cd65272 100644
--- a/bse/bseenginemaster.cc
+++ b/bse/bseenginemaster.cc
@@ -156,10 +156,10 @@ master_jdisconnect_node (EngineNode *node,
 static void
 master_disconnect_node_outputs (EngineNode *src_node, EngineNode *dest_node)
 {
-  for (int i = 0; i < ENGINE_NODE_N_ISTREAMS (dest_node); i++)
+  for (uint i = 0; i < ENGINE_NODE_N_ISTREAMS (dest_node); i++)
     if (dest_node->inputs[i].src_node == src_node)
       master_idisconnect_node (dest_node, i);
-  for (int j = 0; j < ENGINE_NODE_N_JSTREAMS (dest_node); j++)
+  for (uint j = 0; j < ENGINE_NODE_N_JSTREAMS (dest_node); j++)
     for (uint i = 0; i < dest_node->module.jstreams[j].jcount; i++)
       if (dest_node->jinputs[j][i].src_node == src_node)
        master_jdisconnect_node (dest_node, j, i--);
@@ -1040,11 +1040,24 @@ _engine_master_dispatch (void)
   if (master_need_process)
     master_process_flow ();
 }
+
+namespace Bse {
+
+MasterThread::MasterThread (const std::function<void()> &caller_wakeup) :
+  caller_wakeup_ (caller_wakeup)
+{
+  assert (caller_wakeup_ != NULL);
+  if (event_fd_.open() != 0)
+    g_error ("failed to create engine wake-up pipe: %s", strerror (errno));
+  thread_ = std::thread (&MasterThread::master_thread, this); // FIXME: join on exit
+}
+
 void
-bse_engine_master_thread (EngineMasterData *mdata)
+MasterThread::master_thread()
 {
   Bse::TaskRegistry::add ("DSP #1", Rapicorn::ThisThread::process_pid(), Rapicorn::ThisThread::thread_pid());
   bse_message_setup_thread_handler ();
+
   /* assert pollfd equality, since we're simply casting structures */
   BIRNET_STATIC_ASSERT (sizeof (struct pollfd) == sizeof (GPollFD));
   BIRNET_STATIC_ASSERT (G_STRUCT_OFFSET (GPollFD, fd) == G_STRUCT_OFFSET (struct pollfd, fd));
@@ -1053,22 +1066,24 @@ bse_engine_master_thread (EngineMasterData *mdata)
   BIRNET_STATIC_ASSERT (sizeof (((GPollFD*) 0)->events) == sizeof (((struct pollfd*) 0)->events));
   BIRNET_STATIC_ASSERT (G_STRUCT_OFFSET (GPollFD, revents) == G_STRUCT_OFFSET (struct pollfd, revents));
   BIRNET_STATIC_ASSERT (sizeof (((GPollFD*) 0)->revents) == sizeof (((struct pollfd*) 0)->revents));
+
   /* add the thread wakeup pipe to master pollfds,
    * so we get woken  up in time.
    */
-  master_pollfds[0].fd = mdata->wakeup_pipe[0];
+  master_pollfds[0].fd = event_fd_.inputfd();
   master_pollfds[0].events = G_IO_IN;
   master_n_pollfds = 1;
   master_pollfds_changed = TRUE;
   toyprof_stampinit ();
-  while (!sfi_thread_aborted ())        /* also updates accounting information */
+  while (1)
     {
       BseEngineLoop loop;
-      gboolean need_dispatch;
+      bool need_dispatch;
       need_dispatch = _engine_master_prepare (&loop);
+      master_pollfds[0].revents = 0;
       if (!need_dispatch)
        {
-         gint err = poll ((struct pollfd*) loop.fds, loop.n_fds, loop.timeout);
+         int err = poll ((struct pollfd*) loop.fds, loop.n_fds, loop.timeout);
          if (err >= 0)
            loop.revents_filled = TRUE;
          else if (errno != EINTR)
@@ -1078,18 +1093,13 @@ bse_engine_master_thread (EngineMasterData *mdata)
        }
       if (need_dispatch)
        _engine_master_dispatch ();
-      /* clear wakeup pipe */
-      {
-       guint8 data[64];
-       gint l;
-       do
-         l = read (mdata->wakeup_pipe[0], data, sizeof (data));
-       while ((l < 0 && errno == EINTR) || l == sizeof (data));
-      }
-      /* wakeup user thread if necessary */
+      if (master_pollfds[0].revents)    // need to clear wakeup pipe
+        event_fd_.flush();
+      // wakeup user thread if necessary
       if (bse_engine_has_garbage ())
-       sfi_thread_wakeup (mdata->user_thread);
+       caller_wakeup_();
     }
   Bse::TaskRegistry::remove (Rapicorn::ThisThread::thread_pid());
 }
-/* vim:set ts=8 sts=2 sw=2: */
+
+} // Bse
diff --git a/bse/bseenginemaster.hh b/bse/bseenginemaster.hh
index 9598222..d921b6c 100644
--- a/bse/bseenginemaster.hh
+++ b/bse/bseenginemaster.hh
@@ -3,15 +3,28 @@
 #define __BSE_ENGINE_MASTER_H__
 #include <bse/bseengine.hh>
 G_BEGIN_DECLS
+
 /* --- internal (EngineThread) --- */
 gboolean       _engine_master_prepare          (BseEngineLoop          *loop);
 gboolean       _engine_master_check            (const BseEngineLoop    *loop);
 void           _engine_master_dispatch_jobs    (void);
 void           _engine_master_dispatch         (void);
-typedef struct {
-  BirnetThread *user_thread;
-  gint       wakeup_pipe[2];   /* read(wakeup_pipe[0]), write(wakeup_pipe[1]) */
-} EngineMasterData;
-void           bse_engine_master_thread        (EngineMasterData       *mdata);
+
+namespace Bse {
+
+using Rapicorn::EventFd;    // FIXME
+
+class MasterThread {
+  std::thread           thread_;
+  EventFd               event_fd_;
+  std::function<void()> caller_wakeup_;
+  void          master_thread   ();
+public:
+  explicit      MasterThread    (const std::function<void()> &caller_wakeup);
+  void          wakeup          ()      { event_fd_.wakeup(); }
+};
+
+} // Bse
+
 G_END_DECLS
 #endif /* __BSE_ENGINE_MASTER_H__ */