Start(), Stop(), run(), wait() in GNURADIO : top_block

Either you can use tb.start() + tb.wait() OR you can use tb.run() OR you can use tb.start() + tb.wait() + tb.stop()

If you have to do collect finite amount of data then you need to make a gr.head()

Generally the run method doesn't have stop.. you can see it in uhd_rx_cfile
In order to stop in such case, a gr.head() is created with finite number of samples so that the flow graph stops by itself.

All these start(), stop(), run(), wait() are boost library threads. Need to find out more by reading there.

run() = start() + wait() i.e. a call to tb.run() calls both tb.start() and tb.wait()

If you use run(), the flowgraph will not only stop, but will end its lifetime. Once run() has returned, the flowgraph is no longer usable, or as we like to say, further operations on it are "undefined." 

The run() method on a top block is really just a convenient way of telling GNU Radio your application has nothing else to do until the flowgraph exits. 


So if you want the flow graph to continue use start() and then wait()and after that stop() to stop the flow graph

Details on all of the methods can be found in gr_top_block.h as well as gr

Here is the link to important discussions :

http://gnuradio.4.n7.nabble.com/Reg-start-method-in-gr-top-block-h-tt37915.html
http://gnuradio.4.n7.nabble.com/Reg-start-wait-stop-and-run-tt37916.html

Sample example :

Try-1 :

def main():
                t = my_top_block()

                t.start()
                t.wait()
                t.stop()
                print t.c2mag.level() #print power
                time.sleep(3)
                       
                m = my_top_block()
               
                m.start()
                m.wait()
                m.stop()
                print m.c2mag.level() #print power
                time.sleep(3)
                       
                n = my_top_block()
               
                n.start()
                n.wait()
                n.stop()
                print n.c2mag.level() #print power
                time.sleep(3)

                r = my_top_block()
               
                r.start()
                r.wait()
                r.stop()
                print r.c2mag.level() #print power
                time.sleep(3)

Gnuradio scheduler : Part-1 How the GNU Radio scheduler is called and what it does

Today I read an interesting and informative pdf about how does the core of gnuradio works. Its based on gnuradio version 3.3.0

* Please donwload gnuradio 3.3.0 from here
 http://gnuradio.org/releases/gnuradio/gnuradio-3.3.0.tar.gz

The author is Zhuo Lu

Lets take an example for dial_tone.py


dial_tone.py
~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~

from gnuradio import gr
from gnuradio import audio
from gnuradio.eng_option import eng_option
from optparse import OptionParser

class my_top_block(gr.top_block):

    def __init__(self):
        gr.top_block.__init__(self)

        parser = OptionParser(option_class=eng_option)
        parser.add_option("-O", "--audio-output", type="string", default="",
                          help="pcm output device name.  E.g., hw:0,0 or /dev/dsp")
        parser.add_option("-r", "--sample-rate", type="eng_float", default=48000,
                          help="set sample rate to RATE (48000)")
        (options, args) = parser.parse_args ()
        if len(args) != 0:
            parser.print_help()
            raise SystemExit, 1

        sample_rate = int(options.sample_rate)
        ampl = 0.1

        src0 = gr.sig_source_f (sample_rate, gr.GR_SIN_WAVE, 350, ampl)
        src1 = gr.sig_source_f (sample_rate, gr.GR_SIN_WAVE, 440, ampl)
        dst = audio.sink (sample_rate, options.audio_output)
        self.connect (src0, (dst, 0))
        self.connect (src1, (dst, 1))


if __name__ == '__main__':
    try:
        my_top_block().run()
    except KeyboardInterrupt:
        pass
~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~

Besides some running connection for the blocks, gnuradio running threads start at my_top_block.run() only.

Lets see what happens afterwards

** Python and C++ classes have one to one correspondence  via SWIG.

run() is defined in C++ class gr_top_block. To find this go to

/gnuradio-3.3.0/gnuradio-core/src/lib/runtime

There you will find gr_top_block.cc

There you can find the definition for run() as follows :

void
gr_top_block::run()
{
  start();
  wait();
}

Also the definition of start() is

void
gr_top_block::start()
{
  d_impl->start();
}



