git.gag.com Git - debian/gnuradio/blob - mblock/src/lib/mb_worker.cc

   1 /* -*- c++ -*- */
   2 /*
   3  * Copyright 2007,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
  22 #ifdef HAVE_CONFIG_H
  23 #include <config.h>
  24 #endif
  25 #include <mb_worker.h>
  26 #include <mb_runtime_thread_per_block.h>
  27 #include <mblock/exception.h>
  28 #include <mblock/mblock.h>
  29 #include <mb_gettid.h>
  30 #include <mblock/msg_accepter.h>
  31 #include <iostream>
  32 #include <cstdio>
  33 #ifdef HAVE_SCHED_H
  34 #include <sched.h>
  35 #endif
  36
  37 #define VERBOSE 0               // define to 0 or 1
  38
  39 using namespace pmt;
  40
  41 static pmt_t s_worker_state_changed = pmt_intern("%worker-state-changed");
  42
  43
  44 mb_worker::mb_worker(mb_runtime_thread_per_block *runtime,
  45                      mb_mblock_maker_t maker,
  46                      const std::string &instance_name,
  47                      pmt_t user_arg)
  48   : omni_thread((void *) 0, PRIORITY_NORMAL),
  49     d_runtime(runtime), d_maker(maker),
  50     d_instance_name(instance_name), d_user_arg(user_arg),
  51     d_state_cond(&d_mutex), d_state(TS_UNINITIALIZED),
  52     d_why_dead(RIP_NOT_DEAD_YET)
  53 {
  54 }
  55
  56 #if 0
  57 mb_worker::~mb_worker()
  58 {
  59 }
  60 #endif
  61
  62 #ifdef HAVE_SCHED_SETAFFINITY
  63 static void
  64 set_affinity(const std::string &instance_name, const std::string &class_name)
  65 {
  66   //static int  counter = 0;
  67   cpu_set_t     mask;
  68   CPU_ZERO(&mask);
  69
  70   if (0){
  71
  72     //CPU_SET(counter & 0x1, &mask);
  73     //counter++;
  74     CPU_SET(0, &mask);
  75
  76     int r = sched_setaffinity(mb_gettid(), sizeof(mask), &mask);
  77     if (r == -1)
  78       perror("sched_setaffinity");
  79   }
  80 }
  81 #else
  82 static void
  83 set_affinity(const std::string &instance_name, const std::string &class_name)
  84 {
  85 }
  86 #endif
  87
  88 void
  89 mb_worker::set_state(worker_state_t state)
  90 {
  91   {
  92     omni_mutex_lock  l2(d_mutex);
  93
  94     d_state = state;                      // update our state
  95     d_state_cond.broadcast();             // Notify everybody who cares...
  96   }
  97
  98   // send msg to runtime, telling it something changed.
  99   (*d_runtime->accepter())(s_worker_state_changed, PMT_F, PMT_F, MB_PRI_BEST);
 100 }
 101
 102 void *
 103 mb_worker::run_undetached(void *ignored)
 104 {
 105   // FIXME add pthread_sigmask stuff
 106
 107   //set_affinity(d_instance_name, d_class_name);
 108   set_affinity(d_instance_name, "");
 109
 110   try {
 111     worker_thread_top_level();
 112     d_why_dead = RIP_EXIT;
 113   }
 114   catch (mbe_terminate){
 115     d_why_dead = RIP_TERMINATE;
 116   }
 117   catch (mbe_exit){
 118     d_why_dead = RIP_EXIT;
 119   }
 120   catch (std::logic_error e){
 121     if (d_why_dead == RIP_NOT_DEAD_YET)
 122       d_why_dead = RIP_UNHANDLED_EXCEPTION;
 123
 124     std::cerr << "\nmb_worker::run_undetached: unhandled exception:\n";
 125     std::cerr << "  " << e.what() << std::endl;
 126   }
 127   catch (...){
 128     if (d_why_dead == RIP_NOT_DEAD_YET)
 129       d_why_dead = RIP_UNHANDLED_EXCEPTION;
 130   }
 131
 132   if (VERBOSE)
 133     std::cerr << "\nrun_undetached: about to return, d_why_dead = "
 134               << d_why_dead << std::endl;
 135
 136   set_state(TS_DEAD);
 137   return 0;
 138 }
 139
 140 void
 141 mb_worker::worker_thread_top_level()
 142 {
 143   if (VERBOSE)
 144     std::cerr << "worker_thread_top_level (enter):" << std::endl
 145               << "  instance_name: " << d_instance_name << std::endl
 146               << "  omnithread id: " << id() << std::endl
 147               << "  gettid:        " << mb_gettid() << std::endl
 148               << "  getpid:        " << getpid() << std::endl;
 149
 150   cause_of_death_t pending_cause_of_death = RIP_NOT_DEAD_YET;
 151
 152   try {
 153     pending_cause_of_death = RIP_CTOR_EXCEPTION;
 154     d_mblock = d_maker(d_runtime, d_instance_name, d_user_arg);
 155
 156     if (VERBOSE)
 157       std::cerr << "worker_thread_top_level (post-construction):" << std::endl
 158                 << "  instance_name: " << d_instance_name << std::endl;
 159
 160     pending_cause_of_death = RIP_INIT_EXCEPTION;
 161     d_mblock->initial_transition();
 162
 163     if (VERBOSE)
 164       std::cerr << "worker_thread_top_level (post-initial-transition):" << std::endl
 165                 << "  instance_name: " << d_instance_name << std::endl;
 166
 167     set_state(TS_RUNNING);
 168
 169     pending_cause_of_death = RIP_UNHANDLED_EXCEPTION;
 170     d_mblock->main_loop();
 171   }
 172   catch (...){
 173     d_why_dead = pending_cause_of_death;
 174     throw;
 175   }
 176
 177   if (VERBOSE)
 178     std::cerr << "worker_thread_top_level (exit):" << std::endl
 179               << "  instance_name: " << d_instance_name << std::endl;
 180 }