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

   1 /* -*- c++ -*- */
   2 /*
   3  * Copyright 2006,2007 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
  18  * along with GNU Radio; see the file COPYING.  If not, write to
  19  * the Free Software Foundation, Inc., 51 Franklin Street,
  20  * Boston, MA 02110-1301, USA.
  21  */
  22
  23 #ifdef HAVE_CONFIG_H
  24 #include "config.h"
  25 #endif
  26
  27 #include <gr_runtime.h>
  28 #include <gr_runtime_impl.h>
  29 #include <gr_simple_flowgraph.h>
  30 #include <gr_hier_block2.h>
  31 #include <gr_hier_block2_detail.h>
  32 #include <gr_local_sighandler.h>
  33
  34 #ifdef HAVE_SIGNAL_H
  35 #include <signal.h>
  36 #endif
  37
  38 #include <stdexcept>
  39 #include <iostream>
  40
  41 #define GR_RUNTIME_IMPL_DEBUG 0
  42
  43 static gr_runtime_impl *s_runtime = 0;
  44
  45 // FIXME: This prevents using more than one gr_runtime instance
  46 void
  47 runtime_sigint_handler(int signum)
  48 {
  49   if (GR_RUNTIME_IMPL_DEBUG)
  50     std::cout << "SIGINT received, calling stop() on all threads" << std::endl;
  51
  52   if (s_runtime)
  53     s_runtime->stop();
  54 }
  55
  56 gr_runtime_impl::gr_runtime_impl(gr_hier_block2_sptr top_block, gr_runtime *owner)
  57   : d_running(false),
  58     d_top_block(top_block),
  59     d_sfg(gr_make_simple_flowgraph()),
  60     d_owner(owner)
  61 {
  62   s_runtime = this;
  63   top_block->set_runtime(d_owner);
  64 }
  65
  66 gr_runtime_impl::~gr_runtime_impl()
  67 {
  68   s_runtime = 0; // don't call delete we don't own these
  69   d_owner = 0;
  70 }
  71
  72 void
  73 gr_runtime_impl::start()
  74 {
  75   if (GR_RUNTIME_IMPL_DEBUG)
  76     std::cout << "start: entered" << std::endl;
  77
  78   if (d_running)
  79     throw std::runtime_error("already running");
  80
  81   // Create new simple flow graph by flattening hierarchical block
  82   d_sfg->d_detail->reset();
  83   d_top_block->d_detail->flatten(d_sfg);
  84
  85   // Validate new simple flow graph and wire it up
  86   d_sfg->d_detail->validate();
  87   d_sfg->d_detail->setup_connections();
  88
  89   // Execute scheduler threads
  90   start_threads();
  91 }
  92
  93 void
  94 gr_runtime_impl::start_threads()
  95 {
  96   if (GR_RUNTIME_IMPL_DEBUG)
  97     std::cout << "start_threads: entered" << std::endl;
  98
  99   d_graphs = d_sfg->d_detail->partition();
 100   for (std::vector<gr_block_vector_t>::iterator p = d_graphs.begin();
 101        p != d_graphs.end(); p++) {
 102     gr_scheduler_thread *thread = new gr_scheduler_thread(*p);
 103     d_threads.push_back(thread);
 104     if (GR_RUNTIME_IMPL_DEBUG)
 105       std::cout << "start_threads: starting " << thread << std::endl;
 106     thread->start();
 107   }
 108
 109   d_running = true;
 110 }
 111
 112 void
 113 gr_runtime_impl::stop()
 114 {
 115   if (GR_RUNTIME_IMPL_DEBUG)
 116     std::cout << "stop: entered" << std::endl;
 117
 118   for (gr_scheduler_thread_viter_t p = d_threads.begin(); p != d_threads.end(); p++) {
 119     if (GR_RUNTIME_IMPL_DEBUG)
 120       std::cout << "stop: stopping thread " << (*p) << std::endl;
 121     (*p)->stop();
 122   }
 123
 124   d_running = false;
 125 }
 126
 127 void
 128 gr_runtime_impl::wait()
 129 {
 130   if (GR_RUNTIME_IMPL_DEBUG)
 131     std::cout << "wait: entered" << std::endl;
 132
 133   void *dummy_status; // don't ever dereference this
 134   gr_local_sighandler sigint(SIGINT, runtime_sigint_handler);
 135
 136   for (gr_scheduler_thread_viter_t p = d_threads.begin(); p != d_threads.end(); p++) {
 137     if (GR_RUNTIME_IMPL_DEBUG)
 138       std::cout << "wait: joining thread " << (*p) << std::endl;
 139     (*p)->join(&dummy_status); // pthreads will self-delete, so pointer is now dead
 140     (*p) = 0; // FIXME: switch to stl::list and actually remove from container
 141     if (GR_RUNTIME_IMPL_DEBUG)
 142       std::cout << "wait: join returned" << std::endl;
 143   }
 144
 145   d_threads.clear();
 146 }
 147
 148
 149 // N.B. lock() and unlock() cannot be called from a flow graph thread or
 150 // deadlock will occur when reconfiguration happens
 151 void
 152 gr_runtime_impl::lock()
 153 {
 154   omni_mutex_lock lock(d_reconf);
 155   d_lock_count++;
 156   if (GR_RUNTIME_IMPL_DEBUG)
 157     std::cout << "runtime: locked, count = " << d_lock_count <<  std::endl;
 158 }
 159
 160 void
 161 gr_runtime_impl::unlock()
 162 {
 163   omni_mutex_lock lock(d_reconf);
 164   if (d_lock_count == 0)
 165     throw std::runtime_error("unpaired unlock() call");
 166
 167   d_lock_count--;
 168   if (GR_RUNTIME_IMPL_DEBUG)
 169     std::cout << "runtime: unlocked, count = " << d_lock_count << std::endl;
 170
 171   if (d_lock_count == 0)
 172     restart();
 173 }
 174
 175 void
 176 gr_runtime_impl::restart()
 177 {
 178   if (GR_RUNTIME_IMPL_DEBUG)
 179     std::cout << "restart: entered" << std::endl;
 180
 181   if (!d_running)
 182     throw std::runtime_error("not running");
 183
 184   // Stop scheduler threads and wait for completion
 185   stop();
 186   wait();
 187   if (GR_RUNTIME_IMPL_DEBUG)
 188     std::cout << "restart: threads stopped" << std::endl;
 189
 190   // Create new simple flow graph
 191   gr_simple_flowgraph_sptr new_sfg = gr_make_simple_flowgraph();
 192   d_top_block->d_detail->flatten(new_sfg);
 193   new_sfg->validate();
 194   new_sfg->d_detail->merge_connections(d_sfg);
 195
 196   if (GR_RUNTIME_IMPL_DEBUG)
 197     std::cout << "restart: replacing old flow graph with new" << std::endl;
 198   d_sfg = new_sfg;
 199
 200   start_threads();
 201 }
 202
 203 gr_scheduler_thread::gr_scheduler_thread(gr_block_vector_t graph) :
 204   omni_thread(NULL, PRIORITY_NORMAL),
 205   d_sts(gr_make_single_threaded_scheduler(graph))
 206 {
 207 }
 208
 209 gr_scheduler_thread::~gr_scheduler_thread()
 210 {
 211 }
 212
 213 void gr_scheduler_thread::start()
 214 {
 215   start_undetached();
 216 }
 217
 218 void *
 219 gr_scheduler_thread::run_undetached(void *arg)
 220 {
 221   // First code to run in new thread context
 222
 223   // Mask off SIGINT in this thread to gaurantee mainline thread gets signal
 224 #ifdef HAVE_SIGPROCMASK
 225   sigset_t old_set;
 226   sigset_t new_set;
 227   sigemptyset(&new_set);
 228   sigaddset(&new_set, SIGINT);
 229   sigprocmask(SIG_BLOCK, &new_set, &old_set);
 230 #endif
 231   // Run the single-threaded scheduler
 232   d_sts->run();
 233   return 0;
 234 }
 235
 236 void
 237 gr_scheduler_thread::stop()
 238 {
 239   d_sts->stop();
 240 }