server/supernova/dsp_thread_queue/dsp_thread.hpp

   1 //  dsp thread
   2 //  Copyright (C) 2007, 2008, 2009, 2010 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 DSP_THREAD_QUEUE_DSP_THREAD_HPP
  20 #define DSP_THREAD_QUEUE_DSP_THREAD_HPP
  21
  22 #include <cstdint>
  23 #include <iostream>
  24 #include <thread>
  25
  26 #include <boost/ptr_container/ptr_vector.hpp>
  27
  28 #include "dsp_thread_queue.hpp"
  29 #include "../utilities/malloc_aligned.hpp"
  30 #include "nova-tt/mlock.hpp"
  31
  32 namespace nova {
  33
  34 using std::uint16_t;
  35
  36 struct nop_thread_init
  37 {
  38     nop_thread_init(void)
  39     {}
  40
  41     template <typename Arg>
  42     nop_thread_init(Arg const &)
  43     {}
  44
  45     void operator()(int thread_index)
  46     {}
  47 };
  48
  49
  50 /** dsp helper thread
  51  *
  52  *  the dsp helper threads are running with a high real-time priority and are
  53  *  pinned to a specific cpu
  54  */
  55 template <typename runnable,
  56           typename thread_init_functor = nop_thread_init,
  57           typename Alloc = std::allocator<void*>
  58          >
  59 class dsp_thread:
  60     public boost::noncopyable,
  61     public thread_init_functor
  62 {
  63     typedef nova::dsp_queue_interpreter<runnable, Alloc> dsp_queue_interpreter;
  64
  65 public:
  66     dsp_thread(dsp_queue_interpreter & interpreter, uint16_t index, size_t stack_size,
  67                thread_init_functor const & thread_init = thread_init_functor()):
  68         thread_init_functor(thread_init), interpreter(interpreter), stop(false), index(index), stack_ (NULL)
  69     {
  70         if (stack_size) {
  71                 stack_ = malloc_aligned<char>(stack_size);
  72             if (stack_ == NULL)
  73                 throw std::bad_alloc();
  74             // touch stack to avoid page faults
  75             for (int i = 0; i != stack_size; ++i)
  76                 stack_[i] = 0;
  77             mlock(stack_, stack_size);
  78         }
  79     }
  80
  81     ~dsp_thread(void)
  82     {
  83         if (stack_)
  84             free_aligned(stack_);
  85     }
  86
  87     /** thread function
  88      * */
  89     void run(void)
  90     {
  91         thread_init_functor::operator()(index);
  92
  93         for (;;) {
  94             cycle_sem.wait();
  95             if (unlikely(stop.load(boost::memory_order_acquire)))
  96                 return;
  97
  98             interpreter.tick(index);
  99         }
 100     }
 101
 102     static void * run_static(void* arg)
 103     {
 104         dsp_thread * self = static_cast<dsp_thread*>(arg);
 105         self->run();
 106         return NULL;
 107     }
 108
 109     void terminate(void)
 110     {
 111         stop.store(true, boost::memory_order_release);
 112         wake_thread();
 113     }
 114
 115     void wake_thread(void)
 116     {
 117         cycle_sem.post();
 118     }
 119
 120
 121     char * stack(void)
 122     {
 123         return stack_;
 124     }
 125
 126 private:
 127     semaphore cycle_sem;
 128     dsp_queue_interpreter & interpreter;
 129     boost::atomic<bool> stop;
 130     uint16_t index;
 131     char * stack_;
 132 };
 133
 134 /** \brief container for all dsp threads
 135  *
 136  *  - no care is taken, that dsp_threads::run is executed on a valid instance
 137  *
 138  * */
 139 template <typename runnable,
 140           typename thread_init_functor = nop_thread_init,
 141           typename Alloc = std::allocator<void*>
 142          >
 143 class dsp_threads
 144 {
 145     typedef nova::dsp_queue_interpreter<runnable, Alloc> dsp_queue_interpreter;
 146
 147     typedef nova::dsp_thread<runnable, thread_init_functor, Alloc> dsp_thread;
 148
 149
 150 public:
 151     typedef typename dsp_queue_interpreter::node_count_t node_count_t;
 152     typedef typename dsp_queue_interpreter::thread_count_t thread_count_t;
 153
 154     typedef std::unique_ptr<dsp_thread_queue<runnable, Alloc> > dsp_thread_queue_ptr;
 155
 156     dsp_threads(thread_count_t count, thread_init_functor const & init_functor = thread_init_functor()):
 157         interpreter(std::min(count, (thread_count_t)std::thread::hardware_concurrency()))
 158     {
 159         set_dsp_thread_count(interpreter.get_thread_count(), init_functor);
 160     }
 161
 162     void run(void)
 163     {
 164         bool run_tick = interpreter.init_tick();
 165         if (likely(run_tick)) {
 166             wake_threads();
 167             interpreter.tick_master();
 168         }
 169     }
 170
 171     /** reset queue
 172      *
 173      *  don't call, if threads are currently accessing the queue
 174      * */
 175     dsp_thread_queue_ptr reset_queue(dsp_thread_queue_ptr && new_queue)
 176     {
 177         dsp_thread_queue_ptr ret = interpreter.reset_queue(std::move(new_queue));
 178         return std::move(ret);
 179     }
 180
 181     dsp_thread_queue_ptr release_queue(void)
 182     {
 183         return interpreter.release_queue();
 184     }
 185
 186 private:
 187 #if _POSIX_C_SOURCE >= 200112L || _XOPEN_SOURCE >= 600
 188     // we can set the stack
 189     static const int stack_size = 524288;
 190     struct thread_group
 191     {
 192         void push_back(pthread_t t)
 193         {
 194             thread_group_.push_back(t);
 195         }
 196
 197         void join_all(void)
 198         {
 199             for(pthread_t & thread : thread_group_) {
 200                 void * ret;
 201                 int err = pthread_join(thread, &ret);
 202                 if (err)
 203                     printf("Error when joining helper thread\n");
 204             }
 205         }
 206
 207         std::vector<pthread_t> thread_group_;
 208     };
 209
 210     void start_threads_impl(void)
 211     {
 212         for(dsp_thread & thread : threads) {
 213             pthread_attr_t attr;
 214             pthread_attr_init(&attr);
 215             int err = pthread_attr_setstack(&attr, thread.stack(), stack_size);
 216             if (err)
 217                 throw std::logic_error("Cannot set stack of DSP helper thread");
 218
 219             pthread_t thread_id;
 220             err = pthread_create(&thread_id, &attr, dsp_thread::run_static, &thread);
 221             if (err)
 222                 throw std::runtime_error("Cannot create DSP helper thread");
 223             thread_group_.push_back(thread_id);
 224             pthread_attr_destroy(&attr);
 225         }
 226     }
 227 #else
 228     struct thread_group
 229     {
 230         void join_all(void)
 231         {
 232             for (std::thread & thread : threads)
 233                 thread.join();
 234         }
 235
 236         std::vector<std::thread> threads;
 237     };
 238
 239     static const int stack_size = 0;
 240
 241     void start_threads_impl(void)
 242     {
 243         for(dsp_thread & thread : threads)
 244             thread_group_.threads.push_back(std::move(std::thread(std::bind(&dsp_thread::run, std::ref(thread)))));
 245     }
 246
 247 #endif
 248
 249     void set_dsp_thread_count(thread_count_t count, thread_init_functor const & init_functor)
 250     {
 251         for (thread_count_t i = 1; i != count; ++i)
 252             threads.push_back(new dsp_thread(interpreter, i, stack_size, init_functor));
 253         assert(threads.size() == std::size_t(count-1));
 254     }
 255
 256 public:
 257     /** thread handling */
 258     /* @{ */
 259     void start_threads(void)
 260     {
 261         start_threads_impl();
 262     }
 263
 264     void terminate_threads(void)
 265     {
 266         for (dsp_thread & thread : threads)
 267             thread.terminate();
 268
 269         thread_group_.join_all();
 270     }
 271     /* @} */
 272
 273 private:
 274     /** wake dsp threads */
 275     void wake_threads(void)
 276     {
 277         for (thread_count_t i = 0; i != interpreter.get_used_helper_threads(); ++i)
 278             threads[i].wake_thread();
 279     }
 280
 281     dsp_queue_interpreter interpreter;
 282
 283     boost::ptr_vector<dsp_thread> threads; /* container of dsp threads */
 284     thread_group thread_group_;
 285 };
 286
 287 } /* namespace nova */
 288
 289 #endif /* DSP_THREAD_QUEUE_DSP_THREAD_HPP */
 290