src/util/thread.h

   1 /*
   2 Minetest
   3 Copyright (C) 2010-2013 celeron55, Perttu Ahola <celeron55@gmail.com>
   4
   5 This program is free software; you can redistribute it and/or modify
   6 it under the terms of the GNU Lesser General Public License as published by
   7 the Free Software Foundation; either version 2.1 of the License, or
   8 (at your option) any later version.
   9
  10 This program is distributed in the hope that it will be useful,
  11 but WITHOUT ANY WARRANTY; without even the implied warranty of
  12 MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
  13 GNU Lesser General Public License for more details.
  14
  15 You should have received a copy of the GNU Lesser General Public License along
  16 with this program; if not, write to the Free Software Foundation, Inc.,
  17 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301 USA.
  18 */
  19
  20 #ifndef UTIL_THREAD_HEADER
  21 #define UTIL_THREAD_HEADER
  22
  23 #include "../irrlichttypes.h"
  24 #include "../jthread/jthread.h"
  25 #include "../jthread/jmutex.h"
  26 #include "../jthread/jmutexautolock.h"
  27 #include "porting.h"
  28 #include "log.h"
  29
  30 template<typename T>
  31 class MutexedVariable
  32 {
  33 public:
  34         MutexedVariable(T value):
  35                 m_value(value)
  36         {
  37         }
  38
  39         T get()
  40         {
  41                 JMutexAutoLock lock(m_mutex);
  42                 return m_value;
  43         }
  44
  45         void set(T value)
  46         {
  47                 JMutexAutoLock lock(m_mutex);
  48                 m_value = value;
  49         }
  50
  51         // You'll want to grab this in a SharedPtr
  52         JMutexAutoLock * getLock()
  53         {
  54                 return new JMutexAutoLock(m_mutex);
  55         }
  56
  57         // You pretty surely want to grab the lock when accessing this
  58         T m_value;
  59
  60 private:
  61         JMutex m_mutex;
  62 };
  63
  64 /*
  65         A single worker thread - multiple client threads queue framework.
  66 */
  67 template<typename Key, typename T, typename Caller, typename CallerData>
  68 class GetResult
  69 {
  70 public:
  71         Key key;
  72         T item;
  73         std::pair<Caller, CallerData> caller;
  74 };
  75
  76 template<typename Key, typename T, typename Caller, typename CallerData>
  77 class ResultQueue: public MutexedQueue< GetResult<Key, T, Caller, CallerData> >
  78 {
  79 };
  80
  81 template<typename Caller, typename Data, typename Key, typename T>
  82 class CallerInfo
  83 {
  84 public:
  85         Caller caller;
  86         Data data;
  87         ResultQueue< Key, T, Caller, Data>* dest;
  88 };
  89
  90 template<typename Key, typename T, typename Caller, typename CallerData>
  91 class GetRequest
  92 {
  93 public:
  94         GetRequest()
  95         {
  96         }
  97         GetRequest(Key a_key)
  98         {
  99                 key = a_key;
 100         }
 101         ~GetRequest()
 102         {
 103         }
 104
 105         Key key;
 106         std::list<CallerInfo<Caller, CallerData, Key, T> > callers;
 107 };
 108
 109 /**
 110  * Notes for RequestQueue usage
 111  * @param Key unique key to identify a request for a specific resource
 112  * @param T ?
 113  * @param Caller unique id of calling thread
 114  * @param CallerData data passed back to caller
 115  */
 116 template<typename Key, typename T, typename Caller, typename CallerData>
 117 class RequestQueue
 118 {
 119 public:
 120         bool empty()
 121         {
 122                 return m_queue.empty();
 123         }
 124
 125         void add(Key key, Caller caller, CallerData callerdata,
 126                         ResultQueue<Key, T, Caller, CallerData> *dest)
 127         {
 128                 {
 129                         JMutexAutoLock lock(m_queue.getMutex());
 130
 131                         /*
 132                                 If the caller is already on the list, only update CallerData
 133                         */
 134                         for(typename std::deque< GetRequest<Key, T, Caller, CallerData> >::iterator
 135                                         i = m_queue.getQueue().begin();
 136                                         i != m_queue.getQueue().end(); ++i)
 137                         {
 138                                 GetRequest<Key, T, Caller, CallerData> &request = *i;
 139
 140                                 if(request.key == key)
 141                                 {
 142                                         for(typename std::list< CallerInfo<Caller, CallerData, Key, T> >::iterator
 143                                                         i = request.callers.begin();
 144                                                         i != request.callers.end(); ++i)
 145                                         {
 146                                                 CallerInfo<Caller, CallerData, Key, T> &ca = *i;
 147                                                 if(ca.caller == caller)
 148                                                 {
 149                                                         ca.data = callerdata;
 150                                                         return;
 151                                                 }
 152                                         }
 153                                         CallerInfo<Caller, CallerData, Key, T> ca;
 154                                         ca.caller = caller;
 155                                         ca.data = callerdata;
 156                                         ca.dest = dest;
 157                                         request.callers.push_back(ca);
 158                                         return;
 159                                 }
 160                         }
 161                 }
 162
 163                 /*
 164                         Else add a new request to the queue
 165                 */
 166
 167                 GetRequest<Key, T, Caller, CallerData> request;
 168                 request.key = key;
 169                 CallerInfo<Caller, CallerData, Key, T> ca;
 170                 ca.caller = caller;
 171                 ca.data = callerdata;
 172                 ca.dest = dest;
 173                 request.callers.push_back(ca);
 174
 175                 m_queue.push_back(request);
 176         }
 177
 178         GetRequest<Key, T, Caller, CallerData> pop(unsigned int timeout_ms)
 179         {
 180                 return m_queue.pop_front(timeout_ms);
 181         }
 182
 183         GetRequest<Key, T, Caller, CallerData> pop()
 184         {
 185                 return m_queue.pop_frontNoEx();
 186         }
 187
 188         void pushResult(GetRequest<Key, T, Caller, CallerData> req,
 189                                         T res) {
 190
 191                 for(typename std::list< CallerInfo<Caller, CallerData, Key, T> >::iterator
 192                                 i = req.callers.begin();
 193                                 i != req.callers.end(); ++i)
 194                 {
 195                         CallerInfo<Caller, CallerData, Key, T> &ca = *i;
 196
 197                         GetResult<Key,T,Caller,CallerData> result;
 198
 199                         result.key = req.key;
 200                         result.item = res;
 201                         result.caller.first = ca.caller;
 202                         result.caller.second = ca.data;
 203
 204                         ca.dest->push_back(result);
 205                 }
 206         }
 207
 208 private:
 209         MutexedQueue< GetRequest<Key, T, Caller, CallerData> > m_queue;
 210 };
 211
 212 class UpdateThread : public JThread
 213 {
 214 private:
 215         JSemaphore m_update_sem;
 216
 217 protected:
 218         virtual void doUpdate() = 0;
 219         virtual const char *getName() = 0;
 220
 221 public:
 222         UpdateThread()
 223         {
 224         }
 225         ~UpdateThread()
 226         {}
 227
 228         void deferUpdate()
 229         {
 230                 m_update_sem.Post();
 231         }
 232
 233         void Stop()
 234         {
 235                 JThread::Stop();
 236
 237                 // give us a nudge
 238                 m_update_sem.Post();
 239         }
 240
 241         void *Thread()
 242         {
 243                 ThreadStarted();
 244
 245                 const char *thread_name = getName();
 246
 247                 log_register_thread(thread_name);
 248
 249                 DSTACK(__FUNCTION_NAME);
 250
 251                 BEGIN_DEBUG_EXCEPTION_HANDLER
 252
 253                 porting::setThreadName(thread_name);
 254
 255                 while (!StopRequested()) {
 256
 257                         m_update_sem.Wait();
 258
 259                         // Empty the queue, just in case doUpdate() is expensive
 260                         while (m_update_sem.GetValue()) m_update_sem.Wait();
 261
 262                         if (StopRequested()) break;
 263
 264                         doUpdate();
 265                 }
 266                 END_DEBUG_EXCEPTION_HANDLER(errorstream)
 267
 268                 return NULL;
 269         }
 270 };
 271
 272 #endif
 273