[Commit-gnuradio] r4916 - gnuradio/branches/developers/eb/ibu/mblock/src/lib

[eb]

Author: eb
Date: 2007-04-07 12:39:52 -0600 (Sat, 07 Apr 2007)
New Revision: 4916

Modified:
   gnuradio/branches/developers/eb/ibu/mblock/src/lib/mb_mblock.cc
   gnuradio/branches/developers/eb/ibu/mblock/src/lib/mb_mblock.h
   gnuradio/branches/developers/eb/ibu/mblock/src/lib/mb_mblock_impl.cc
   gnuradio/branches/developers/eb/ibu/mblock/src/lib/mb_mblock_impl.h
   gnuradio/branches/developers/eb/ibu/mblock/src/lib/mb_runtime.cc
   gnuradio/branches/developers/eb/ibu/mblock/src/lib/mb_runtime.h
   
gnuradio/branches/developers/eb/ibu/mblock/src/lib/mb_runtime_thread_per_block.cc
   
gnuradio/branches/developers/eb/ibu/mblock/src/lib/mb_runtime_thread_per_block.h
   gnuradio/branches/developers/eb/ibu/mblock/src/lib/mb_worker.cc
   gnuradio/branches/developers/eb/ibu/mblock/src/lib/mb_worker.h
Log:
work-in-progress on mblocks

Modified: gnuradio/branches/developers/eb/ibu/mblock/src/lib/mb_mblock.cc
===================================================================
--- gnuradio/branches/developers/eb/ibu/mblock/src/lib/mb_mblock.cc     
2007-04-07 14:23:09 UTC (rev 4915)
+++ gnuradio/branches/developers/eb/ibu/mblock/src/lib/mb_mblock.cc     
2007-04-07 18:39:52 UTC (rev 4916)
@@ -25,8 +25,13 @@
 
 #include <mb_mblock.h>
 #include <mb_mblock_impl.h>
+#include <mb_runtime.h>
+#include <mb_exception.h>
 
 
+static pmt_t s_sys_port = pmt_intern("%sys-port");
+static pmt_t s_halt = pmt_intern("%halt");
+
 mb_visitor::~mb_visitor()
 {
   // nop base case for virtual destructor.
@@ -60,6 +65,31 @@
   // default implementation does nothing
 }
 
+
+void
+mb_mblock::main_loop()
+{
+  while (1){
+    mb_message_sptr msg;
+    try {
+      while (1){
+       msg = impl()->msgq().get_highest_pri_msg();
+
+       // check for %halt from %sys-port
+       if (pmt_eq(msg->port_id(), s_sys_port) && pmt_eq(msg->signal(), s_halt))
+         exit();
+
+       handle_message(msg);
+      }
+    }
+    catch (pmt_exception e){
+      std::cerr << std::endl << instance_name()
+               << "::main_loop: ignoring pmt_exception: "
+               << e.what() << std::endl;
+    }
+  }
+}
+
 ////////////////////////////////////////////////////////////////////////
 //           Forward other methods to implementation class            //
 ////////////////////////////////////////////////////////////////////////
@@ -157,11 +187,12 @@
 void
 mb_mblock::exit()
 {
-  d_impl->exit();
+  throw mbe_exit();    // adios...
 }
 
 void
-mb_mblock::terminate_all()
+mb_mblock::shutdown_all()
 {
-  d_impl->terminate_all();
+  d_impl->runtime()->request_shutdown();
+  exit();
 }

Modified: gnuradio/branches/developers/eb/ibu/mblock/src/lib/mb_mblock.h
===================================================================
--- gnuradio/branches/developers/eb/ibu/mblock/src/lib/mb_mblock.h      
2007-04-07 14:23:09 UTC (rev 4915)
+++ gnuradio/branches/developers/eb/ibu/mblock/src/lib/mb_mblock.h      
2007-04-07 18:39:52 UTC (rev 4916)
@@ -52,6 +52,7 @@
 
   friend class mb_runtime;
   friend class mb_mblock_impl;
