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