3 files changed, 27 insertions, 90 deletions
diff --git a/gnuradio-runtime/lib/basic_block.cc b/gnuradio-runtime/lib/basic_block.cc
index 082d0753c8..89aa9b8671 100644
--- a/gnuradio-runtime/lib/basic_block.cc
+++ b/gnuradio-runtime/lib/basic_block.cc
@@ -228,29 +228,6 @@ namespace gr {
   }
 
   pmt::pmt_t
-  basic_block::delete_head_blocking(pmt::pmt_t which_port, unsigned int millisec)
-  {
-    gr::thread::scoped_lock guard(mutex);
-
-    if (millisec) {
-       boost::system_time const timeout = boost::get_system_time() + boost::posix_time::milliseconds(millisec);
-       while (empty_p(which_port)) {
-         if (!msg_queue_ready[which_port]->timed_wait(guard, timeout)) {
-           return pmt::pmt_t();
-	 }
-       }
-    } else {
-      while(empty_p(which_port)) {
-        msg_queue_ready[which_port]->wait(guard);
-      }
-    }
-
-    pmt::pmt_t m(msg_queue[which_port].front());
-    msg_queue[which_port].pop_front();
-    return m;
-  }
-
-  pmt::pmt_t
   basic_block::message_subscribers(pmt::pmt_t port)
   {
     return pmt::dict_ref(d_message_subscribers,port,pmt::PMT_NIL);
diff --git a/gnuradio-runtime/lib/block.cc b/gnuradio-runtime/lib/block.cc
index 2bae8ea9f8..e329e0657a 100644
--- a/gnuradio-runtime/lib/block.cc
+++ b/gnuradio-runtime/lib/block.cc
@@ -707,8 +707,7 @@ namespace gr {
     //std::cout << "system_handler " << msg << "\n";
     pmt::pmt_t op = pmt::car(msg);
     if(pmt::eqv(op, pmt::mp("done"))){
-        d_finished = pmt::to_long(pmt::cdr(msg));
-        global_block_registry.notify_blk(alias());
+        d_finished = pmt::to_bool(pmt::cdr(msg));
     } else {
         std::cout << "WARNING: bad message op on system port!\n";
         pmt::print(msg);
@@ -737,7 +736,7 @@ namespace gr {
 
         currlist = pmt::cdr(currlist);
         basic_block_sptr blk = global_block_registry.block_lookup(block);
-        blk->post(port, pmt::cons(pmt::mp("done"), pmt::mp(true)));
+        blk->post(port, pmt::cons(pmt::mp("done"), pmt::PMT_T));
 
         //std::cout << "notify finished --> ";
         //pmt::print(pmt::cons(block,port));
@@ -750,7 +749,7 @@ namespace gr {
   bool
   block::finished()
   {
-    if((detail()->ninputs() != 0) || (detail()->noutputs() != 0))
+    if(detail()->ninputs() != 0)
       return false;
     else
       return d_finished;
diff --git a/gnuradio-runtime/lib/tpb_thread_body.cc b/gnuradio-runtime/lib/tpb_thread_body.cc
index 93591feee2..32c846d8a4 100644
--- a/gnuradio-runtime/lib/tpb_thread_body.cc
+++ b/gnuradio-runtime/lib/tpb_thread_body.cc
@@ -93,9 +93,10 @@ namespace gr {
     block->clear_finished();
 
     while(1) {
-      tpb_loop_top:
       boost::this_thread::interruption_point();
 
+      d->d_tpb.clear_changed();
+
       // handle any queued up messages
       BOOST_FOREACH(basic_block::msg_queue_map_t::value_type &i, block->msg_queue) {
         // Check if we have a message handler attached before getting
@@ -116,97 +117,57 @@ namespace gr {
         }
       }
 
-      d->d_tpb.clear_changed();
       // run one iteration if we are a connected stream block
       if(d->noutputs() >0 || d->ninputs()>0){
         s = d_exec.run_one_iteration();
       }
       else {
         s = block_executor::BLKD_IN;
+        // a msg port only block wants to shutdown
+        if(block->finished()) {
+          s = block_executor::DONE;
+        }
       }
 
-      // if msg ports think we are done, we are done
-      if(block->finished())
+      if(block->finished() && s == block_executor::READY_NO_OUTPUT) {
         s = block_executor::DONE;
+        d->set_done(true);
+      }
+
+      if(!d->ninputs() && s == block_executor::READY_NO_OUTPUT) {
+        s = block_executor::BLKD_IN;
+      }
 
       switch(s){
-      case block_executor::READY:		// Tell neighbors we made progress.
+      case block_executor::READY:              // Tell neighbors we made progress.
         d->d_tpb.notify_neighbors(d);
         break;
 
-      case block_executor::READY_NO_OUTPUT:	// Notify upstream only
+      case block_executor::READY_NO_OUTPUT:    // Notify upstream only
         d->d_tpb.notify_upstream(d);
         break;
 
-      case block_executor::DONE:		// Game over.
+      case block_executor::DONE:               // Game over.
         block->notify_msg_neighbors();
         d->d_tpb.notify_neighbors(d);
         return;
 
-      case block_executor::BLKD_IN:		// Wait for input.
+      case block_executor::BLKD_IN:            // Wait for input.
       {
         gr::thread::scoped_lock guard(d->d_tpb.mutex);
-        while(!d->d_tpb.input_changed) {
-
-          // wait for input or message
-          while(!d->d_tpb.input_changed && block->empty_handled_p()){
-            boost::system_time const timeout=boost::get_system_time()+ boost::posix_time::milliseconds(250);
-            if(!d->d_tpb.input_cond.timed_wait(guard, timeout)){
-              goto tpb_loop_top; // timeout occurred (perform sanity checks up top)
-            }
-          }
 
-          // handle all pending messages
-          BOOST_FOREACH(basic_block::msg_queue_map_t::value_type &i, block->msg_queue) {
-            if(block->has_msg_handler(i.first)) {
-              while((msg = block->delete_head_nowait(i.first))) {
-                guard.unlock();			// release lock while processing msg
-                block->dispatch_msg(i.first, msg);
-                guard.lock();
-              }
-            }
-            else {
-              // leave msg in queue if no handler is defined
-              // start dropping if we have too many
-              if(block->nmsgs(i.first) > max_nmsgs){
-                GR_LOG_WARN(LOG,"asynchronous message buffer overflowing, dropping message");
-                msg = block->delete_head_nowait(i.first);
-              }
-            }
-          }
-	  if (d->done()) {
-	    return;
-	  }
+        if(!d->d_tpb.input_changed) {
+          boost::system_time const timeout=boost::get_system_time()+ boost::posix_time::milliseconds(250);
+          d->d_tpb.input_cond.timed_wait(guard, timeout);
         }
       }
       break;
 
-      case block_executor::BLKD_OUT:	// Wait for output buffer space.
+      case block_executor::BLKD_OUT:           // Wait for output buffer space.
       {
-	gr::thread::scoped_lock guard(d->d_tpb.mutex);
-	while(!d->d_tpb.output_changed) {
-	  // wait for output room or message
-	  while(!d->d_tpb.output_changed && block->empty_handled_p())
-	    d->d_tpb.output_cond.wait(guard);
-
-	  // handle all pending messages
-          BOOST_FOREACH(basic_block::msg_queue_map_t::value_type &i, block->msg_queue) {
-            if(block->has_msg_handler(i.first)) {
-                while((msg = block->delete_head_nowait(i.first))) {
-                  guard.unlock();			// release lock while processing msg
-                  block->dispatch_msg(i.first, msg);
-                  guard.lock();
-                }
-            }
-            else {
-                // leave msg in queue if no handler is defined
-                // start dropping if we have too many
-                if(block->nmsgs(i.first) > max_nmsgs){
-                  GR_LOG_WARN(LOG,"asynchronous message buffer overflowing, dropping message");
-                  msg = block->delete_head_nowait(i.first);
-                }
-            }
-          }
+        gr::thread::scoped_lock guard(d->d_tpb.mutex);
+        while(!d->d_tpb.output_changed) {
+            d->d_tpb.output_cond.wait(guard);
         }
       }
       break;