src/include/xbt/parmap.hpp

   1 /* A thread pool (C++ version).                                             */
   2
   3 /* Copyright (c) 2004-2018 The SimGrid Team. All rights reserved.           */
   4
   5 /* This program is free software; you can redistribute it and/or modify it
   6  * under the terms of the license (GNU LGPL) which comes with this package. */
   7
   8 #ifndef XBT_PARMAP_HPP
   9 #define XBT_PARMAP_HPP
  10
  11 #include "src/internal_config.h" // HAVE_FUTEX_H
  12 #include "src/kernel/context/Context.hpp"
  13 #include "xbt/xbt_os_thread.h"
  14
  15 #include <boost/optional.hpp>
  16
  17 #if HAVE_FUTEX_H
  18 #include <linux/futex.h>
  19 #include <sys/syscall.h>
  20 #endif
  21
  22 XBT_LOG_EXTERNAL_CATEGORY(xbt_parmap);
  23
  24 namespace simgrid {
  25 namespace xbt {
  26
  27 /** \addtogroup XBT_parmap
  28   * \ingroup XBT_misc
  29   * \brief Parallel map class
  30   * \{
  31   */
  32 template <typename T> class Parmap {
  33 public:
  34   Parmap(unsigned num_workers, e_xbt_parmap_mode_t mode);
  35   Parmap(const Parmap&) = delete;
  36   Parmap& operator=(const Parmap&) = delete;
  37   ~Parmap();
  38   void apply(void (*fun)(T), const std::vector<T>& data);
  39   boost::optional<T> next();
  40
  41 private:
  42   enum Flag { PARMAP_WORK, PARMAP_DESTROY };
  43
  44   /**
  45    * \brief Thread data transmission structure
  46    */
  47   class ThreadData {
  48   public:
  49     ThreadData(Parmap<T>& parmap, int id) : parmap(parmap), worker_id(id) {}
  50     Parmap<T>& parmap;
  51     int worker_id;
  52   };
  53
  54   /**
  55    * \brief Synchronization object (different specializations).
  56    */
  57   class Synchro {
  58   public:
  59     explicit Synchro(Parmap<T>& parmap) : parmap(parmap) {}
  60     virtual ~Synchro() = default;
  61     /**
  62      * \brief Wakes all workers and waits for them to finish the tasks.
  63      *
  64      * This function is called by the controller thread.
  65      */
  66     virtual void master_signal() = 0;
  67     /**
  68      * \brief Starts the parmap: waits for all workers to be ready and returns.
  69      *
  70      * This function is called by the controller thread.
  71      */
  72     virtual void master_wait() = 0;
  73     /**
  74      * \brief Ends the parmap: wakes the controller thread when all workers terminate.
  75      *
  76      * This function is called by all worker threads when they end (not including the controller).
  77      */
  78     virtual void worker_signal() = 0;
  79     /**
  80      * \brief Waits for some work to process.
  81      *
  82      * This function is called by each worker thread (not including the controller) when it has no more work to do.
  83      *
  84      * \param round  the expected round number
  85      */
  86     virtual void worker_wait(unsigned) = 0;
  87
  88     Parmap<T>& parmap;
  89   };
  90
  91   class PosixSynchro : public Synchro {
  92   public:
  93     explicit PosixSynchro(Parmap<T>& parmap);
  94     ~PosixSynchro();
  95     void master_signal();
  96     void master_wait();
  97     void worker_signal();
  98     void worker_wait(unsigned round);
  99
 100   private:
 101     xbt_os_cond_t ready_cond;
 102     xbt_os_mutex_t ready_mutex;
 103     xbt_os_cond_t done_cond;
 104     xbt_os_mutex_t done_mutex;
 105   };
 106
 107 #if HAVE_FUTEX_H
 108   class FutexSynchro : public Synchro {
 109   public:
 110     explicit FutexSynchro(Parmap<T>& parmap) : Synchro(parmap) {}
 111     void master_signal();
 112     void master_wait();
 113     void worker_signal();
 114     void worker_wait(unsigned);
 115
 116   private:
 117     static void futex_wait(unsigned* uaddr, unsigned val);
 118     static void futex_wake(unsigned* uaddr, unsigned val);
 119   };
 120 #endif
 121
 122   class BusyWaitSynchro : public Synchro {
 123   public:
 124     explicit BusyWaitSynchro(Parmap<T>& parmap) : Synchro(parmap) {}
 125     void master_signal();
 126     void master_wait();
 127     void worker_signal();
 128     void worker_wait(unsigned);
 129   };
 130
 131   static void* worker_main(void* arg);
 132   Synchro* new_synchro(e_xbt_parmap_mode_t mode);
 133   void work();
 134
 135   Flag status;              /**< is the parmap active or being destroyed? */
 136   unsigned work_round;      /**< index of the current round */
 137   xbt_os_thread_t* workers; /**< worker thread handlers */
 138   unsigned num_workers;     /**< total number of worker threads including the controller */
 139   Synchro* synchro;         /**< synchronization object */
 140
 141   unsigned thread_counter    = 0;       /**< number of workers that have done the work */
 142   void (*fun)(const T)       = nullptr; /**< function to run in parallel on each element of data */
 143   const std::vector<T>* data = nullptr; /**< parameters to pass to fun in parallel */
 144   std::atomic<unsigned> index;          /**< index of the next element of data to pick */
 145 };
 146
 147 /**
 148  * \brief Creates a parallel map object
 149  * \param num_workers number of worker threads to create
 150  * \param mode how to synchronize the worker threads
 151  */
 152 template <typename T> Parmap<T>::Parmap(unsigned num_workers, e_xbt_parmap_mode_t mode)
 153 {
 154   XBT_CDEBUG(xbt_parmap, "Create new parmap (%u workers)", num_workers);
 155
 156   /* Initialize the thread pool data structure */
 157   this->status      = PARMAP_WORK;
 158   this->work_round  = 0;
 159   this->workers     = new xbt_os_thread_t[num_workers];
 160   this->num_workers = num_workers;
 161   this->synchro     = new_synchro(mode);
 162
 163   /* Create the pool of worker threads */
 164   this->workers[0] = nullptr;
 165 #if HAVE_PTHREAD_SETAFFINITY
 166   int core_bind = 0;
 167 #endif
 168   for (unsigned i = 1; i < num_workers; i++) {
 169     ThreadData* data = new ThreadData(*this, i);
 170     this->workers[i] = xbt_os_thread_create(nullptr, worker_main, data, nullptr);
 171 #if HAVE_PTHREAD_SETAFFINITY
 172     xbt_os_thread_bind(this->workers[i], core_bind);
 173     if (core_bind != xbt_os_get_numcores() - 1)
 174       core_bind++;
 175     else
 176       core_bind = 0;
 177 #endif
 178   }
 179 }
 180
 181 /**
 182  * \brief Destroys a parmap
 183  */
 184 template <typename T> Parmap<T>::~Parmap()
 185 {
 186   status = PARMAP_DESTROY;
 187   synchro->master_signal();
 188
 189   for (unsigned i = 1; i < num_workers; i++)
 190     xbt_os_thread_join(workers[i], nullptr);
 191
 192   delete[] workers;
 193   delete synchro;
 194 }
 195
 196 /**
 197  * \brief Applies a list of tasks in parallel.
 198  * \param fun the function to call in parallel
 199  * \param data each element of this vector will be passed as an argument to fun
 200  */
 201 template <typename T> void Parmap<T>::apply(void (*fun)(T), const std::vector<T>& data)
 202 {
 203   /* Assign resources to worker threads (we are maestro here)*/
 204   this->fun   = fun;
 205   this->data  = &data;
 206   this->index = 0;
 207   this->synchro->master_signal(); // maestro runs futex_wake to wake all the minions (the working threads)
 208   this->work();                   // maestro works with its minions
 209   this->synchro->master_wait();   // When there is no more work to do, then maestro waits for the last minion to stop
 210   XBT_CDEBUG(xbt_parmap, "Job done"); //   ... and proceeds
 211 }
 212
 213 /**
 214  * \brief Returns a next task to process.
 215  *
 216  * Worker threads call this function to get more work.
 217  *
 218  * \return the next task to process, or throws a std::out_of_range exception if there is no more work
 219  */
 220 template <typename T> boost::optional<T> Parmap<T>::next()
 221 {
 222   unsigned index = this->index.fetch_add(1, std::memory_order_relaxed);
 223   if (index < this->data->size())
 224     return (*this->data)[index];
 225   else
 226     return boost::none;
 227 }
 228
 229 /**
 230  * \brief Main work loop: applies fun to elements in turn.
 231  */
 232 template <typename T> void Parmap<T>::work()
 233 {
 234   unsigned length = this->data->size();
 235   unsigned index  = this->index.fetch_add(1, std::memory_order_relaxed);
 236   while (index < length) {
 237     this->fun((*this->data)[index]);
 238     index = this->index.fetch_add(1, std::memory_order_relaxed);
 239   }
 240 }
 241
 242 /**
 243  * Get a synchronization object for given mode.
 244  * \param mode the synchronization mode
 245  */
 246 template <typename T> typename Parmap<T>::Synchro* Parmap<T>::new_synchro(e_xbt_parmap_mode_t mode)
 247 {
 248   if (mode == XBT_PARMAP_DEFAULT) {
 249 #if HAVE_FUTEX_H
 250     mode = XBT_PARMAP_FUTEX;
 251 #else
 252     mode = XBT_PARMAP_POSIX;
 253 #endif
 254   }
 255   Synchro* res;
 256   switch (mode) {
 257     case XBT_PARMAP_POSIX:
 258       res = new PosixSynchro(*this);
 259       break;
 260     case XBT_PARMAP_FUTEX:
 261 #if HAVE_FUTEX_H
 262       res = new FutexSynchro(*this);
 263 #else
 264       xbt_die("Futex is not available on this OS.");
 265 #endif
 266       break;
 267     case XBT_PARMAP_BUSY_WAIT:
 268       res = new BusyWaitSynchro(*this);
 269       break;
 270     default:
 271       THROW_IMPOSSIBLE;
 272   }
 273   return res;
 274 }
 275
 276 /**
 277  * \brief Main function of a worker thread.
 278  */
 279 template <typename T> void* Parmap<T>::worker_main(void* arg)
 280 {
 281   ThreadData* data      = static_cast<ThreadData*>(arg);
 282   Parmap<T>& parmap     = data->parmap;
 283   unsigned round        = 0;
 284   smx_context_t context = SIMIX_context_new(std::function<void()>(), nullptr, nullptr);
 285   SIMIX_context_set_current(context);
 286
 287   XBT_CDEBUG(xbt_parmap, "New worker thread created");
 288
 289   /* Worker's main loop */
 290   while (1) {
 291     round++;
 292     parmap.synchro->worker_wait(round);
 293     if (parmap.status == PARMAP_DESTROY)
 294       break;
 295
 296     XBT_CDEBUG(xbt_parmap, "Worker %d got a job", data->worker_id);
 297     parmap.work();
 298     parmap.synchro->worker_signal();
 299     XBT_CDEBUG(xbt_parmap, "Worker %d has finished", data->worker_id);
 300   }
 301   /* We are destroying the parmap */
 302   delete context;
 303   delete data;
 304   return nullptr;
 305 }
 306
 307 template <typename T> Parmap<T>::PosixSynchro::PosixSynchro(Parmap<T>& parmap) : Synchro(parmap)
 308 {
 309   ready_cond  = xbt_os_cond_init();
 310   ready_mutex = xbt_os_mutex_init();
 311   done_cond   = xbt_os_cond_init();
 312   done_mutex  = xbt_os_mutex_init();
 313 }
 314
 315 template <typename T> Parmap<T>::PosixSynchro::~PosixSynchro()
 316 {
 317   xbt_os_cond_destroy(ready_cond);
 318   xbt_os_mutex_destroy(ready_mutex);
 319   xbt_os_cond_destroy(done_cond);
 320   xbt_os_mutex_destroy(done_mutex);
 321 }
 322
 323 template <typename T> void Parmap<T>::PosixSynchro::master_signal()
 324 {
 325   xbt_os_mutex_acquire(ready_mutex);
 326   this->parmap.thread_counter = 1;
 327   this->parmap.work_round++;
 328   /* wake all workers */
 329   xbt_os_cond_broadcast(ready_cond);
 330   xbt_os_mutex_release(ready_mutex);
 331 }
 332
 333 template <typename T> void Parmap<T>::PosixSynchro::master_wait()
 334 {
 335   xbt_os_mutex_acquire(done_mutex);
 336   while (this->parmap.thread_counter < this->parmap.num_workers) {
 337     /* wait for all workers to be ready */
 338     xbt_os_cond_wait(done_cond, done_mutex);
 339   }
 340   xbt_os_mutex_release(done_mutex);
 341 }
 342
 343 template <typename T> void Parmap<T>::PosixSynchro::worker_signal()
 344 {
 345   xbt_os_mutex_acquire(done_mutex);
 346   this->parmap.thread_counter++;
 347   if (this->parmap.thread_counter == this->parmap.num_workers) {
 348     /* all workers have finished, wake the controller */
 349     xbt_os_cond_signal(done_cond);
 350   }
 351   xbt_os_mutex_release(done_mutex);
 352 }
 353
 354 template <typename T> void Parmap<T>::PosixSynchro::worker_wait(unsigned round)
 355 {
 356   xbt_os_mutex_acquire(ready_mutex);
 357   /* wait for more work */
 358   while (this->parmap.work_round != round) {
 359     xbt_os_cond_wait(ready_cond, ready_mutex);
 360   }
 361   xbt_os_mutex_release(ready_mutex);
 362 }
 363
 364 #if HAVE_FUTEX_H
 365 template <typename T> inline void Parmap<T>::FutexSynchro::futex_wait(unsigned* uaddr, unsigned val)
 366 {
 367   XBT_CVERB(xbt_parmap, "Waiting on futex %p", uaddr);
 368   syscall(SYS_futex, uaddr, FUTEX_WAIT_PRIVATE, val, nullptr, nullptr, 0);
 369 }
 370
 371 template <typename T> inline void Parmap<T>::FutexSynchro::futex_wake(unsigned* uaddr, unsigned val)
 372 {
 373   XBT_CVERB(xbt_parmap, "Waking futex %p", uaddr);
 374   syscall(SYS_futex, uaddr, FUTEX_WAKE_PRIVATE, val, nullptr, nullptr, 0);
 375 }
 376
 377 template <typename T> void Parmap<T>::FutexSynchro::master_signal()
 378 {
 379   __atomic_store_n(&this->parmap.thread_counter, 1, __ATOMIC_SEQ_CST);
 380   __atomic_add_fetch(&this->parmap.work_round, 1, __ATOMIC_SEQ_CST);
 381   /* wake all workers */
 382   futex_wake(&this->parmap.work_round, std::numeric_limits<int>::max());
 383 }
 384
 385 template <typename T> void Parmap<T>::FutexSynchro::master_wait()
 386 {
 387   unsigned count = __atomic_load_n(&this->parmap.thread_counter, __ATOMIC_SEQ_CST);
 388   while (count < this->parmap.num_workers) {
 389     /* wait for all workers to be ready */
 390     futex_wait(&this->parmap.thread_counter, count);
 391     count = __atomic_load_n(&this->parmap.thread_counter, __ATOMIC_SEQ_CST);
 392   }
 393 }
 394
 395 template <typename T> void Parmap<T>::FutexSynchro::worker_signal()
 396 {
 397   unsigned count = __atomic_add_fetch(&this->parmap.thread_counter, 1, __ATOMIC_SEQ_CST);
 398   if (count == this->parmap.num_workers) {
 399     /* all workers have finished, wake the controller */
 400     futex_wake(&this->parmap.thread_counter, std::numeric_limits<int>::max());
 401   }
 402 }
 403
 404 template <typename T> void Parmap<T>::FutexSynchro::worker_wait(unsigned round)
 405 {
 406   unsigned work_round = __atomic_load_n(&this->parmap.work_round, __ATOMIC_SEQ_CST);
 407   /* wait for more work */
 408   while (work_round != round) {
 409     futex_wait(&this->parmap.work_round, work_round);
 410     work_round = __atomic_load_n(&this->parmap.work_round, __ATOMIC_SEQ_CST);
 411   }
 412 }
 413 #endif
 414
 415 template <typename T> void Parmap<T>::BusyWaitSynchro::master_signal()
 416 {
 417   __atomic_store_n(&this->parmap.thread_counter, 1, __ATOMIC_SEQ_CST);
 418   __atomic_add_fetch(&this->parmap.work_round, 1, __ATOMIC_SEQ_CST);
 419 }
 420
 421 template <typename T> void Parmap<T>::BusyWaitSynchro::master_wait()
 422 {
 423   while (__atomic_load_n(&this->parmap.thread_counter, __ATOMIC_SEQ_CST) < this->parmap.num_workers) {
 424     xbt_os_thread_yield();
 425   }
 426 }
 427
 428 template <typename T> void Parmap<T>::BusyWaitSynchro::worker_signal()
 429 {
 430   __atomic_add_fetch(&this->parmap.thread_counter, 1, __ATOMIC_SEQ_CST);
 431 }
 432
 433 template <typename T> void Parmap<T>::BusyWaitSynchro::worker_wait(unsigned round)
 434 {
 435   /* wait for more work */
 436   while (__atomic_load_n(&this->parmap.work_round, __ATOMIC_SEQ_CST) != round) {
 437     xbt_os_thread_yield();
 438   }
 439 }
 440
 441 /** \} */
 442 }
 443 }
 444
 445 #endif