Merged r5230:5237 from jcorgan/disc2. Trunk passes distcheck.

git-svn-id: http://gnuradio.org/svn/gnuradio/trunk@5238 221aa14e-8319-0410-a670-987f0aec2ac5

12 files changed, 418 insertions, 128 deletions

diff --git a/gnuradio-core/src/lib/runtime/gr_block_detail.h b/gnuradio-core/src/lib/runtime/gr_block_detail.h
index d89b7fea1..9aa5ef4b0 100644
--- a/gnuradio-core/src/lib/runtime/gr_block_detail.h
+++ b/gnuradio-core/src/lib/runtime/gr_block_detail.h
@@ -55,6 +55,14 @@ class gr_block_detail {
     return d_input[which];
   }
 
+  void clear_input (unsigned int which)
+  {
+    if (which >= d_ninputs)
+      throw std::invalid_argument ("gr_block_detail::input");
+    if (d_input[which])
+      d_input[which].reset();
+  }
+
   void set_output (unsigned int which, gr_buffer_sptr buffer);
   gr_buffer_sptr output (unsigned int which)
   {
diff --git a/gnuradio-core/src/lib/runtime/gr_block_detail.i b/gnuradio-core/src/lib/runtime/gr_block_detail.i
index 9ac5ae96d..6a01482c6 100644
--- a/gnuradio-core/src/lib/runtime/gr_block_detail.i
+++ b/gnuradio-core/src/lib/runtime/gr_block_detail.i
@@ -46,6 +46,14 @@ class gr_block_detail {
     return d_input[which];
   }
 
+  void clear_input (unsigned int which)
+  {
+    if (which >= d_ninputs)
+      throw std::invalid_argument ("gr_block_detail::input");
+    if (d_input[which])
+      d_input[which].reset();
+  }
+
   void set_output (unsigned int which, gr_buffer_sptr buffer);
   gr_buffer_sptr output (unsigned int which)
   {
diff --git a/gnuradio-core/src/lib/runtime/gr_buffer.h b/gnuradio-core/src/lib/runtime/gr_buffer.h
index 46017f6ad..51474c0e7 100644
--- a/gnuradio-core/src/lib/runtime/gr_buffer.h
+++ b/gnuradio-core/src/lib/runtime/gr_buffer.h
@@ -157,6 +157,12 @@ class gr_buffer_reader {
   int items_available () const;
 
   /*!
+   * \brief Return buffer this reader reads from.
+   */
+  gr_buffer_sptr buffer () const { return d_buffer; }
+
+
+  /*!
    * \brief Return maximum number of items that could ever be available for reading.
    * This is used as a sanity check in the scheduler to avoid looping forever.
    */
diff --git a/gnuradio-core/src/lib/runtime/gr_hier_block2_detail.cc b/gnuradio-core/src/lib/runtime/gr_hier_block2_detail.cc
index b41d0c65c..af029bdc7 100644
--- a/gnuradio-core/src/lib/runtime/gr_hier_block2_detail.cc
+++ b/gnuradio-core/src/lib/runtime/gr_hier_block2_detail.cc
@@ -55,20 +55,20 @@ gr_hier_block2_detail::connect(gr_basic_block_sptr src, int src_port,
   if (src.get() == dst.get())
     throw std::invalid_argument("src and destination blocks cannot be the same");
 
-  // Assignments to block inputs or outputs
+  // Connectments to block inputs or outputs
   int max_port;
   if (src.get() == d_owner) {
     max_port = src->input_signature()->max_streams();
     if ((max_port != -1 && (src_port >= max_port)) || src_port < 0)
       throw std::invalid_argument("source port out of range");
-    return assign_input(src_port, dst_port, dst);
+    return connect_input(src_port, dst_port, dst);
   }
 
   if (dst.get() == d_owner) {
     max_port = dst->output_signature()->max_streams();
     if ((max_port != -1 && (dst_port >= max_port)) || dst_port < 0)
       throw std::invalid_argument("source port out of range");
-    return assign_output(dst_port, src_port, src);
+    return connect_output(dst_port, src_port, src);
   }
 
   // Internal connections
@@ -81,13 +81,25 @@ void
 gr_hier_block2_detail::disconnect(gr_basic_block_sptr src, int src_port, 
                                   gr_basic_block_sptr dst, int dst_port)
 {
-  // Handle disconnecting inputs and outputs
+  if (GR_HIER_BLOCK2_DETAIL_DEBUG)
+    std::cout << "disconnecting: " << gr_endpoint(src, src_port)
+              << " -> " << gr_endpoint(dst, dst_port) << std::endl;
+
+  if (src.get() == dst.get())
+    throw std::invalid_argument("src and destination blocks cannot be the same");
+
+  if (src.get() == d_owner)
+    return disconnect_input(src_port, dst_port, dst);
 
+  if (dst.get() == d_owner)
+    return disconnect_output(dst_port, src_port, src);
+
+  // Internal connections
   d_fg->disconnect(src, src_port, dst, dst_port);
 }
 
 void
-gr_hier_block2_detail::assign_input(int my_port, int port, gr_basic_block_sptr block)
+gr_hier_block2_detail::connect_input(int my_port, int port, gr_basic_block_sptr block)
 {
   if (my_port < 0 || my_port >= (signed)d_inputs.size())
     throw std::invalid_argument("input port number out of range");
@@ -99,7 +111,7 @@ gr_hier_block2_detail::assign_input(int my_port, int port, gr_basic_block_sptr b
 }
 
 void
-gr_hier_block2_detail::assign_output(int my_port, int port, gr_basic_block_sptr block)
+gr_hier_block2_detail::connect_output(int my_port, int port, gr_basic_block_sptr block)
 {
   if (my_port < 0 || my_port >= (signed)d_outputs.size())
     throw std::invalid_argument("output port number out of range");
@@ -110,6 +122,30 @@ gr_hier_block2_detail::assign_output(int my_port, int port, gr_basic_block_sptr
   d_outputs[my_port] = gr_endpoint(block, port);
 }
 
+void
+gr_hier_block2_detail::disconnect_input(int my_port, int port, gr_basic_block_sptr block)
+{
+  if (my_port < 0 || my_port >= (signed)d_inputs.size())
+    throw std::invalid_argument("input port number out of range");
+
+  if (d_inputs[my_port].block() != block)
+    throw std::invalid_argument("block not assigned to given input, can't disconnect");
+
+  d_inputs[my_port] = gr_endpoint();
+}
+
+void
+gr_hier_block2_detail::disconnect_output(int my_port, int port, gr_basic_block_sptr block)
+{
+  if (my_port < 0 || my_port >= (signed)d_outputs.size())
+    throw std::invalid_argument("input port number out of range");
+
+  if (d_outputs[my_port].block() != block)
+    throw std::invalid_argument("block not assigned to given output, can't disconnect");
+
+  d_outputs[my_port] = gr_endpoint();
+}
+
 gr_endpoint
 gr_hier_block2_detail::resolve_port(int port, bool is_input)
 {
diff --git a/gnuradio-core/src/lib/runtime/gr_hier_block2_detail.h b/gnuradio-core/src/lib/runtime/gr_hier_block2_detail.h
index f0e7b7e82..15c8548c6 100644
--- a/gnuradio-core/src/lib/runtime/gr_hier_block2_detail.h
+++ b/gnuradio-core/src/lib/runtime/gr_hier_block2_detail.h
@@ -45,8 +45,10 @@ private:
                  gr_basic_block_sptr dst, int dst_port);
     void disconnect(gr_basic_block_sptr, int src_port, 
                     gr_basic_block_sptr, int dst_port);
-    void assign_input(int my_port, int port, gr_basic_block_sptr block);
-    void assign_output(int my_port, int port, gr_basic_block_sptr block);
+    void connect_input(int my_port, int port, gr_basic_block_sptr block);
+    void connect_output(int my_port, int port, gr_basic_block_sptr block);
+    void disconnect_input(int my_port, int port, gr_basic_block_sptr block);
+    void disconnect_output(int my_port, int port, gr_basic_block_sptr block);
     void flatten(gr_simple_flowgraph_sptr sfg);
     gr_endpoint resolve_port(int port, bool is_input);
     gr_endpoint resolve_endpoint(const gr_endpoint &endp, bool is_input);
diff --git a/gnuradio-core/src/lib/runtime/gr_runtime.cc b/gnuradio-core/src/lib/runtime/gr_runtime.cc
index a8b932f21..3a992f689 100644
--- a/gnuradio-core/src/lib/runtime/gr_runtime.cc
+++ b/gnuradio-core/src/lib/runtime/gr_runtime.cc
@@ -26,11 +26,8 @@
 
 #include <gr_runtime.h>
 #include <gr_runtime_impl.h>
-#include <gr_local_sighandler.h>
 #include <iostream>
 
-static gr_runtime *s_runtime = 0;
-
 gr_runtime_sptr 
 gr_make_runtime(gr_hier_block2_sptr top_block)
 {
@@ -40,28 +37,16 @@ gr_make_runtime(gr_hier_block2_sptr top_block)
 gr_runtime::gr_runtime(gr_hier_block2_sptr top_block)
 {
   d_impl = new gr_runtime_impl(top_block);
-  s_runtime = this;
 }
   
 gr_runtime::~gr_runtime()
 {
-  s_runtime = 0; // we don't own this
   delete d_impl;
 }
 
-// FIXME: This prevents using more than one gr_runtime instance
-static void 
-runtime_sigint_handler(int signum)
-{
-  if (s_runtime)
-    s_runtime->stop();
-}
-
 void 
 gr_runtime::start()
 {
-  gr_local_sighandler sigint(SIGINT, runtime_sigint_handler);
-
   d_impl->start();
 }
 
@@ -74,24 +59,18 @@ gr_runtime::stop()
 void 
 gr_runtime::wait()
 {
-  gr_local_sighandler sigint(SIGINT, runtime_sigint_handler);
-
   d_impl->wait();
 }
 
 void 
 gr_runtime::run()
 {
-  gr_local_sighandler sigint(SIGINT, runtime_sigint_handler);
-
-  d_impl->start();
-  d_impl->wait();
+  start();
+  wait();
 }
 
 void
 gr_runtime::restart()
 {
-  gr_local_sighandler sigint(SIGINT, runtime_sigint_handler);
-
   d_impl->restart();
 }
diff --git a/gnuradio-core/src/lib/runtime/gr_runtime_impl.cc b/gnuradio-core/src/lib/runtime/gr_runtime_impl.cc
index 5cd75f3ce..64fbca9c9 100644
--- a/gnuradio-core/src/lib/runtime/gr_runtime_impl.cc
+++ b/gnuradio-core/src/lib/runtime/gr_runtime_impl.cc
@@ -1,6 +1,6 @@
 /* -*- c++ -*- */
 /*
- * Copyright 2006 Free Software Foundation, Inc.
+ * Copyright 2006,2007 Free Software Foundation, Inc.
  *
  * This file is part of GNU Radio
  *
@@ -24,10 +24,12 @@
 #include "config.h"
 #endif
 
+#include <gr_runtime.h>
 #include <gr_runtime_impl.h>
 #include <gr_simple_flowgraph.h>
 #include <gr_hier_block2.h>
 #include <gr_hier_block2_detail.h>
+#include <gr_local_sighandler.h>
 
 #ifdef HAVE_SIGNAL_H
 #include <signal.h>
@@ -36,24 +38,42 @@
 #include <stdexcept>
 #include <iostream>
 
+#define GR_RUNTIME_IMPL_DEBUG 1
+
+static gr_runtime_impl *s_runtime = 0;
+
+// FIXME: This prevents using more than one gr_runtime instance
+void 
+runtime_sigint_handler(int signum)
+{
+  if (GR_RUNTIME_IMPL_DEBUG)
+    std::cout << "SIGINT received, calling stop() on all threads" << std::endl;
+
+  if (s_runtime)
+    s_runtime->stop();
+}
+
 gr_runtime_impl::gr_runtime_impl(gr_hier_block2_sptr top_block) 
   : d_running(false),
     d_top_block(top_block),
     d_sfg(gr_make_simple_flowgraph())
 {
+  s_runtime = this;
 }
 
 gr_runtime_impl::~gr_runtime_impl()
 {
+  s_runtime = 0; // we don't own this
 }
 
 void
 gr_runtime_impl::start()
 {
+  if (GR_RUNTIME_IMPL_DEBUG)
+    std::cout << "start: entered" << std::endl;
+
   if (d_running)
     throw std::runtime_error("already running");
-  else
-    d_running = true;
 
   // Create new simple flow graph by flattening hierarchical block
   d_sfg->d_detail->reset();
@@ -70,24 +90,33 @@ gr_runtime_impl::start()
 void
 gr_runtime_impl::start_threads()
 {
+  if (GR_RUNTIME_IMPL_DEBUG)
+    std::cout << "start_threads: entered" << std::endl;
+
   d_graphs = d_sfg->d_detail->partition();
-  d_threads.clear();
   for (std::vector<gr_block_vector_t>::iterator p = d_graphs.begin();
        p != d_graphs.end(); p++) {
     gr_scheduler_thread *thread = new gr_scheduler_thread(*p);
-    thread->start();
     d_threads.push_back(thread);
+    if (GR_RUNTIME_IMPL_DEBUG)
+      std::cout << "start_threads: starting " << thread << std::endl;
+    thread->start();
   }
+
+  d_running = true;
 }
 
 void
 gr_runtime_impl::stop()
 {
-  if (!d_running)
-    throw std::runtime_error("not running");
+  if (GR_RUNTIME_IMPL_DEBUG)
+    std::cout << "stop: entered" << std::endl;
 
-  for (gr_scheduler_thread_viter_t p = d_threads.begin(); p != d_threads.end(); p++)
-    (*p)->stop(); 
+  for (gr_scheduler_thread_viter_t p = d_threads.begin(); p != d_threads.end(); p++) {
+    if (GR_RUNTIME_IMPL_DEBUG)
+      std::cout << "stop: stopping thread " << (*p) << std::endl;
+    (*p)->stop();
+  }
 
   d_running = false;
 }
@@ -95,30 +124,48 @@ gr_runtime_impl::stop()
 void
 gr_runtime_impl::wait()
 {
+  if (GR_RUNTIME_IMPL_DEBUG)
+    std::cout << "wait: entered" << std::endl;
+
   void *dummy_status; // don't ever dereference this
+  gr_local_sighandler sigint(SIGINT, runtime_sigint_handler);
 
   for (gr_scheduler_thread_viter_t p = d_threads.begin(); p != d_threads.end(); p++) {
+    if (GR_RUNTIME_IMPL_DEBUG)
+      std::cout << "wait: joining thread " << (*p) << std::endl;
     (*p)->join(&dummy_status); // pthreads will self-delete, so pointer is now dead
     (*p) = 0; // FIXME: switch to stl::list and actually remove from container
+    if (GR_RUNTIME_IMPL_DEBUG)
+      std::cout << "wait: join returned" << std::endl;
   }
+
+  d_threads.clear();
 }
 
 void
 gr_runtime_impl::restart()
 {
+  if (GR_RUNTIME_IMPL_DEBUG)
+    std::cout << "restart: entered" << std::endl;
+
   if (!d_running)
     throw std::runtime_error("not running");
 
   // Stop scheduler threads and wait for completion
   stop();
   wait();
-  
+  if (GR_RUNTIME_IMPL_DEBUG)
+    std::cout << "restart: threads stopped" << std::endl;
+
   // Create new simple flow graph 
   gr_simple_flowgraph_sptr new_sfg = gr_make_simple_flowgraph();
   d_top_block->d_detail->flatten(new_sfg);
   new_sfg->validate();
   new_sfg->d_detail->merge_connections(d_sfg);
-  // d_sfg = new_sfg;
+
+  if (GR_RUNTIME_IMPL_DEBUG)
+    std::cout << "restart: replacing old flow graph with new" << std::endl;
+  d_sfg = new_sfg;
 
   start_threads();
 }
@@ -161,3 +208,4 @@ gr_scheduler_thread::stop()
 {
   d_sts->stop();
 }
+
diff --git a/gnuradio-core/src/lib/runtime/gr_runtime_impl.h b/gnuradio-core/src/lib/runtime/gr_runtime_impl.h
index bb2a08334..0fb6e1f0d 100644
--- a/gnuradio-core/src/lib/runtime/gr_runtime_impl.h
+++ b/gnuradio-core/src/lib/runtime/gr_runtime_impl.h
@@ -67,6 +67,7 @@ class gr_runtime_impl
 {
 private:
   gr_runtime_impl(gr_hier_block2_sptr top_block);
+  friend void runtime_sigint_handler(int signum);
   friend class gr_runtime;
     
   bool                           d_running;
diff --git a/gnuradio-core/src/lib/runtime/gr_simple_flowgraph.h b/gnuradio-core/src/lib/runtime/gr_simple_flowgraph.h
index a22e558ff..04b83c8cc 100644
--- a/gnuradio-core/src/lib/runtime/gr_simple_flowgraph.h
+++ b/gnuradio-core/src/lib/runtime/gr_simple_flowgraph.h
@@ -62,6 +62,7 @@ class gr_simple_flowgraph
 {
 private:
   friend class gr_runtime_impl;
+  friend class gr_simple_flowgraph_detail;
   friend class gr_hier_block2_detail;
   friend gr_simple_flowgraph_sptr gr_make_simple_flowgraph();
   gr_simple_flowgraph();
diff --git a/gnuradio-core/src/lib/runtime/gr_simple_flowgraph_detail.cc b/gnuradio-core/src/lib/runtime/gr_simple_flowgraph_detail.cc
index 13d23efe5..a725adb6e 100644
--- a/gnuradio-core/src/lib/runtime/gr_simple_flowgraph_detail.cc
+++ b/gnuradio-core/src/lib/runtime/gr_simple_flowgraph_detail.cc
@@ -31,6 +31,9 @@
 #include <gr_buffer.h>
 #include <iostream>
 #include <stdexcept>
+#include <map>
+
+#define GR_SIMPLE_FLOWGRAPH_DETAIL_DEBUG 1
 
 gr_edge_sptr
 gr_make_edge(const gr_endpoint &src, const gr_endpoint &dst)
@@ -115,6 +118,9 @@ gr_simple_flowgraph_detail::validate()
     std::vector<int> used_ports;
     int ninputs, noutputs;
 
+    if (GR_SIMPLE_FLOWGRAPH_DETAIL_DEBUG)
+      std::cout << "Validating block: " << (*p) << std::endl;
+
     used_ports = calc_used_ports(*p, true); // inputs
     ninputs = used_ports.size();
     check_contiguity(*p, used_ports, true); // inputs
@@ -192,76 +198,6 @@ gr_simple_flowgraph_detail::check_contiguity(gr_basic_block_sptr block,
   }
 }
 
-void
-gr_simple_flowgraph_detail::setup_connections()
-{
-  // Assign block details to blocks
-  for (gr_basic_block_viter_t p = d_blocks.begin(); p != d_blocks.end(); p++) {
-    int ninputs = calc_used_ports(*p, true).size();
-    int noutputs = calc_used_ports(*p, false).size();
-    gr_block_detail_sptr detail = gr_make_block_detail(ninputs, noutputs);
-    for (int i = 0; i < noutputs; i++)
-      detail->set_output(i, allocate_buffer(*p, i));
-
-    boost::dynamic_pointer_cast<gr_block, gr_basic_block>(*p)->set_detail(detail);
-  }
-
-  // Connect inputs to outputs for each block
-  for(gr_basic_block_viter_t p = d_blocks.begin(); p != d_blocks.end(); p++) {
-    gr_block_sptr grblock(boost::dynamic_pointer_cast<gr_block, gr_basic_block>(*p));
-    if (!grblock)
-      throw std::runtime_error("setup_connections found non-gr_block");
-
-    // Get its detail and edges that feed into it
-    gr_block_detail_sptr detail = grblock->detail();
-    gr_edge_vector_t in_edges = calc_upstream_edges(*p);
-
-    // For each edge that feeds into it
-    for (gr_edge_viter_t e = in_edges.begin(); e != in_edges.end(); e++) {
-      // Set the input reader on the destination port to the output
-      // buffer on the source port
-      int dst_port = (*e)->dst().port();
-      int src_port = (*e)->src().port();
-      gr_basic_block_sptr src_block = (*e)->src().block();
-      gr_block_sptr src_grblock(boost::dynamic_pointer_cast<gr_block, gr_basic_block>(src_block));
-      if (!grblock)
-	throw std::runtime_error("setup_connections found non-gr_block");
-      gr_buffer_sptr src_buffer = src_grblock->detail()->output(src_port);
-
-      detail->set_input(dst_port, gr_buffer_add_reader(src_buffer, grblock->history()-1));
-    }
-  }
-}
-
-gr_buffer_sptr
-gr_simple_flowgraph_detail::allocate_buffer(gr_basic_block_sptr block, int port)
-{
-  gr_block_sptr grblock(boost::dynamic_pointer_cast<gr_block, gr_basic_block>(block));
-  if (!grblock)
-    throw std::runtime_error("allocate_buffer found non-gr_block");
-  int item_size = block->output_signature()->sizeof_stream_item(port);
-  int nitems = s_fixed_buffer_size/item_size;
-
-  // Make sure there are at least twice the output_multiple no. of items
-  if (nitems < 2*grblock->output_multiple())	// Note: this means output_multiple()
-    nitems = 2*grblock->output_multiple();	// can't be changed by block dynamically
-
-  // If any downstream blocks are decimators and/or have a large output_multiple,
-  // ensure we have a buffer at least twice their decimation factor*output_multiple
-  gr_basic_block_vector_t blocks = calc_downstream_blocks(block, port);
-  for (gr_basic_block_viter_t p = blocks.begin(); p != blocks.end(); p++) {
-    gr_block_sptr dgrblock(boost::dynamic_pointer_cast<gr_block, gr_basic_block>(*p));
-      if (!dgrblock)
-	throw std::runtime_error("allocate_buffer found non-gr_block");
-    int decimation = (int)(1.0/dgrblock->relative_rate());
-    int multiple   = dgrblock->output_multiple();
-    int history    = dgrblock->history();
-    nitems = std::max(nitems, 2*(decimation*multiple+history));
-  }
-
-  return gr_make_buffer(nitems, item_size);
-}
-
 gr_basic_block_vector_t
 gr_simple_flowgraph_detail::calc_downstream_blocks(gr_basic_block_sptr block, int port)
 {
@@ -476,8 +412,198 @@ gr_simple_flowgraph_detail::topological_dfs_visit(gr_basic_block_sptr block, gr_
   output.push_back(result_block);
 }
 
+bool
+gr_simple_flowgraph_detail::has_block_p(gr_basic_block_sptr block)
+{
+  gr_basic_block_viter_t result;
+  result = std::find(d_blocks.begin(), d_blocks.end(), block);
+  return (result != d_blocks.end());
+}
+
+gr_edge_sptr
+gr_simple_flowgraph_detail::calc_upstream_edge(gr_basic_block_sptr block, int port)
+{
+  gr_edge_sptr result;
+
+  for (gr_edge_viter_t p = d_edges.begin(); p != d_edges.end(); p++) {
+    if ((*p)->dst().block() == block && (*p)->dst().port() == port) {
+      result = (*p);
+      break;
+    }
+  }
+
+  return result;
+}
+
+gr_block_detail_sptr
+gr_simple_flowgraph_detail::allocate_block_detail(gr_basic_block_sptr block, gr_block_detail_sptr old_detail)
+{
+  int ninputs = calc_used_ports(block, true).size();
+  int noutputs = calc_used_ports(block, false).size();
+  gr_block_detail_sptr detail = gr_make_block_detail(ninputs, noutputs);
+
+  if (GR_SIMPLE_FLOWGRAPH_DETAIL_DEBUG)
+    std::cout << "Creating block detail for " << block << std::endl;
+
+  // Re-use or allocate output buffers
+  for (int i = 0; i < noutputs; i++) {
+    gr_buffer_sptr buffer;
+
+    if (!old_detail || i >= old_detail->noutputs()) {
+      if (GR_SIMPLE_FLOWGRAPH_DETAIL_DEBUG)
+	std::cout << "Allocating new buffer for output " << i << std::endl;
+      buffer = allocate_buffer(block, i);
+    }
+    else {
+      if (GR_SIMPLE_FLOWGRAPH_DETAIL_DEBUG)
+	std::cout << "Reusing old buffer for output " << i << std::endl;
+      buffer = old_detail->output(i);
+    }
+
+    detail->set_output(i, buffer);
+  }
+
+  return detail;
+}
+
 void
-gr_simple_flowgraph_detail::merge_connections(gr_simple_flowgraph_sptr sfg)
+gr_simple_flowgraph_detail::connect_block_inputs(gr_basic_block_sptr block)
+{
+  gr_block_sptr grblock(boost::dynamic_pointer_cast<gr_block, gr_basic_block>(block));
+  if (!grblock)
+    throw std::runtime_error("found non-gr_block");
+  
+  // Get its detail and edges that feed into it
+  gr_block_detail_sptr detail = grblock->detail();
+  gr_edge_vector_t in_edges = calc_upstream_edges(block);
+  
+  // For each edge that feeds into it
+  for (gr_edge_viter_t e = in_edges.begin(); e != in_edges.end(); e++) {
+    // Set the buffer reader on the destination port to the output
+    // buffer on the source port
+    int dst_port = (*e)->dst().port();
+    int src_port = (*e)->src().port();
+    gr_basic_block_sptr src_block = (*e)->src().block();
+    gr_block_sptr src_grblock(boost::dynamic_pointer_cast<gr_block, gr_basic_block>(src_block));
+    if (!grblock)
+      throw std::runtime_error("found non-gr_block");
+    gr_buffer_sptr src_buffer = src_grblock->detail()->output(src_port);
+    
+    if (GR_SIMPLE_FLOWGRAPH_DETAIL_DEBUG)
+      std::cout << "Setting input " << dst_port << " from edge " << (*e) << std::endl;
+
+    detail->set_input(dst_port, gr_buffer_add_reader(src_buffer, grblock->history()-1));
+  }
+}
+
+gr_buffer_sptr
+gr_simple_flowgraph_detail::allocate_buffer(gr_basic_block_sptr block, int port)
 {
-    // NOT IMPLEMENTED
+  gr_block_sptr grblock(boost::dynamic_pointer_cast<gr_block, gr_basic_block>(block));
+  if (!grblock)
+    throw std::runtime_error("allocate_buffer found non-gr_block");
+  int item_size = block->output_signature()->sizeof_stream_item(port);
+  int nitems = s_fixed_buffer_size/item_size;
+
+  // Make sure there are at least twice the output_multiple no. of items
+  if (nitems < 2*grblock->output_multiple())	// Note: this means output_multiple()
+    nitems = 2*grblock->output_multiple();	// can't be changed by block dynamically
+
+  // If any downstream blocks are decimators and/or have a large output_multiple,
+  // ensure we have a buffer at least twice their decimation factor*output_multiple
+  gr_basic_block_vector_t blocks = calc_downstream_blocks(block, port);
+  for (gr_basic_block_viter_t p = blocks.begin(); p != blocks.end(); p++) {
+    gr_block_sptr dgrblock(boost::dynamic_pointer_cast<gr_block, gr_basic_block>(*p));
+      if (!dgrblock)
+	throw std::runtime_error("allocate_buffer found non-gr_block");
+    int decimation = (int)(1.0/dgrblock->relative_rate());
+    int multiple   = dgrblock->output_multiple();
+    int history    = dgrblock->history();
+    nitems = std::max(nitems, 2*(decimation*multiple+history));
+  }
+
+  return gr_make_buffer(nitems, item_size);
+}
+
+void
+gr_simple_flowgraph_detail::setup_connections()
+{
+  // Assign block details to blocks
+  for (gr_basic_block_viter_t p = d_blocks.begin(); p != d_blocks.end(); p++)
+    boost::dynamic_pointer_cast<gr_block, gr_basic_block>(*p)->set_detail(allocate_block_detail(*p));
+
+  // Connect inputs to outputs for each block
+  for(gr_basic_block_viter_t p = d_blocks.begin(); p != d_blocks.end(); p++)
+    connect_block_inputs(*p);
+}
+
+void
+gr_simple_flowgraph_detail::merge_connections(gr_simple_flowgraph_sptr old_sfg)
+{
+  std::map<gr_block_sptr, gr_block_detail_sptr> old_details;
+
+  // Allocate or reuse output buffers
+  for (gr_basic_block_viter_t p = d_blocks.begin(); p != d_blocks.end(); p++) {
+    gr_block_sptr block(boost::dynamic_pointer_cast<gr_block, gr_basic_block>(*p));
+
+    gr_block_detail_sptr old_detail = block->detail();
+    block->set_detail(allocate_block_detail(block, old_detail));
+
+    // Save old detail for use in next step
+    old_details[block] = old_detail;
+  }
+
+  for (gr_basic_block_viter_t p = d_blocks.begin(); p != d_blocks.end(); p++) {
+    gr_block_sptr block(boost::dynamic_pointer_cast<gr_block, gr_basic_block>(*p));
+
+    if (GR_SIMPLE_FLOWGRAPH_DETAIL_DEBUG)
+      std::cout << "merge: testing " << (*p) << "...";
+    
+    if (old_sfg->d_detail->has_block_p(*p)) {
+      // Block exists in old flow graph
+      if (GR_SIMPLE_FLOWGRAPH_DETAIL_DEBUG)
+	std::cout << "used in old flow graph" << std::endl;
+      gr_block_detail_sptr detail = block->detail();
+
+      // Iterate through the inputs and see what needs to be done
+      for (int i = 0; i < detail->ninputs(); i++) {
+	if (GR_SIMPLE_FLOWGRAPH_DETAIL_DEBUG)
+	  std::cout << "Checking input " << i << "...";
+
+	gr_edge_sptr edge = calc_upstream_edge(*p, i);
+	if (!edge)
+	  throw std::runtime_error("merge: missing input edge");
+
+	// Fish out old buffer reader and see if it matches correct buffer from edge list
+	gr_block_sptr src_block(boost::dynamic_pointer_cast<gr_block, gr_basic_block>(edge->src().block()));
+	gr_block_detail_sptr src_detail = src_block->detail();
+	gr_buffer_sptr src_buffer = src_detail->output(edge->src().port());
+	gr_buffer_reader_sptr old_reader;
+	gr_block_detail_sptr old_detail = old_details[block];
+	if (old_detail && i < old_detail->ninputs())
+	  old_reader = old_detail->input(i);
+	
+	// If there's a match, use it
+	if (old_reader && (src_buffer == old_reader->buffer())) {
+	  if (GR_SIMPLE_FLOWGRAPH_DETAIL_DEBUG)
+	    std::cout << "matched" << std::endl;
+	  detail->set_input(i, old_reader);
+
+	}
+	else {
+	  if (GR_SIMPLE_FLOWGRAPH_DETAIL_DEBUG)
+	    std::cout << "needs a new reader" << std::endl;
+
+	  // Create new buffer reader and assign
+	  detail->set_input(i, gr_buffer_add_reader(src_buffer, block->history()-1));
+	}
+      }
+    }
+    else {
+      // Block is new, it just needs buffer readers at this point
+      if (GR_SIMPLE_FLOWGRAPH_DETAIL_DEBUG)
+	std::cout << "new block" << std::endl;
+      connect_block_inputs(block);
+    }
+  }  
 }
diff --git a/gnuradio-core/src/lib/runtime/gr_simple_flowgraph_detail.h b/gnuradio-core/src/lib/runtime/gr_simple_flowgraph_detail.h
index 6e9058bba..5d01c38f0 100644
--- a/gnuradio-core/src/lib/runtime/gr_simple_flowgraph_detail.h
+++ b/gnuradio-core/src/lib/runtime/gr_simple_flowgraph_detail.h
@@ -79,9 +79,13 @@ private:
   void setup_connections();
   void merge_connections(gr_simple_flowgraph_sptr sfg);
 
+  void connect_block_inputs(gr_basic_block_sptr block);
+  gr_block_detail_sptr allocate_block_detail(gr_basic_block_sptr block, 
+					     gr_block_detail_sptr old_detail=gr_block_detail_sptr());
   gr_buffer_sptr allocate_buffer(gr_basic_block_sptr block, int port);
   gr_basic_block_vector_t calc_downstream_blocks(gr_basic_block_sptr block, int port);
   gr_basic_block_vector_t calc_downstream_blocks(gr_basic_block_sptr block);
+  gr_edge_sptr calc_upstream_edge(gr_basic_block_sptr block, int port);
   gr_edge_vector_t calc_upstream_edges(gr_basic_block_sptr block);
   gr_basic_block_vector_t calc_used_blocks();
   std::vector<gr_block_vector_t> partition();
@@ -92,7 +96,8 @@ private:
   bool source_p(gr_basic_block_sptr block);
   gr_basic_block_vector_t sort_sources_first(gr_basic_block_vector_t &blocks);
   void topological_dfs_visit(gr_basic_block_sptr block, gr_block_vector_t &output);
-        
+  bool has_block_p(gr_basic_block_sptr block);
+
 public:
   ~gr_simple_flowgraph_detail();
 };
@@ -100,7 +105,7 @@ public:
 inline std::ostream&
 operator <<(std::ostream &os, const gr_endpoint endp)
 {
-  os << endp.block()->name() << ":" << endp.port();
+  os << endp.block() << ":" << endp.port();
   return os;
 }
 
diff --git a/gnuradio-core/src/python/gnuradio/gr/qa_hier_block2.py b/gnuradio-core/src/python/gnuradio/gr/qa_hier_block2.py
index 98691cfb5..1e12a5d73 100755
--- a/gnuradio-core/src/python/gnuradio/gr/qa_hier_block2.py
+++ b/gnuradio-core/src/python/gnuradio/gr/qa_hier_block2.py
@@ -158,17 +158,87 @@ class test_hier_block2(gr_unittest.TestCase):
         self.assertEquals(expected, actual1)
         self.assertEquals(expected, actual2)
 
-    def test_015_connect_disconnect(self):
-        expected = (1.0, 2.0, 3.0, 4.0)
-        hblock = gr.top_block("test_block")
-        src = gr.vector_source_f(expected, False)
-        sink1 = gr.vector_sink_f()
-        sink2 = gr.vector_sink_f()
-        hblock.connect(src, sink1)
-        hblock.connect(src, sink2)
-        hblock.disconnect(src, sink2)
-        runtime = gr.runtime(hblock)
-        runtime.run()
-    
+    def test_015_disconnect_input(self):
+	hblock = gr.hier_block2("test_block", 
+				gr.io_signature(1,1,gr.sizeof_int), 
+				gr.io_signature(1,1,gr.sizeof_int))
+	nop1 = gr.nop(gr.sizeof_int)
+	hblock.connect(hblock, nop1)
+        hblock.disconnect(hblock, nop1)
+   
+    def test_016_disconnect_input_not_connected(self):
+	hblock = gr.hier_block2("test_block", 
+				gr.io_signature(1,1,gr.sizeof_int), 
+				gr.io_signature(1,1,gr.sizeof_int))
+	nop1 = gr.nop(gr.sizeof_int)
+        nop2 = gr.nop(gr.sizeof_int)
+	hblock.connect(hblock, nop1)
+        self.assertRaises(ValueError,
+            lambda: hblock.disconnect(hblock, nop2))
+   
+    def test_017_disconnect_input_neg(self):
+	hblock = gr.hier_block2("test_block", 
+				gr.io_signature(1,1,gr.sizeof_int), 
+				gr.io_signature(1,1,gr.sizeof_int))
+	nop1 = gr.nop(gr.sizeof_int)
+	hblock.connect(hblock, nop1)
+        self.assertRaises(ValueError,
+            lambda: hblock.disconnect((hblock, -1), nop1))
+
+    def test_018_disconnect_input_exceeds(self):
+	hblock = gr.hier_block2("test_block", 
+				gr.io_signature(1,1,gr.sizeof_int), 
+				gr.io_signature(1,1,gr.sizeof_int))
+	nop1 = gr.nop(gr.sizeof_int)
+	hblock.connect(hblock, nop1)
+        self.assertRaises(ValueError,
+            lambda: hblock.disconnect((hblock, 1), nop1))
+
+    def test_019_disconnect_output(self):
+	hblock = gr.hier_block2("test_block", 
+				gr.io_signature(1,1,gr.sizeof_int), 
+				gr.io_signature(1,1,gr.sizeof_int))
+	nop1 = gr.nop(gr.sizeof_int)
+	hblock.connect(nop1, hblock)
+        hblock.disconnect(nop1, hblock)
+   
+    def test_020_disconnect_output_not_connected(self):
+	hblock = gr.hier_block2("test_block", 
+				gr.io_signature(1,1,gr.sizeof_int), 
+				gr.io_signature(1,1,gr.sizeof_int))
+	nop1 = gr.nop(gr.sizeof_int)
+        nop2 = gr.nop(gr.sizeof_int)
+	hblock.connect(nop1, hblock)
+        self.assertRaises(ValueError,
+            lambda: hblock.disconnect(nop2, hblock))
+   
+    def test_021_disconnect_output_neg(self):
+	hblock = gr.hier_block2("test_block", 
+				gr.io_signature(1,1,gr.sizeof_int), 
+				gr.io_signature(1,1,gr.sizeof_int))
+	nop1 = gr.nop(gr.sizeof_int)
+	hblock.connect(hblock, nop1)
+        self.assertRaises(ValueError,
+            lambda: hblock.disconnect(nop1, (hblock, -1)))
+
+    def test_022_disconnect_output_exceeds(self):
+	hblock = gr.hier_block2("test_block", 
+				gr.io_signature(1,1,gr.sizeof_int), 
+				gr.io_signature(1,1,gr.sizeof_int))
+	nop1 = gr.nop(gr.sizeof_int)
+	hblock.connect(nop1, hblock)
+        self.assertRaises(ValueError,
+            lambda: hblock.disconnect(nop1, (hblock, 1)))
+
+    def test_023_run(self):
+	hblock = gr.top_block("test_block")
+	data = (1.0, 2.0, 3.0, 4.0)
+	src = gr.vector_source_f(data, False)
+	dst = gr.vector_sink_f()
+	hblock.connect(src, dst)
+	r = gr.runtime(hblock)
+	r.run()
+	self.assertEquals(data, dst.data())
+
 if __name__ == "__main__":
     gr_unittest.main()