d_impl is a member that points to the class gr_top_block_impl. This class can be found in  /gnuradio-3.3.0/gnuradio-core/src/lib/runtime/gr_top_block_impl.cc

* -> is a structure dereference operator. a->b means member b of the object    pointed to by a
*  while . is a structure reference operator. a.b means member b of object a


So here it means start() method as defined in the object pointed by d_impl i.e. gr_top_block_impl.cc (see line 45 of gr_top_block.cc)


Lets open the file gr_top_block_impl.cc and see the definition of start()

void
gr_top_block_impl::start()
{
  gruel::scoped_lock    l(d_mutex);

  if (d_state != IDLE)
    throw std::runtime_error("top_block::start: top block already running or  wait() not called after previous stop()");

  if (d_lock_count > 0)
    throw std::runtime_error("top_block::start: can't start with flow graph locked");

  // Create new flat flow graph by flattening hierarchy
  d_ffg = d_owner->flatten();

  // Validate new simple flow graph and wire it up
  d_ffg->validate();
  d_ffg->setup_connections();

  d_scheduler = make_scheduler(d_ffg);
  d_state = RUNNING;
}

The codes do some sanity check and then create the GNU Radio Scheduler by calling

d scheduler = make_scheduler ( d_ffg ) ;





Now lets go to the definition of make_scheduler(), it is also defined in gr_top_block_impl.cc as follows

static gr_scheduler_sptr make_scheduler(gr_flat_flowgraph_sptr ffg)
{
  static scheduler_maker  factory = 0;

  if (factory == 0){
    char *v = getenv("GR_SCHEDULER");
    if (!v)
      factory = scheduler_table[0].f;    // use default
    else {
      for (size_t i = 0; i < sizeof(scheduler_table)/sizeof(scheduler_table[0]); i++){
    if (strcmp(v, scheduler_table[i].name) == 0){
      factory = scheduler_table[i].f;
      break;
    }
      }
      if (factory == 0){
    std::cerr << "warning: Invalid GR_SCHEDULER environment variable value \""
          << v << "\".  Using \"" << scheduler_table[0].name << "\"\n";
    factory = scheduler_table[0].f;
      }
    }
  }
  return factory(ffg);
}

In the above definition we have a strange variable

"static scheduler_maker  factory" which has been initialized to 0

* A variable declared static in a function retains its state between calls to that function.

Its definition is given in the very beginning of the file as

 typedef gr_scheduler_sptr (*scheduler_maker)(gr_flat_flowgraph_sptr ffg)

So it seems that factory is a function pointer and where does it point to ?

Well its written that

factory = scheduler_table[i].f

Lets see what is inside scheduler_table. Its actually an array.   See in the begining of the file gr_top_block_impl.cc

static struct scheduler_table {
  const char            *name;
  scheduler_maker    f;
} scheduler_table[] = {
  { "TPB",    gr_scheduler_tpb::make },    // first entry is default
  { "STS",    gr_scheduler_sts::make }
};
 


It seems that it points to the member function "make" in the scheduler's class.

So it checks whether there exists a Linux environment variable:
GR_SCHEDULER.

If no, use the default scheduler(TPB); otherwise, use user’s choice. And we do not have so many choices on the scheduler:



1. TPB (default): multi-threaded scheduler.
2. STS: single-threaded scheduler.

So by default, gr_scheduler_tpb::make will be called.





Now lets see what is inside gr_scheduler_tpb.cc You can find it in
/gnuradio-3.3.0/gnuradio-core/src/lib/runtime/gr_scheduler_tpb.cc

In the following 2 lines the constructor of the gr_scheduler_tpb is called

gr_scheduler_tpb::make(gr_flat_flowgraph_sptr ffg)
{
  return gr_scheduler_sptr(new gr_scheduler_tpb(ffg));
}

The next few line are

