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 <thread>
  23
  24 #include "branch_hints.hpp"
  25 #include "callback_system.hpp"
  26
  27 #include "nova-tt/semaphore.hpp"
  28 #include "nova-tt/thread_priority.hpp"
  29
  30 #include <boost/atomic.hpp>
  31 #include <boost/checked_delete.hpp>
  32
  33
  34 namespace nova {
  35
  36 namespace detail {
  37
  38 struct nop_functor
  39 {
  40     void operator()() const {}
  41 };
  42
  43 } /* namespace detail */
  44
  45
  46
  47 template <class callback_type,
  48           bool mpmc = true,
  49           class callback_deleter = boost::checked_deleter<callback_type> >
  50 class callback_interpreter:
  51     callback_system<callback_type, mpmc, callback_deleter>
  52 {
  53     typedef callback_system<callback_type, mpmc, callback_deleter> super_t;
  54
  55 public:
  56     callback_interpreter(void):
  57         sem(0), running(false)
  58     {}
  59
  60     void add_callback(callback_type * cb)
  61     {
  62         super_t::add_callback(cb);
  63         sem.post();
  64     }
  65
  66     void run(void)
  67     {
  68         running.store(true, boost::memory_order_relaxed);
  69         perform();
  70     }
  71
  72     void terminate(void)
  73     {
  74         running.store(false, boost::memory_order_relaxed);
  75         sem.post();
  76     }
  77
  78 protected:
  79     void run(semaphore & sync_sem)
  80     {
  81         running.store(true, boost::memory_order_relaxed);
  82         sync_sem.post();
  83         perform();
  84     }
  85
  86 private:
  87     void perform(void)
  88     {
  89         do
  90         {
  91             sem.wait();
  92             super_t::run_callbacks();
  93         } while(likely(running.load(boost::memory_order_relaxed)));
  94     }
  95
  96 protected:
  97     semaphore sem;
  98     boost::atomic<bool> running;
  99 };
 100
 101
 102 template <class callback_type,
 103           class init_functor = detail::nop_functor,
 104           class callback_deleter = boost::checked_deleter<callback_type> >
 105 class threaded_callback_interpreter:
 106     public callback_interpreter<callback_type, true, callback_deleter>,
 107     init_functor
 108 {
 109     typedef callback_interpreter<callback_type, true, callback_deleter> super;
 110
 111     std::thread callback_thread;
 112
 113 public:
 114     threaded_callback_interpreter(void)
 115     {}
 116
 117     ~threaded_callback_interpreter(void)
 118     {
 119         if (super::running.load())
 120             join_thread();
 121     }
 122
 123     void start_thread(void)
 124     {
 125         using namespace std;
 126         semaphore sync_sem;
 127         semaphore_sync<semaphore> sync(sync_sem);
 128         thread thr(bind(&threaded_callback_interpreter::run_thread, this, ref(sync_sem)));
 129         callback_thread = 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         callback_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         using namespace std;
 160
 161         for (uint16_t i = 0; i != worker_thread_count; ++i)
 162             threads.emplace_back(std::bind(&callback_interpreter_threadpool::run, this, boost::ref(sync_sem)));
 163
 164         for (uint16_t i = 0; i != worker_thread_count; ++i)
 165             sync_sem.wait();
 166     }
 167
 168     ~callback_interpreter_threadpool(void)
 169     {
 170         if (super::running.load())
 171             join_threads();
 172     }
 173
 174     void join_threads(void)
 175     {
 176         super::running.store(false);
 177
 178         for (uint16_t i = 0; i != worker_thread_count_; ++i)
 179             super::sem.post();
 180
 181         for (std::thread & thread : threads)
 182             thread.join();
 183     }
 184
 185 private:
 186     void run(semaphore & sync_sem)
 187     {
 188 #ifdef NOVA_TT_PRIORITY_RT
 189         if (rt)
 190             thread_set_priority_rt(priority);
 191         else
 192 #endif
 193             thread_set_priority(priority);
 194
 195         init_functor::operator()();
 196         super::run(sync_sem);
 197     }
 198
 199     std::vector<std::thread> threads;
 200     uint16_t worker_thread_count_;
 201     uint16_t priority;
 202     bool rt;
 203 };
 204
 205
 206 } /* namespace nova */
 207
 208 #endif /* UTILITIES_CALLBACK_INTERPRETER_HPP */