git.gag.com Git - debian/gnuradio/blob - gnuradio-core/src/lib/runtime/gr_tpb_thread_body.cc

   1 /* -*- c++ -*- */
   2 /*
   3  * Copyright 2008,2009 Free Software Foundation, Inc.
   4  *
   5  * This file is part of GNU Radio
   6  *
   7  * GNU Radio is free software; you can redistribute it and/or modify
   8  * it under the terms of the GNU General Public License as published by
   9  * the Free Software Foundation; either version 3, or (at your option)
  10  * any later version.
  11  *
  12  * GNU Radio is distributed in the hope that it will be useful,
  13  * but WITHOUT ANY WARRANTY; without even the implied warranty of
  14  * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
  15  * GNU General Public License for more details.
  16  *
  17  * You should have received a copy of the GNU General Public License along
  18  * with this program; if not, write to the Free Software Foundation, Inc.,
  19  * 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301 USA.
  20  */
  21 #ifdef HAVE_CONFIG_H
  22 #include <config.h>
  23 #endif
  24 #include <gr_tpb_thread_body.h>
  25 #include <iostream>
  26 #include <boost/thread.hpp>
  27 #include <gruel/pmt.h>
  28
  29 using namespace pmt;
  30
  31 gr_tpb_thread_body::gr_tpb_thread_body(gr_block_sptr block)
  32   : d_exec(block)
  33 {
  34   // std::cerr << "gr_tpb_thread_body: " << block << std::endl;
  35
  36   gr_block_detail *d = block->detail().get();
  37   gr_block_executor::state s;
  38   pmt_t msg;
  39
  40
  41   while (1){
  42     boost::this_thread::interruption_point();
  43
  44     // handle any queued up messages
  45     while ((msg = d->d_tpb.delete_head_nowait()))
  46       block->handle_msg(msg);
  47
  48     d->d_tpb.clear_changed();
  49     s = d_exec.run_one_iteration();
  50
  51     switch(s){
  52     case gr_block_executor::READY:              // Tell neighbors we made progress.
  53       d->d_tpb.notify_neighbors(d);
  54       break;
  55
  56     case gr_block_executor::READY_NO_OUTPUT:    // Notify upstream only
  57       d->d_tpb.notify_upstream(d);
  58       break;
  59
  60     case gr_block_executor::DONE:               // Game over.
  61       d->d_tpb.notify_neighbors(d);
  62       return;
  63
  64     case gr_block_executor::BLKD_IN:            // Wait for input.
  65       {
  66         gruel::scoped_lock guard(d->d_tpb.mutex);
  67         while (!d->d_tpb.input_changed){
  68
  69           // wait for input or message
  70           while(!d->d_tpb.input_changed && d->d_tpb.empty_p())
  71             d->d_tpb.input_cond.wait(guard);
  72
  73           // handle all pending messages
  74           while ((msg = d->d_tpb.delete_head_nowait_already_holding_mutex())){
  75             guard.unlock();                     // release lock while processing msg
  76             block->handle_msg(msg);
  77             guard.lock();
  78           }
  79         }
  80       }
  81       break;
  82
  83
  84     case gr_block_executor::BLKD_OUT:           // Wait for output buffer space.
  85       {
  86         gruel::scoped_lock guard(d->d_tpb.mutex);
  87         while (!d->d_tpb.output_changed){
  88
  89           // wait for output room or message
  90           while(!d->d_tpb.output_changed && d->d_tpb.empty_p())
  91             d->d_tpb.output_cond.wait(guard);
  92
  93           // handle all pending messages
  94           while ((msg = d->d_tpb.delete_head_nowait_already_holding_mutex())){
  95             guard.unlock();                     // release lock while processing msg
  96             block->handle_msg(msg);
  97             guard.lock();
  98           }
  99         }
 100       }
 101       break;
 102
 103     default:
 104       assert(0);
 105     }
 106   }
 107 }
 108
 109 gr_tpb_thread_body::~gr_tpb_thread_body()
 110 {
 111 }