+  friend class mb_worker;
 
 protected:
   /*!
@@ -189,27 +190,41 @@
   /*!
    * \brief Tell runtime that we are done.
    *
-   * runtime->run(...) returns when all blocks have called exit.
+   * This method does not return.
    */
   void exit();
 
   /*!
-   * \brief Ask runtime to start termination procedure for all blocks.
+   * \brief Ask runtime to execute the shutdown procedure for all blocks.
    * 
-   * The runtime first sends a maximum priority %Terminate message to
-   * all blocks.  All blocks should handle the %Terminate message,
+   * The runtime first sends a maximum priority %shutdown message to
+   * all blocks.  All blocks should handle the %shutdown message,
    * perform whatever clean up is required, and call this->exit();
    *
-   * After a period of time (~200ms), any blocks which haven't yet
-   * called this->exit() are sent a SIG<FOO> (TBD) signal, which will
-   * blow them out of any blocking system calls and raise a
-   * mbe_terminate exception.  The default top-level runtime-provided
-   * exception handler will call this->exit() to finish the process.
+   * After a period of time (~100ms), any blocks which haven't yet
+   * called this->exit() are sent a maximum priority %halt message.
+   * %halt is detected in main_loop, and this->exit() is called.
    *
+   * After an additional period of time (~100ms), any blocks which
+   * still haven't yet called this->exit() are sent a SIG<FOO> (TBD)
+   * signal, which will blow them out of any blocking system calls and
+   * raise an mbe_terminate exception.  The default top-level
+   * runtime-provided exception handler will call this->exit() to
+   * finish the process.
+   *
+   * This method does not return.
+   *
    * runtime->run(...) returns when all blocks have called exit.
    */
-  void terminate_all();
+  void shutdown_all();
 
+  /*!
+   * \brief main event dispatching loop
+   *
+   * Although it is possible to override this, the default implementation
+   * should work for virtually all cases.
+   */
+  virtual void main_loop();
   
 public:
   virtual ~mb_mblock();

Modified: gnuradio/branches/developers/eb/ibu/mblock/src/lib/mb_mblock_impl.cc
===================================================================
--- gnuradio/branches/developers/eb/ibu/mblock/src/lib/mb_mblock_impl.cc        
2007-04-07 14:23:09 UTC (rev 4915)
+++ gnuradio/branches/developers/eb/ibu/mblock/src/lib/mb_mblock_impl.cc        
2007-04-07 18:39:52 UTC (rev 4916)
@@ -294,15 +294,3 @@
 {
   d_class_name = name;
 }
-
-void
-mb_mblock_impl::exit()
-{
-  throw mbe_exit();    // adios...
-}
-
-void
-mb_mblock_impl::terminate_all()
-{
-  // FIXME
-}

Modified: gnuradio/branches/developers/eb/ibu/mblock/src/lib/mb_mblock_impl.h
===================================================================
--- gnuradio/branches/developers/eb/ibu/mblock/src/lib/mb_mblock_impl.h 
2007-04-07 14:23:09 UTC (rev 4915)
+++ gnuradio/branches/developers/eb/ibu/mblock/src/lib/mb_mblock_impl.h 
2007-04-07 18:39:52 UTC (rev 4916)
@@ -142,30 +142,6 @@
   int
   nconnections();
 
-  /*!
-   * \brief Tell runtime that we are done.
-   *
-   * runtime->run(...) returns when all blocks have called exit.
-   */
-  void exit();
-
-  /*!
-   * \brief Ask runtime to start termination procedure for all blocks.
-   * 
-   * The runtime first sends a maximum priority %Terminate message to
-   * all blocks.  All blocks should handle the %Terminate message,
-   * perform whatever clean up is required, and call this->exit();
-   *
-   * After a period of time (~200ms), any blocks which haven't yet
-   * called this->exit() are sent a SIG<FOO> (TBD) signal, which will
-   * blow them out of any blocking system calls and raise a
-   * mbe_terminate exception.  The default top-level runtime-provided
-   * exception handler will call this->exit() to finish the process.
-   *
-   * runtime->run(...) returns when all blocks have called exit.
-   */
-  void terminate_all();
-
   bool
   walk_tree(mb_visitor *visitor);
   

Modified: gnuradio/branches/developers/eb/ibu/mblock/src/lib/mb_runtime.cc
===================================================================
--- gnuradio/branches/developers/eb/ibu/mblock/src/lib/mb_runtime.cc    
2007-04-07 14:23:09 UTC (rev 4915)
+++ gnuradio/branches/developers/eb/ibu/mblock/src/lib/mb_runtime.cc    
2007-04-07 18:39:52 UTC (rev 4916)
@@ -36,3 +36,9 @@
 {
   // nop
 }
+
+void
+mb_runtime::request_shutdown()
+{
+  // nop
+}

Modified: gnuradio/branches/developers/eb/ibu/mblock/src/lib/mb_runtime.h
===================================================================
--- gnuradio/branches/developers/eb/ibu/mblock/src/lib/mb_runtime.h     
2007-04-07 14:23:09 UTC (rev 4915)
+++ gnuradio/branches/developers/eb/ibu/mblock/src/lib/mb_runtime.h     
2007-04-07 18:39:52 UTC (rev 4916)
@@ -78,6 +78,8 @@
    */
   inline void unlock() { d_brl.unlock(); }
 