gr_scheduler_tpb::gr_scheduler_tpb(gr_flat_flowgraph_sptr ffg)
  : gr_scheduler(ffg)
{
  // Get a topologically sorted vector of all the blocks in use.
  // Being topologically sorted probably isn't going to matter, but
  // there's a non-zero chance it might help...

  gr_basic_block_vector_t used_blocks = ffg->calc_used_blocks();
  used_blocks = ffg->topological_sort(used_blocks);
  gr_block_vector_t blocks = gr_flat_flowgraph::make_block_vector(used_blocks);

  // Ensure that the done flag is clear on all blocks

  for (size_t i = 0; i < blocks.size(); i++){
    blocks[i]->detail()->set_done(false);
  }

  // Fire off a thead for each block

  for (size_t i = 0; i < blocks.size(); i++){
    std::stringstream name;
    name << "thread-per-block[" << i << "]: " << blocks[i];
    d_threads.create_thread(
      gruel::thread_body_wrapper(tpb_container(blocks[i]), name.str()));
  }
}

Last two lines are important here

d_threads.create_thread(
      gruel::thread_body_wrapper(tpb_container(blocks[i]), name.str()));


thread_body_wrapper wraps the main thread with the block name. Then, the thread begins from thread_body_wrapper().


Lets see inside thread_body_wrapper.h
you can locate it in gnuradio-3.3.0/gruel/src/include/gruel and the .cc file is in
gnuradio-3.3.0/gruel/src/lib


  template
  class thread_body_wrapper
  {
    F         d_f;
    std::string d_name;

  public:

    explicit thread_body_wrapper(F f, const std::string &name="")
      : d_f(f), d_name(name) {}

    void operator()()
    {
      mask_signals();

      try {
    d_f();
      }

So operator() has been overloaded here and d_f() is called actually and it is explicitly linked to tpb_container class . You can see this in the gr_scheduler_tpb.cc file in the definition of the class tpb_container

Lets see the code of tpb_container class :

class tpb_container
{
  gr_block_sptr    d_block;
 
public:
  tpb_container(gr_block_sptr block) : d_block(block) {}

  void operator()()
  {
    gr_tpb_thread_body    body(d_block);
  }
};

So the overloading of operate() just constructs another class "gr_tpb_thread_body"


From here the schedulers work is done.

So in brief the gnuradio scheduler does the following :

1. Analyze used blocks in gr_top_block
2. Default scheduler is TPB which creates multi-threads for blocks
3. The scheduler creates one concurrent thread for each block
4. For each block the thread's entry is gr_tpb_thread_body body(d_block)

Gnuradio scheduler : Part-2 How a thread of each block works

The TPB scheduler (the default one) generates a thread for each block whose entry starts from the constructor of class gr_tpb_thread_body

Lets see the constructor in the file gr_tpb_thread_body.cc
You can locate it in

gnuradio-3.3.0/gnuradio-core/src/lib/runtime/gr_tpb_thread_body.cc 

 gr_tpb_thread_body::gr_tpb_thread_body(gr_block_sptr block)
  : d_exec(block)
{
  // std::cerr << "gr_tpb_thread_body: " << block << std::endl;

  gr_block_detail *d = block->detail().get();
  gr_block_executor::state s;
  pmt_t msg;

main loop of the thread starts from here

while(1)

First the thread processes all signals

boost::this_thread::interruption_point();

    // handle any queued up messages
    while ((msg = d->d_tpb.delete_head_nowait()))
      block->handle_msg(msg);

    d->d_tpb.clear_changed();
    s = d_exec.run_one_iteration();

This run_one_iteration() is the main function call of the block and it finishes the main functionality of the block in

s = d_exec.run_one_iteration();

s is the return result of the run_one_iteration() 

The next thing is to just act on the result of different outcomes according to switch case :

    switch(s){
    case gr_block_executor::READY:        // Tell neighbors we made progress.
      d->d_tpb.notify_neighbors(d);
      break;

    case gr_block_executor::READY_NO_OUTPUT:    // Notify upstream only
      d->d_tpb.notify_upstream(d);
      break;

    case gr_block_executor::DONE:        // Game over.
      d->d_tpb.notify_neighbors(d);
      return;

    case gr_block_executor::BLKD_IN:        // Wait for input.
      {
    gruel::scoped_lock guard(d->d_tpb.mutex);
    while (!d->d_tpb.input_changed){
     
      // wait for input or message
      while(!d->d_tpb.input_changed && d->d_tpb.empty_p())
        d->d_tpb.input_cond.wait(guard);

      // handle all pending messages
      while ((msg = d->d_tpb.delete_head_nowait_already_holding_mutex())){
        guard.unlock();            // release lock while processing msg
        block->handle_msg(msg);
        guard.lock();
      }
    }
      }
      break;

     
    case gr_block_executor::BLKD_OUT:        // Wait for output buffer space.
      {
    gruel::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 && d->d_tpb.empty_p())
        d->d_tpb.output_cond.wait(guard);

      // handle all pending messages
      while ((msg = d->d_tpb.delete_head_nowait_already_holding_mutex())){
        guard.unlock();            // release lock while processing msg
        block->handle_msg(msg);
        guard.lock();
      }
    }
      }
      break;

    default:
      assert(0);
    }


So we can see that run_one_iteration() is the key in the whole thread and it includes the major functionality of the block. Lets see its code
gr_block_executor.cc

you can locate it in

gnuradio-3.3.0/gnuradio-core/src/lib/runtime/gr_block_executor.cc

The code is kind of two long but overall it does following :

1. Whether there exist sufficient data for output. NO -> return BLKD_OUT

 if (noutput_items == 0){        // we're output blocked
      LOG(*d_log << "  BLKD_OUT\n");
      return BLKD_OUT;

2. Whether there are sufficient data available.  No -> return BLKD_IN

    int i;
    for (i = 0; i < d->ninputs (); i++)
      if (d_ninput_items_required[i] > d_ninput_items[i])    // not enough
    break;

    if (i < d->ninputs ()){            // not enough input on input[i]
      // if we can, try reducing the size of our output request
      if (noutput_items > m->output_multiple ()){
    noutput_items /= 2;
    noutput_items = round_up (noutput_items, m->output_multiple ());
    goto try_again;
      }

      // We're blocked on input
      LOG(*d_log << "  BLKD_IN\n");
      if (d_input_done[i])     // If the upstream block is done, we're done
    goto were_done;

      // Is it possible to ever fulfill this request?
      if (d_ninput_items_required[i] > d->input(i)->max_possible_items_available ()){
    // Nope, never going to happen...
    std::cerr << "\nsched: name()
          << " (" << m->unique_id() << ")>"
          << " is requesting more input data\n"
          << "  than we can provide.\n"
          << "  ninput_items_required = "
          << d_ninput_items_required[i] << "\n"
          << "  max_possible_items_available = "
          << d->input(i)->max_possible_items_available() << "\n"
          << "  If this is a filter, consider reducing the number of taps.\n";
    goto were_done;
      }

      return BLKD_IN;
    }

3. If there are sufficient input data and sufficient output space, the ocde runs the actual work of the block i.e. general_work()

    // We've got enough data on each input to produce noutput_items.
    // Finish setting up the call to work.

    for (int i = 0; i < d->ninputs (); i++)
      d_input_items[i] = d->input(i)->read_pointer();

  setup_call_to_work:

    d->d_produce_or = 0;
    for (int i = 0; i < d->noutputs (); i++)
      d_output_items[i] = d->output(i)->write_pointer();

    // Do the actual work of the block
    int n = m->general_work (noutput_items, d_ninput_items,
                 d_input_items, d_output_items);
    LOG(*d_log << "  general_work: noutput_items = " << noutput_items
    << " result = " << n << std::endl);

    if (n == gr_block::WORK_DONE)
      goto were_done;

    if (n != gr_block::WORK_CALLED_PRODUCE)
      d->produce_each (n);    // advance write pointers
   
    if (d->d_produce_or > 0)    // block produced something
      return READY;

    // We didn't produce any output even though we called general_work.
    // We have (most likely) consumed some input.

    // If this is a source, it's broken.
    if (d->source_p()){
      std::cerr << "gr_block_executor: source " << m
        << " produced no output.  We're marking it DONE.\n";
      // FIXME maybe we ought to raise an exception...
      goto were_done;
    }


So briefly summarize how each thread in gnuradio core works :

1. A thread for each block has a while(1) loop

2. The loop processes signals and run the key function run_one_iteration()

3. run_one_iteration() checks if there are sufficient data at the input and sufficient space for the output of the block

4. If yes, the general_work() is called to run the main functionality of the block.

5. If no, return BLKD_OUT, BLKD_IN or others