1// Copyright 2011 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// Author: Skal (pascal.massimino@gmail.com)
13
14#include <assert.h>
15#include <string.h>   // for memset()
16#include "./thread.h"
17
18#if defined(__cplusplus) || defined(c_plusplus)
19extern "C" {
20#endif
21
22#ifdef WEBP_USE_THREAD
23
24#if defined(_WIN32)
25
26//------------------------------------------------------------------------------
27// simplistic pthread emulation layer
28
29#include <process.h>
30
31// _beginthreadex requires __stdcall
32#define THREADFN unsigned int __stdcall
33#define THREAD_RETURN(val) (unsigned int)((DWORD_PTR)val)
34
35static int pthread_create(pthread_t* const thread, const void* attr,
36                          unsigned int (__stdcall *start)(void*), void* arg) {
37  (void)attr;
38  *thread = (pthread_t)_beginthreadex(NULL,   /* void *security */
39                                      0,      /* unsigned stack_size */
40                                      start,
41                                      arg,
42                                      0,      /* unsigned initflag */
43                                      NULL);  /* unsigned *thrdaddr */
44  if (*thread == NULL) return 1;
45  SetThreadPriority(*thread, THREAD_PRIORITY_ABOVE_NORMAL);
46  return 0;
47}
48
49static int pthread_join(pthread_t thread, void** value_ptr) {
50  (void)value_ptr;
51  return (WaitForSingleObject(thread, INFINITE) != WAIT_OBJECT_0 ||
52          CloseHandle(thread) == 0);
53}
54
55// Mutex
56static int pthread_mutex_init(pthread_mutex_t* const mutex, void* mutexattr) {
57  (void)mutexattr;
58  InitializeCriticalSection(mutex);
59  return 0;
60}
61
62static int pthread_mutex_lock(pthread_mutex_t* const mutex) {
63  EnterCriticalSection(mutex);
64  return 0;
65}
66
67static int pthread_mutex_unlock(pthread_mutex_t* const mutex) {
68  LeaveCriticalSection(mutex);
69  return 0;
70}
71
72static int pthread_mutex_destroy(pthread_mutex_t* const mutex) {
73  DeleteCriticalSection(mutex);
74  return 0;
75}
76
77// Condition
78static int pthread_cond_destroy(pthread_cond_t* const condition) {
79  int ok = 1;
80  ok &= (CloseHandle(condition->waiting_sem_) != 0);
81  ok &= (CloseHandle(condition->received_sem_) != 0);
82  ok &= (CloseHandle(condition->signal_event_) != 0);
83  return !ok;
84}
85
86static int pthread_cond_init(pthread_cond_t* const condition, void* cond_attr) {
87  (void)cond_attr;
88  condition->waiting_sem_ = CreateSemaphore(NULL, 0, 1, NULL);
89  condition->received_sem_ = CreateSemaphore(NULL, 0, 1, NULL);
90  condition->signal_event_ = CreateEvent(NULL, FALSE, FALSE, NULL);
91  if (condition->waiting_sem_ == NULL ||
92      condition->received_sem_ == NULL ||
93      condition->signal_event_ == NULL) {
94    pthread_cond_destroy(condition);
95    return 1;
96  }
97  return 0;
98}
99
100static int pthread_cond_signal(pthread_cond_t* const condition) {
101  int ok = 1;
102  if (WaitForSingleObject(condition->waiting_sem_, 0) == WAIT_OBJECT_0) {
103    // a thread is waiting in pthread_cond_wait: allow it to be notified
104    ok = SetEvent(condition->signal_event_);
105    // wait until the event is consumed so the signaler cannot consume
106    // the event via its own pthread_cond_wait.
107    ok &= (WaitForSingleObject(condition->received_sem_, INFINITE) !=
108           WAIT_OBJECT_0);
109  }
110  return !ok;
111}
112
113static int pthread_cond_wait(pthread_cond_t* const condition,
114                             pthread_mutex_t* const mutex) {
115  int ok;
116  // note that there is a consumer available so the signal isn't dropped in
117  // pthread_cond_signal
118  if (!ReleaseSemaphore(condition->waiting_sem_, 1, NULL))
119    return 1;
120  // now unlock the mutex so pthread_cond_signal may be issued
121  pthread_mutex_unlock(mutex);
122  ok = (WaitForSingleObject(condition->signal_event_, INFINITE) ==
123        WAIT_OBJECT_0);
124  ok &= ReleaseSemaphore(condition->received_sem_, 1, NULL);
125  pthread_mutex_lock(mutex);
126  return !ok;
127}
128
129#else  // _WIN32
130# define THREADFN void*
131# define THREAD_RETURN(val) val
132#endif
133
134//------------------------------------------------------------------------------
135
136static THREADFN WebPWorkerThreadLoop(void *ptr) {    // thread loop
137  WebPWorker* const worker = (WebPWorker*)ptr;
138  int done = 0;
139  while (!done) {
140    pthread_mutex_lock(&worker->mutex_);
141    while (worker->status_ == OK) {   // wait in idling mode
142      pthread_cond_wait(&worker->condition_, &worker->mutex_);
143    }
144    if (worker->status_ == WORK) {
145      if (worker->hook) {
146        worker->had_error |= !worker->hook(worker->data1, worker->data2);
147      }
148      worker->status_ = OK;
149    } else if (worker->status_ == NOT_OK) {   // finish the worker
150      done = 1;
151    }
152    // signal to the main thread that we're done (for Sync())
153    pthread_cond_signal(&worker->condition_);
154    pthread_mutex_unlock(&worker->mutex_);
155  }
156  return THREAD_RETURN(NULL);    // Thread is finished
157}
158
159// main thread state control
160static void WebPWorkerChangeState(WebPWorker* const worker,
161                                  WebPWorkerStatus new_status) {
162  // no-op when attempting to change state on a thread that didn't come up
163  if (worker->status_ < OK) return;
164
165  pthread_mutex_lock(&worker->mutex_);
166  // wait for the worker to finish
167  while (worker->status_ != OK) {
168    pthread_cond_wait(&worker->condition_, &worker->mutex_);
169  }
170  // assign new status and release the working thread if needed
171  if (new_status != OK) {
172    worker->status_ = new_status;
173    pthread_cond_signal(&worker->condition_);
174  }
175  pthread_mutex_unlock(&worker->mutex_);
176}
177
178#endif
179
180//------------------------------------------------------------------------------
181
182void WebPWorkerInit(WebPWorker* const worker) {
183  memset(worker, 0, sizeof(*worker));
184  worker->status_ = NOT_OK;
185}
186
187int WebPWorkerSync(WebPWorker* const worker) {
188#ifdef WEBP_USE_THREAD
189  WebPWorkerChangeState(worker, OK);
190#endif
191  assert(worker->status_ <= OK);
192  return !worker->had_error;
193}
194
195int WebPWorkerReset(WebPWorker* const worker) {
196  int ok = 1;
197  worker->had_error = 0;
198  if (worker->status_ < OK) {
199#ifdef WEBP_USE_THREAD
200    if (pthread_mutex_init(&worker->mutex_, NULL) ||
201        pthread_cond_init(&worker->condition_, NULL)) {
202      return 0;
203    }
204    pthread_mutex_lock(&worker->mutex_);
205    ok = !pthread_create(&worker->thread_, NULL, WebPWorkerThreadLoop, worker);
206    if (ok) worker->status_ = OK;
207    pthread_mutex_unlock(&worker->mutex_);
208#else
209    worker->status_ = OK;
210#endif
211  } else if (worker->status_ > OK) {
212    ok = WebPWorkerSync(worker);
213  }
214  assert(!ok || (worker->status_ == OK));
215  return ok;
216}
217
218void WebPWorkerLaunch(WebPWorker* const worker) {
219#ifdef WEBP_USE_THREAD
220  WebPWorkerChangeState(worker, WORK);
221#else
222  if (worker->hook)
223    worker->had_error |= !worker->hook(worker->data1, worker->data2);
224#endif
225}
226
227void WebPWorkerEnd(WebPWorker* const worker) {
228  if (worker->status_ >= OK) {
229#ifdef WEBP_USE_THREAD
230    WebPWorkerChangeState(worker, NOT_OK);
231    pthread_join(worker->thread_, NULL);
232    pthread_mutex_destroy(&worker->mutex_);
233    pthread_cond_destroy(&worker->condition_);
234#else
235    worker->status_ = NOT_OK;
236#endif
237  }
238  assert(worker->status_ == NOT_OK);
239}
240
241//------------------------------------------------------------------------------
242
243#if defined(__cplusplus) || defined(c_plusplus)
244}    // extern "C"
245#endif
246