vpx_util/vpx_thread.h

   1 // Copyright 2013 Google Inc. All Rights Reserved.
   2 //
   3 // Use of this source code is governed by a BSD-style license
   4 // that can be found in the COPYING file in the root of the source
   5 // tree. An additional intellectual property rights grant can be found
   6 // in the file PATENTS. All contributing project authors may
   7 // be found in the AUTHORS file in the root of the source tree.
   8 // -----------------------------------------------------------------------------
   9 //
  10 // Multi-threaded worker
  11 //
  12 // Original source:
  13 //  http://git.chromium.org/webm/libwebp.git
  14 //  100644 blob 7bd451b124ae3b81596abfbcc823e3cb129d3a38  src/utils/thread.h
  15
  16 #ifndef VPX_THREAD_H_
  17 #define VPX_THREAD_H_
  18
  19 #include "./vpx_config.h"
  20
  21 #ifdef __cplusplus
  22 extern "C" {
  23 #endif
  24
  25 // Set maximum decode threads to be 8 due to the limit of frame buffers
  26 // and not enough semaphores in the emulation layer on windows.
  27 #define MAX_DECODE_THREADS 8
  28
  29 #if CONFIG_MULTITHREAD
  30
  31 #if defined(_WIN32) && !HAVE_PTHREAD_H
  32 #include <errno.h>  // NOLINT
  33 #include <process.h>  // NOLINT
  34 #include <windows.h>  // NOLINT
  35 typedef HANDLE pthread_t;
  36 typedef CRITICAL_SECTION pthread_mutex_t;
  37 typedef struct {
  38   HANDLE waiting_sem_;
  39   HANDLE received_sem_;
  40   HANDLE signal_event_;
  41 } pthread_cond_t;
  42
  43 //------------------------------------------------------------------------------
  44 // simplistic pthread emulation layer
  45
  46 // _beginthreadex requires __stdcall
  47 #define THREADFN unsigned int __stdcall
  48 #define THREAD_RETURN(val) (unsigned int)((DWORD_PTR)val)
  49
  50 static INLINE int pthread_create(pthread_t* const thread, const void* attr,
  51                                  unsigned int (__stdcall *start)(void*),
  52                                  void* arg) {
  53   (void)attr;
  54   *thread = (pthread_t)_beginthreadex(NULL,   /* void *security */
  55                                       0,      /* unsigned stack_size */
  56                                       start,
  57                                       arg,
  58                                       0,      /* unsigned initflag */
  59                                       NULL);  /* unsigned *thrdaddr */
  60   if (*thread == NULL) return 1;
  61   SetThreadPriority(*thread, THREAD_PRIORITY_ABOVE_NORMAL);
  62   return 0;
  63 }
  64
  65 static INLINE int pthread_join(pthread_t thread, void** value_ptr) {
  66   (void)value_ptr;
  67   return (WaitForSingleObject(thread, INFINITE) != WAIT_OBJECT_0 ||
  68           CloseHandle(thread) == 0);
  69 }
  70
  71 // Mutex
  72 static INLINE int pthread_mutex_init(pthread_mutex_t *const mutex,
  73                                      void* mutexattr) {
  74   (void)mutexattr;
  75   InitializeCriticalSection(mutex);
  76   return 0;
  77 }
  78
  79 static INLINE int pthread_mutex_trylock(pthread_mutex_t *const mutex) {
  80   return TryEnterCriticalSection(mutex) ? 0 : EBUSY;
  81 }
  82
  83 static INLINE int pthread_mutex_lock(pthread_mutex_t *const mutex) {
  84   EnterCriticalSection(mutex);
  85   return 0;
  86 }
  87
  88 static INLINE int pthread_mutex_unlock(pthread_mutex_t *const mutex) {
  89   LeaveCriticalSection(mutex);
  90   return 0;
  91 }
  92
  93 static INLINE int pthread_mutex_destroy(pthread_mutex_t *const mutex) {
  94   DeleteCriticalSection(mutex);
  95   return 0;
  96 }
  97
  98 // Condition
  99 static INLINE int pthread_cond_destroy(pthread_cond_t *const condition) {
 100   int ok = 1;
 101   ok &= (CloseHandle(condition->waiting_sem_) != 0);
 102   ok &= (CloseHandle(condition->received_sem_) != 0);
 103   ok &= (CloseHandle(condition->signal_event_) != 0);
 104   return !ok;
 105 }
 106
 107 static INLINE int pthread_cond_init(pthread_cond_t *const condition,
 108                                     void* cond_attr) {
 109   (void)cond_attr;
 110   condition->waiting_sem_ = CreateSemaphore(NULL, 0, MAX_DECODE_THREADS, NULL);
 111   condition->received_sem_ = CreateSemaphore(NULL, 0, MAX_DECODE_THREADS, NULL);
 112   condition->signal_event_ = CreateEvent(NULL, FALSE, FALSE, NULL);
 113   if (condition->waiting_sem_ == NULL ||
 114       condition->received_sem_ == NULL ||
 115       condition->signal_event_ == NULL) {
 116     pthread_cond_destroy(condition);
 117     return 1;
 118   }
 119   return 0;
 120 }
 121
 122 static INLINE int pthread_cond_signal(pthread_cond_t *const condition) {
 123   int ok = 1;
 124   if (WaitForSingleObject(condition->waiting_sem_, 0) == WAIT_OBJECT_0) {
 125     // a thread is waiting in pthread_cond_wait: allow it to be notified
 126     ok = SetEvent(condition->signal_event_);
 127     // wait until the event is consumed so the signaler cannot consume
 128     // the event via its own pthread_cond_wait.
 129     ok &= (WaitForSingleObject(condition->received_sem_, INFINITE) !=
 130            WAIT_OBJECT_0);
 131   }
 132   return !ok;
 133 }
 134
 135 static INLINE int pthread_cond_wait(pthread_cond_t *const condition,
 136                                     pthread_mutex_t *const mutex) {
 137   int ok;
 138   // note that there is a consumer available so the signal isn't dropped in
 139   // pthread_cond_signal
 140   if (!ReleaseSemaphore(condition->waiting_sem_, 1, NULL))
 141     return 1;
 142   // now unlock the mutex so pthread_cond_signal may be issued
 143   pthread_mutex_unlock(mutex);
 144   ok = (WaitForSingleObject(condition->signal_event_, INFINITE) ==
 145         WAIT_OBJECT_0);
 146   ok &= ReleaseSemaphore(condition->received_sem_, 1, NULL);
 147   pthread_mutex_lock(mutex);
 148   return !ok;
 149 }
 150 #else  // _WIN32
 151 #include <pthread.h> // NOLINT
 152 # define THREADFN void*
 153 # define THREAD_RETURN(val) val
 154 #endif
 155
 156 #endif  // CONFIG_MULTITHREAD
 157
 158 // State of the worker thread object
 159 typedef enum {
 160   NOT_OK = 0,   // object is unusable
 161   OK,           // ready to work
 162   WORK          // busy finishing the current task
 163 } VPxWorkerStatus;
 164
 165 // Function to be called by the worker thread. Takes two opaque pointers as
 166 // arguments (data1 and data2), and should return false in case of error.
 167 typedef int (*VPxWorkerHook)(void*, void*);
 168
 169 // Platform-dependent implementation details for the worker.
 170 typedef struct VPxWorkerImpl VPxWorkerImpl;
 171
 172 // Synchronization object used to launch job in the worker thread
 173 typedef struct {
 174   VPxWorkerImpl *impl_;
 175   VPxWorkerStatus status_;
 176   VPxWorkerHook hook;     // hook to call
 177   void *data1;            // first argument passed to 'hook'
 178   void *data2;            // second argument passed to 'hook'
 179   int had_error;          // return value of the last call to 'hook'
 180 } VPxWorker;
 181
 182 // The interface for all thread-worker related functions. All these functions
 183 // must be implemented.
 184 typedef struct {
 185   // Must be called first, before any other method.
 186   void (*init)(VPxWorker *const worker);
 187   // Must be called to initialize the object and spawn the thread. Re-entrant.
 188   // Will potentially launch the thread. Returns false in case of error.
 189   int (*reset)(VPxWorker *const worker);
 190   // Makes sure the previous work is finished. Returns true if worker->had_error
 191   // was not set and no error condition was triggered by the working thread.
 192   int (*sync)(VPxWorker *const worker);
 193   // Triggers the thread to call hook() with data1 and data2 arguments. These
 194   // hook/data1/data2 values can be changed at any time before calling this
 195   // function, but not be changed afterward until the next call to Sync().
 196   void (*launch)(VPxWorker *const worker);
 197   // This function is similar to launch() except that it calls the
 198   // hook directly instead of using a thread. Convenient to bypass the thread
 199   // mechanism while still using the VPxWorker structs. sync() must
 200   // still be called afterward (for error reporting).
 201   void (*execute)(VPxWorker *const worker);
 202   // Kill the thread and terminate the object. To use the object again, one
 203   // must call reset() again.
 204   void (*end)(VPxWorker *const worker);
 205 } VPxWorkerInterface;
 206
 207 // Install a new set of threading functions, overriding the defaults. This
 208 // should be done before any workers are started, i.e., before any encoding or
 209 // decoding takes place. The contents of the interface struct are copied, it
 210 // is safe to free the corresponding memory after this call. This function is
 211 // not thread-safe. Return false in case of invalid pointer or methods.
 212 int vpx_set_worker_interface(const VPxWorkerInterface *const winterface);
 213
 214 // Retrieve the currently set thread worker interface.
 215 const VPxWorkerInterface *vpx_get_worker_interface(void);
 216
 217 //------------------------------------------------------------------------------
 218
 219 #ifdef __cplusplus
 220 }    // extern "C"
 221 #endif
 222
 223 #endif  // VPX_THREAD_H_