native_client_sdk/src/libraries/sdk_util/thread_pool.cc

   1 // Copyright (c) 2013 The Chromium Authors. All rights reserved.
   2 // Use of this source code is governed by a BSD-style license that can be
   3 // found in the LICENSE file.
   4
   5 #include "sdk_util/thread_pool.h"
   6
   7 #include <pthread.h>
   8 #include <semaphore.h>
   9 #include <stdio.h>
  10 #include <stdlib.h>
  11
  12 #include "sdk_util/auto_lock.h"
  13
  14 namespace sdk_util {
  15
  16 #ifdef __APPLE__
  17 #pragma clang diagnostic push
  18 #pragma clang diagnostic ignored "-Wdeprecated-declarations"
  19 #endif
  20
  21 // Initializes mutex, semaphores and a pool of threads.  If 0 is passed for
  22 // num_threads, all work will be performed on the dispatch thread.
  23 ThreadPool::ThreadPool(int num_threads)
  24     : threads_(NULL), counter_(0), num_threads_(num_threads), exiting_(false),
  25       user_data_(NULL), user_work_function_(NULL) {
  26   if (num_threads_ > 0) {
  27     int status;
  28     status = sem_init(&work_sem_, 0, 0);
  29     if (-1 == status) {
  30       fprintf(stderr, "Failed to initialize semaphore!\n");
  31       exit(-1);
  32     }
  33     status = sem_init(&done_sem_, 0, 0);
  34     if (-1 == status) {
  35       fprintf(stderr, "Failed to initialize semaphore!\n");
  36       exit(-1);
  37     }
  38     threads_ = new pthread_t[num_threads_];
  39     for (int i = 0; i < num_threads_; i++) {
  40       status = pthread_create(&threads_[i], NULL, WorkerThreadEntry, this);
  41       if (0 != status) {
  42         fprintf(stderr, "Failed to create thread!\n");
  43         exit(-1);
  44       }
  45     }
  46   }
  47 }
  48
  49 // Post exit request, wait for all threads to join, and cleanup.
  50 ThreadPool::~ThreadPool() {
  51   if (num_threads_ > 0) {
  52     PostExitAndJoinAll();
  53     delete[] threads_;
  54     sem_destroy(&done_sem_);
  55     sem_destroy(&work_sem_);
  56   }
  57 }
  58
  59 // Setup work parameters.  This function is called from the dispatch thread,
  60 // when all worker threads are sleeping.
  61 void ThreadPool::Setup(int counter, WorkFunction work, void *data) {
  62   counter_ = counter;
  63   user_work_function_ = work;
  64   user_data_ = data;
  65 }
  66
  67 // Return decremented task counter.  This function
  68 // can be called from multiple threads at any given time.
  69 int ThreadPool::DecCounter() {
  70   return AtomicAddFetch(&counter_, -1);
  71 }
  72
  73 // Set exit flag, post and join all the threads in the pool.  This function is
  74 // called only from the dispatch thread, and only when all worker threads are
  75 // sleeping.
  76 void ThreadPool::PostExitAndJoinAll() {
  77   exiting_ = true;
  78   // Wake up all the sleeping worker threads.
  79   for (int i = 0; i < num_threads_; ++i)
  80     sem_post(&work_sem_);
  81   void* retval;
  82   for (int i = 0; i < num_threads_; ++i)
  83     pthread_join(threads_[i], &retval);
  84 }
  85
  86 // Main work loop - one for each worker thread.
  87 void ThreadPool::WorkLoop() {
  88   while (true) {
  89     // Wait for work. If no work is availble, this thread will sleep here.
  90     sem_wait(&work_sem_);
  91     if (exiting_) break;
  92     while (true) {
  93       // Grab a task index to work on from the counter.
  94       int task_index = DecCounter();
  95       if (task_index < 0)
  96         break;
  97       user_work_function_(task_index, user_data_);
  98     }
  99     // Post to dispatch thread work is done.
 100     sem_post(&done_sem_);
 101   }
 102 }
 103
 104 // pthread entry point for a worker thread.
 105 void* ThreadPool::WorkerThreadEntry(void* thiz) {
 106   static_cast<ThreadPool*>(thiz)->WorkLoop();
 107   return NULL;
 108 }
 109
 110 // DispatchMany() will dispatch a set of tasks across worker threads.
 111 // Note: This function will block until all work has completed.
 112 void ThreadPool::DispatchMany(int num_tasks, WorkFunction work, void* data) {
 113   // On entry, all worker threads are sleeping.
 114   Setup(num_tasks, work, data);
 115
 116   // Wake up the worker threads & have them process tasks.
 117   for (int i = 0; i < num_threads_; i++)
 118     sem_post(&work_sem_);
 119
 120   // Worker threads are now awake and busy.
 121
 122   // This dispatch thread will now sleep-wait for the worker threads to finish.
 123   for (int i = 0; i < num_threads_; i++)
 124     sem_wait(&done_sem_);
 125   // On exit, all tasks are done and all worker threads are sleeping again.
 126 }
 127
 128 #ifdef __APPLE__
 129 #pragma clang diagnostic pop
 130 #endif
 131
 132 //  DispatchHere will dispatch all tasks on this thread.
 133 void ThreadPool::DispatchHere(int num_tasks, WorkFunction work, void* data) {
 134   for (int i = 0; i < num_tasks; i++)
 135     work(i, data);
 136 }
 137
 138 // Dispatch() will invoke the user supplied work function across
 139 // one or more threads for each task.
 140 // Note: This function will block until all work has completed.
 141 void ThreadPool::Dispatch(int num_tasks, WorkFunction work, void* data) {
 142   if (num_threads_ > 0)
 143     DispatchMany(num_tasks, work, data);
 144   else
 145     DispatchHere(num_tasks, work, data);
 146 }
 147
 148 }  // namespace sdk_util
 149