+  virtual void request_shutdown();
+
 protected:
   virtual mb_mblock_sptr
   create_component(const std::string &instance_name,

Modified: 
gnuradio/branches/developers/eb/ibu/mblock/src/lib/mb_runtime_thread_per_block.cc
===================================================================
--- 
gnuradio/branches/developers/eb/ibu/mblock/src/lib/mb_runtime_thread_per_block.cc
   2007-04-07 14:23:09 UTC (rev 4915)
+++ 
gnuradio/branches/developers/eb/ibu/mblock/src/lib/mb_runtime_thread_per_block.cc
   2007-04-07 18:39:52 UTC (rev 4916)
@@ -32,7 +32,7 @@
 
 
 mb_runtime_thread_per_block::mb_runtime_thread_per_block()
-  : d_runtime_cond(&d_mutex)
+  : d_runtime_cond(&d_mutex), d_shutdown_requested(false)
 {
   // nop for now
 }
@@ -40,8 +40,19 @@
 mb_runtime_thread_per_block::~mb_runtime_thread_per_block()
 {
   // FIXME iterate over workers and ensure that they are dead.
+
+  std::cerr << "\nmb_runtime_thread_per_block: dtor (# workers = "
+           << d_workers.size() << ")\n";
 }
 
+void
+mb_runtime_thread_per_block::request_shutdown()
+{
+  omni_mutex_lock l1(d_mutex);
+  d_shutdown_requested = true;
+  d_runtime_cond.broadcast();
+}
+
 bool
 mb_runtime_thread_per_block::run(const std::string &instance_name,
                                 const std::string &class_name,
@@ -51,26 +62,34 @@
   // subcomponents.
   mb_mblock_sptr top = create_component(instance_name, class_name, user_arg);
 
-  // FIXME!  Rethink when to run initial_transitions.
+  while (1){
+    omni_mutex_lock l1(d_mutex);
 
-  // Now tell them all to run their initial_transitions
-  {
-    omni_mutex_lock    l1(d_mutex);      // lock runtime first...
+    //
+    // reap dead threads
+    //
+    for (worker_iter_t wi = d_workers.begin(); wi != d_workers.end(); ){
+      omni_mutex_lock l2((*wi)->d_mutex);
+      if ((*wi)->d_state == mb_worker::TS_DEAD){
+       std::cerr << "\nruntime: "
+                 << (*wi)->d_mblock->instance_name() << "is TS_DEAD\n";
+       void *ignore;
+       (*wi)->join(&ignore);
+       wi = d_workers.erase(wi);
+       continue;
+      }
+      ++wi;
+    }
 
-    for (worker_iter_t w = d_workers.begin(); w != d_workers.end(); ++w){
-      omni_mutex_lock  l2((*w)->d_mutex);        // ...then worker.
-      switch ((*w)->d_state){
+    if (d_workers.empty())     // no work left to do...
+      return true;             
 
-      case mb_worker::TS_CONSTRUCTED:          // expected case
-       (*w)->d_state = mb_worker::TS_RUN_INITIAL;
-       (*w)->d_state_cond.broadcast();
-       break;
+    if (d_shutdown_requested){
+      // FIXME implement something...
+      continue;
+    }
 
-      default:
-       // FIXME...
-       break;
-      }
-    }
+    d_runtime_cond.wait();     // wait for something to do.
   }
 
   return true;
@@ -98,11 +117,11 @@
 
   w->start_undetached();  // start it
 
-  // Wait for it to reach TS_CONSTRUCTED or TS_DEAD
+  // Wait for it to reach TS_RUNNING or TS_DEAD
   {
     omni_mutex_lock l(w->d_mutex);
-    while (!(w->d_state != mb_worker::TS_CONSTRUCTED
-            || w->d_state != mb_worker::TS_DEAD))
+    while (!(w->d_state == mb_worker::TS_RUNNING
+            || w->d_state == mb_worker::TS_DEAD))
       w->d_state_cond.wait();
   }
 
@@ -125,3 +144,4 @@
 
   return w->d_mblock;
 }
+

Modified: 
gnuradio/branches/developers/eb/ibu/mblock/src/lib/mb_runtime_thread_per_block.h
===================================================================
--- 
gnuradio/branches/developers/eb/ibu/mblock/src/lib/mb_runtime_thread_per_block.h
    2007-04-07 14:23:09 UTC (rev 4915)
+++ 
gnuradio/branches/developers/eb/ibu/mblock/src/lib/mb_runtime_thread_per_block.h
    2007-04-07 18:39:52 UTC (rev 4916)
@@ -36,6 +36,7 @@
   omni_mutex                 d_mutex;
   omni_condition             d_runtime_cond;  // runtime waits here
   std::vector<mb_worker_sptr> d_workers;
+  bool                       d_shutdown_requested;
 
   typedef std::vector<mb_worker_sptr>::iterator  worker_iter_t;
 
@@ -46,6 +47,8 @@
           const std::string &class_name,
           pmt_t user_arg);
 
+  void request_shutdown();
+
 protected:
   mb_mblock_sptr
   create_component(const std::string &instance_name,

Modified: gnuradio/branches/developers/eb/ibu/mblock/src/lib/mb_worker.cc
===================================================================
--- gnuradio/branches/developers/eb/ibu/mblock/src/lib/mb_worker.cc     
2007-04-07 14:23:09 UTC (rev 4915)
+++ gnuradio/branches/developers/eb/ibu/mblock/src/lib/mb_worker.cc     
2007-04-07 18:39:52 UTC (rev 4916)
@@ -71,7 +71,6 @@
   d_runtime->d_runtime_cond.broadcast();
 }
 
-
 void *
 mb_worker::run_undetached(void *ignored)
 {
@@ -105,48 +104,31 @@
            << "  gettid:        " << mb_gettid() << std::endl
            << "  getpid:        " << getpid() << std::endl;
 
+  cause_of_death_t pending_cause_of_death = RIP_NOT_DEAD_YET;
+  
   try {
+    pending_cause_of_death = RIP_CTOR_EXCEPTION;
     d_mblock = d_maker(d_runtime, d_instance_name, d_user_arg);
-  }
-  catch (...){
-    d_why_dead = RIP_CTOR_EXCEPTION;
-    throw;
-  }
 
-  // We've got an mblock.  Let runtime know we're good so far.
-  set_state(TS_CONSTRUCTED);
+    std::cerr << "worker_thread_top_level (post-construction):" << std::endl
+             << "  instance_name: " << d_instance_name << std::endl;
 
+    pending_cause_of_death = RIP_INIT_EXCEPTION;
+    d_mblock->initial_transition();
 
-  std::cerr << "worker_thread_top_level (post-construction):" << std::endl
-           << "  instance_name: " << d_instance_name << std::endl;
+    std::cerr << "worker_thread_top_level (post-initial-transition):" << 
std::endl
+             << "  instance_name: " << d_instance_name << std::endl;
 
-  // Wait for runtime to change our state to TS_RUN_INITIAL.
-  {
-    omni_mutex_lock l(d_mutex);
-    while (d_state != TS_RUN_INITIAL)
-      d_state_cond.wait();
-  }
+    set_state(TS_RUNNING);
 
-  std::cerr << "worker_thread_top_level (got RUN_INITIAL):" << std::endl
-           << "  instance_name: " << d_instance_name << std::endl;
-
-  try {
-    d_mblock->initial_transition();
+    pending_cause_of_death = RIP_UNHANDLED_EXCEPTION;
+    d_mblock->main_loop();
   }
   catch (...){
-    d_why_dead = RIP_INIT_EXCEPTION;
+    d_why_dead = pending_cause_of_death;
     throw;
   }
 
-  // initial_transition was OK, set state to TS_RUNNING
-  set_state(TS_RUNNING);
-
-  std::cerr << "worker_thread_top_level (post-initial-transition):" << 
std::endl
-           << "  instance_name: " << d_instance_name << std::endl;
-
-  // FIXME run the main_loop here
-  sleep(5);
-
   std::cerr << "worker_thread_top_level (exit):" << std::endl
            << "  instance_name: " << d_instance_name << std::endl;
 }

Modified: gnuradio/branches/developers/eb/ibu/mblock/src/lib/mb_worker.h
===================================================================
--- gnuradio/branches/developers/eb/ibu/mblock/src/lib/mb_worker.h      
2007-04-07 14:23:09 UTC (rev 4915)
+++ gnuradio/branches/developers/eb/ibu/mblock/src/lib/mb_worker.h      
2007-04-07 18:39:52 UTC (rev 4916)
@@ -42,8 +42,6 @@
   //! worker thread states
   enum worker_state_t {
     TS_UNINITIALIZED,  // new, uninitialized
-    TS_CONSTRUCTED,    // mblock was successfully constructed by thread
-    TS_RUN_INITIAL,    // thread should run initial_transition
     TS_RUNNING,                // normal steady-state condition.
     TS_DEAD            // thread is dead
   };