server/supernova/utilities/callback_interpreter.hpp

   1 //  templated callback system
   2 //  Copyright (C) 2008, 2009 Tim Blechmann
   3 //
   4 //  This program is free software; you can redistribute it and/or modify
   5 //  it under the terms of the GNU General Public License as published by
   6 //  the Free Software Foundation; either version 2 of the License, or
   7 //  (at your option) any later version.
   8 //
   9 //  This program is distributed in the hope that it will be useful,
  10 //  but WITHOUT ANY WARRANTY; without even the implied warranty of
  11 //  MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
  12 //  GNU General Public License for more details.
  13 //
  14 //  You should have received a copy of the GNU General Public License
  15 //  along with this program; see the file COPYING.  If not, write to
  16 //  the Free Software Foundation, Inc., 59 Temple Place - Suite 330,
  17 //  Boston, MA 02111-1307, USA.
  18
  19 #ifndef UTILITIES_CALLBACK_INTERPRETER_HPP
  20 #define UTILITIES_CALLBACK_INTERPRETER_HPP
  21
  22 #include "branch_hints.hpp"
  23 #include "callback_system.hpp"
  24
  25 #include "nova-tt/semaphore.hpp"
  26 #include "nova-tt/thread_priority.hpp"
  27
  28 #include <boost/atomic.hpp>
  29 #include <boost/bind.hpp>
  30 #include <boost/checked_delete.hpp>
  31 #include <boost/ref.hpp>
  32 #include <boost/thread.hpp>
  33
  34
  35 namespace nova
  36 {
  37
  38 namespace detail
  39 {
  40 struct nop_functor
  41 {
  42     void operator()() const {}
  43 };
  44 } /* namespace detail */
  45
  46
  47
  48 template <class callback_type,
  49           bool mpmc = true,
  50           class callback_deleter = boost::checked_deleter<callback_type> >
  51 class callback_interpreter:
  52     callback_system<callback_type, mpmc, callback_deleter>
  53 {
  54     typedef callback_system<callback_type, mpmc, callback_deleter> super_t;
  55
  56 public:
  57     callback_interpreter(void):
  58         sem(0), running(false)
  59     {}
  60
  61     void add_callback(callback_type * cb)
  62     {
  63         super_t::add_callback(cb);
  64         sem.post();
  65     }
  66
  67     void run(void)
  68     {
  69         running.store(true, boost::memory_order_relaxed);
  70         perform();
  71     }
  72
  73     void terminate(void)
  74     {
  75         running.store(false, boost::memory_order_relaxed);
  76         sem.post();
  77     }
  78
  79 protected:
  80     void run(semaphore & sync_sem)
  81     {
  82         running.store(true, boost::memory_order_relaxed);
  83         sync_sem.post();
  84         perform();
  85     }
  86
  87 private:
  88     void perform(void)
  89     {
  90         do
  91         {
  92             sem.wait();
  93             super_t::run_callbacks();
  94         } while(likely(running.load(boost::memory_order_relaxed)));
  95     }
  96
  97 protected:
  98     semaphore sem;
  99     boost::atomic<bool> running;
 100 };
 101
 102
 103 template <class callback_type,
 104           class init_functor = detail::nop_functor,
 105           class callback_deleter = boost::checked_deleter<callback_type> >
 106 class threaded_callback_interpreter:
 107     public callback_interpreter<callback_type, true, callback_deleter>,
 108     init_functor
 109 {
 110     typedef callback_interpreter<callback_type, true, callback_deleter> super;
 111
 112     boost::thread thread;
 113
 114 public:
 115     threaded_callback_interpreter(void)
 116     {}
 117
 118     ~threaded_callback_interpreter(void)
 119     {
 120         if (super::running.load())
 121             join_thread();
 122     }
 123
 124     void start_thread(void)
 125     {
 126         semaphore sync_sem;
 127         semaphore_sync<semaphore> sync(sync_sem);
 128         boost::thread thr(boost::bind(&threaded_callback_interpreter::run_thread, this, boost::ref(sync_sem)));
 129         thread = boost::move(thr);
 130     }
 131
 132     void run_thread(semaphore & sync_sem)
 133     {
 134         init_functor::operator()();
 135         super::run(sync_sem);
 136     }
 137
 138     void join_thread(void)
 139     {
 140         super::terminate();
 141         thread.join();
 142     }
 143 };
 144
 145 template <class callback_type,
 146           class init_functor = detail::nop_functor,
 147           class callback_deleter = boost::checked_deleter<callback_type> >
 148 class callback_interpreter_threadpool:
 149     public callback_interpreter<callback_type, true, callback_deleter>,
 150     init_functor
 151 {
 152     typedef callback_interpreter<callback_type, true, callback_deleter> super;
 153
 154 public:
 155     callback_interpreter_threadpool(uint16_t worker_thread_count, bool rt, uint16_t priority):
 156         worker_thread_count_(worker_thread_count), priority(priority), rt(rt)
 157     {
 158         semaphore sync_sem;
 159         for (uint16_t i = 0; i != worker_thread_count; ++i)
 160             threads.create_thread(boost::bind(&callback_interpreter_threadpool::run, this, boost::ref(sync_sem)));
 161
 162         for (uint16_t i = 0; i != worker_thread_count; ++i)
 163             sync_sem.wait();
 164     }
 165
 166     ~callback_interpreter_threadpool(void)
 167     {
 168         if (super::running.load())
 169             join_threads();
 170     }
 171
 172     void join_threads(void)
 173     {
 174         super::running.store(false);
 175
 176         for (uint16_t i = 0; i != worker_thread_count_; ++i)
 177             super::sem.post();
 178         threads.join_all();
 179     }
 180
 181 private:
 182     void run(semaphore & sync_sem)
 183     {
 184 #ifdef NOVA_TT_PRIORITY_RT
 185         if (rt)
 186             thread_set_priority_rt(priority);
 187         else
 188 #endif
 189             thread_set_priority(priority);
 190
 191         init_functor::operator()();
 192         super::run(sync_sem);
 193     }
 194
 195     boost::thread_group threads;
 196     uint16_t worker_thread_count_;
 197     uint16_t priority;
 198     bool rt;
 199 };
 200
 201
 202 } /* namespace nova */
 203
 204 #endif /* UTILITIES_CALLBACK_INTERPRETER_HPP */