vpx_thread.h 12 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370
  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. #ifndef VPX_THREAD_H_
  16. #define VPX_THREAD_H_
  17. #include "./vpx_config.h"
  18. #ifdef __cplusplus
  19. extern "C" {
  20. #endif
  21. // Set maximum decode threads to be 8 due to the limit of frame buffers
  22. // and not enough semaphores in the emulation layer on windows.
  23. #define MAX_DECODE_THREADS 8
  24. #if CONFIG_MULTITHREAD
  25. #if defined(_WIN32) && !HAVE_PTHREAD_H
  26. #include <errno.h> // NOLINT
  27. #include <process.h> // NOLINT
  28. #include <windows.h> // NOLINT
  29. typedef HANDLE pthread_t;
  30. typedef CRITICAL_SECTION pthread_mutex_t;
  31. typedef struct {
  32. HANDLE waiting_sem_;
  33. HANDLE received_sem_;
  34. HANDLE signal_event_;
  35. } pthread_cond_t;
  36. //------------------------------------------------------------------------------
  37. // simplistic pthread emulation layer
  38. // _beginthreadex requires __stdcall
  39. #define THREADFN unsigned int __stdcall
  40. #define THREAD_RETURN(val) (unsigned int)((DWORD_PTR)val)
  41. static INLINE int pthread_create(pthread_t* const thread, const void* attr,
  42. unsigned int (__stdcall *start)(void*),
  43. void* arg) {
  44. (void)attr;
  45. *thread = (pthread_t)_beginthreadex(NULL, /* void *security */
  46. 0, /* unsigned stack_size */
  47. start,
  48. arg,
  49. 0, /* unsigned initflag */
  50. NULL); /* unsigned *thrdaddr */
  51. if (*thread == NULL) return 1;
  52. SetThreadPriority(*thread, THREAD_PRIORITY_ABOVE_NORMAL);
  53. return 0;
  54. }
  55. static INLINE int pthread_join(pthread_t thread, void** value_ptr) {
  56. (void)value_ptr;
  57. return (WaitForSingleObject(thread, INFINITE) != WAIT_OBJECT_0 ||
  58. CloseHandle(thread) == 0);
  59. }
  60. // Mutex
  61. static INLINE int pthread_mutex_init(pthread_mutex_t *const mutex,
  62. void* mutexattr) {
  63. (void)mutexattr;
  64. InitializeCriticalSection(mutex);
  65. return 0;
  66. }
  67. static INLINE int pthread_mutex_trylock(pthread_mutex_t *const mutex) {
  68. return TryEnterCriticalSection(mutex) ? 0 : EBUSY;
  69. }
  70. static INLINE int pthread_mutex_lock(pthread_mutex_t *const mutex) {
  71. EnterCriticalSection(mutex);
  72. return 0;
  73. }
  74. static INLINE int pthread_mutex_unlock(pthread_mutex_t *const mutex) {
  75. LeaveCriticalSection(mutex);
  76. return 0;
  77. }
  78. static INLINE int pthread_mutex_destroy(pthread_mutex_t *const mutex) {
  79. DeleteCriticalSection(mutex);
  80. return 0;
  81. }
  82. // Condition
  83. static INLINE int pthread_cond_destroy(pthread_cond_t *const condition) {
  84. int ok = 1;
  85. ok &= (CloseHandle(condition->waiting_sem_) != 0);
  86. ok &= (CloseHandle(condition->received_sem_) != 0);
  87. ok &= (CloseHandle(condition->signal_event_) != 0);
  88. return !ok;
  89. }
  90. static INLINE int pthread_cond_init(pthread_cond_t *const condition,
  91. void* cond_attr) {
  92. (void)cond_attr;
  93. condition->waiting_sem_ = CreateSemaphore(NULL, 0, MAX_DECODE_THREADS, NULL);
  94. condition->received_sem_ = CreateSemaphore(NULL, 0, MAX_DECODE_THREADS, NULL);
  95. condition->signal_event_ = CreateEvent(NULL, FALSE, FALSE, NULL);
  96. if (condition->waiting_sem_ == NULL ||
  97. condition->received_sem_ == NULL ||
  98. condition->signal_event_ == NULL) {
  99. pthread_cond_destroy(condition);
  100. return 1;
  101. }
  102. return 0;
  103. }
  104. static INLINE int pthread_cond_signal(pthread_cond_t *const condition) {
  105. int ok = 1;
  106. if (WaitForSingleObject(condition->waiting_sem_, 0) == WAIT_OBJECT_0) {
  107. // a thread is waiting in pthread_cond_wait: allow it to be notified
  108. ok = SetEvent(condition->signal_event_);
  109. // wait until the event is consumed so the signaler cannot consume
  110. // the event via its own pthread_cond_wait.
  111. ok &= (WaitForSingleObject(condition->received_sem_, INFINITE) !=
  112. WAIT_OBJECT_0);
  113. }
  114. return !ok;
  115. }
  116. static INLINE int pthread_cond_wait(pthread_cond_t *const condition,
  117. pthread_mutex_t *const mutex) {
  118. int ok;
  119. // note that there is a consumer available so the signal isn't dropped in
  120. // pthread_cond_signal
  121. if (!ReleaseSemaphore(condition->waiting_sem_, 1, NULL))
  122. return 1;
  123. // now unlock the mutex so pthread_cond_signal may be issued
  124. pthread_mutex_unlock(mutex);
  125. ok = (WaitForSingleObject(condition->signal_event_, INFINITE) ==
  126. WAIT_OBJECT_0);
  127. ok &= ReleaseSemaphore(condition->received_sem_, 1, NULL);
  128. pthread_mutex_lock(mutex);
  129. return !ok;
  130. }
  131. #elif defined(__OS2__)
  132. #define INCL_DOS
  133. #include <os2.h> // NOLINT
  134. #include <errno.h> // NOLINT
  135. #include <stdlib.h> // NOLINT
  136. #include <sys/builtin.h> // NOLINT
  137. #define pthread_t TID
  138. #define pthread_mutex_t HMTX
  139. typedef struct {
  140. HEV event_sem_;
  141. HEV ack_sem_;
  142. volatile unsigned wait_count_;
  143. } pthread_cond_t;
  144. //------------------------------------------------------------------------------
  145. // simplistic pthread emulation layer
  146. #define THREADFN void *
  147. #define THREAD_RETURN(val) (val)
  148. typedef struct {
  149. void* (*start_)(void*);
  150. void* arg_;
  151. } thread_arg;
  152. static void thread_start(void* arg) {
  153. thread_arg targ = *(thread_arg *)arg;
  154. free(arg);
  155. targ.start_(targ.arg_);
  156. }
  157. static INLINE int pthread_create(pthread_t* const thread, const void* attr,
  158. void* (*start)(void*),
  159. void* arg) {
  160. int tid;
  161. thread_arg *targ = (thread_arg *)malloc(sizeof(*targ));
  162. if (targ == NULL) return 1;
  163. (void)attr;
  164. targ->start_ = start;
  165. targ->arg_ = arg;
  166. tid = (pthread_t)_beginthread(thread_start, NULL, 1024 * 1024, targ);
  167. if (tid == -1) {
  168. free(targ);
  169. return 1;
  170. }
  171. *thread = tid;
  172. return 0;
  173. }
  174. static INLINE int pthread_join(pthread_t thread, void** value_ptr) {
  175. (void)value_ptr;
  176. return DosWaitThread(&thread, DCWW_WAIT) != 0;
  177. }
  178. // Mutex
  179. static INLINE int pthread_mutex_init(pthread_mutex_t *const mutex,
  180. void* mutexattr) {
  181. (void)mutexattr;
  182. return DosCreateMutexSem(NULL, mutex, 0, FALSE) != 0;
  183. }
  184. static INLINE int pthread_mutex_trylock(pthread_mutex_t *const mutex) {
  185. return DosRequestMutexSem(*mutex, SEM_IMMEDIATE_RETURN) == 0 ? 0 : EBUSY;
  186. }
  187. static INLINE int pthread_mutex_lock(pthread_mutex_t *const mutex) {
  188. return DosRequestMutexSem(*mutex, SEM_INDEFINITE_WAIT) != 0;
  189. }
  190. static INLINE int pthread_mutex_unlock(pthread_mutex_t *const mutex) {
  191. return DosReleaseMutexSem(*mutex) != 0;
  192. }
  193. static INLINE int pthread_mutex_destroy(pthread_mutex_t *const mutex) {
  194. return DosCloseMutexSem(*mutex) != 0;
  195. }
  196. // Condition
  197. static INLINE int pthread_cond_destroy(pthread_cond_t *const condition) {
  198. int ok = 1;
  199. ok &= DosCloseEventSem(condition->event_sem_) == 0;
  200. ok &= DosCloseEventSem(condition->ack_sem_) == 0;
  201. return !ok;
  202. }
  203. static INLINE int pthread_cond_init(pthread_cond_t *const condition,
  204. void* cond_attr) {
  205. int ok = 1;
  206. (void)cond_attr;
  207. ok &= DosCreateEventSem(NULL, &condition->event_sem_, DCE_POSTONE, FALSE)
  208. == 0;
  209. ok &= DosCreateEventSem(NULL, &condition->ack_sem_, DCE_POSTONE, FALSE) == 0;
  210. if (!ok) {
  211. pthread_cond_destroy(condition);
  212. return 1;
  213. }
  214. condition->wait_count_ = 0;
  215. return 0;
  216. }
  217. static INLINE int pthread_cond_signal(pthread_cond_t *const condition) {
  218. int ok = 1;
  219. if (!__atomic_cmpxchg32(&condition->wait_count_, 0, 0)) {
  220. ok &= DosPostEventSem(condition->event_sem_) == 0;
  221. ok &= DosWaitEventSem(condition->ack_sem_, SEM_INDEFINITE_WAIT) == 0;
  222. }
  223. return !ok;
  224. }
  225. static INLINE int pthread_cond_broadcast(pthread_cond_t *const condition) {
  226. int ok = 1;
  227. while (!__atomic_cmpxchg32(&condition->wait_count_, 0, 0))
  228. ok &= pthread_cond_signal(condition) == 0;
  229. return !ok;
  230. }
  231. static INLINE int pthread_cond_wait(pthread_cond_t *const condition,
  232. pthread_mutex_t *const mutex) {
  233. int ok = 1;
  234. __atomic_increment(&condition->wait_count_);
  235. ok &= pthread_mutex_unlock(mutex) == 0;
  236. ok &= DosWaitEventSem(condition->event_sem_, SEM_INDEFINITE_WAIT) == 0;
  237. __atomic_decrement(&condition->wait_count_);
  238. ok &= DosPostEventSem(condition->ack_sem_) == 0;
  239. pthread_mutex_lock(mutex);
  240. return !ok;
  241. }
  242. #else // _WIN32
  243. #include <pthread.h> // NOLINT
  244. # define THREADFN void*
  245. # define THREAD_RETURN(val) val
  246. #endif
  247. #endif // CONFIG_MULTITHREAD
  248. // State of the worker thread object
  249. typedef enum {
  250. NOT_OK = 0, // object is unusable
  251. OK, // ready to work
  252. WORK // busy finishing the current task
  253. } VPxWorkerStatus;
  254. // Function to be called by the worker thread. Takes two opaque pointers as
  255. // arguments (data1 and data2), and should return false in case of error.
  256. typedef int (*VPxWorkerHook)(void*, void*);
  257. // Platform-dependent implementation details for the worker.
  258. typedef struct VPxWorkerImpl VPxWorkerImpl;
  259. // Synchronization object used to launch job in the worker thread
  260. typedef struct {
  261. VPxWorkerImpl *impl_;
  262. VPxWorkerStatus status_;
  263. VPxWorkerHook hook; // hook to call
  264. void *data1; // first argument passed to 'hook'
  265. void *data2; // second argument passed to 'hook'
  266. int had_error; // return value of the last call to 'hook'
  267. } VPxWorker;
  268. // The interface for all thread-worker related functions. All these functions
  269. // must be implemented.
  270. typedef struct {
  271. // Must be called first, before any other method.
  272. void (*init)(VPxWorker *const worker);
  273. // Must be called to initialize the object and spawn the thread. Re-entrant.
  274. // Will potentially launch the thread. Returns false in case of error.
  275. int (*reset)(VPxWorker *const worker);
  276. // Makes sure the previous work is finished. Returns true if worker->had_error
  277. // was not set and no error condition was triggered by the working thread.
  278. int (*sync)(VPxWorker *const worker);
  279. // Triggers the thread to call hook() with data1 and data2 arguments. These
  280. // hook/data1/data2 values can be changed at any time before calling this
  281. // function, but not be changed afterward until the next call to Sync().
  282. void (*launch)(VPxWorker *const worker);
  283. // This function is similar to launch() except that it calls the
  284. // hook directly instead of using a thread. Convenient to bypass the thread
  285. // mechanism while still using the VPxWorker structs. sync() must
  286. // still be called afterward (for error reporting).
  287. void (*execute)(VPxWorker *const worker);
  288. // Kill the thread and terminate the object. To use the object again, one
  289. // must call reset() again.
  290. void (*end)(VPxWorker *const worker);
  291. } VPxWorkerInterface;
  292. // Install a new set of threading functions, overriding the defaults. This
  293. // should be done before any workers are started, i.e., before any encoding or
  294. // decoding takes place. The contents of the interface struct are copied, it
  295. // is safe to free the corresponding memory after this call. This function is
  296. // not thread-safe. Return false in case of invalid pointer or methods.
  297. int vpx_set_worker_interface(const VPxWorkerInterface *const winterface);
  298. // Retrieve the currently set thread worker interface.
  299. const VPxWorkerInterface *vpx_get_worker_interface(void);
  300. //------------------------------------------------------------------------------
  301. #ifdef __cplusplus
  302. } // extern "C"
  303. #endif
  304. #endif // VPX_THREAD_H_