10 // Multi-threaded worker
38 static void execute(VPxWorker *const worker);  // Forward declaration.
41   VPxWorker *const worker = (VPxWorker *)ptr;  in thread_loop()  local
43   if (worker->thread_name != NULL) {  in thread_loop()
50     strncpy(thread_name, worker->thread_name, sizeof(thread_name) - 1);  in thread_loop()
55   if (worker->thread_name != NULL) {  in thread_loop()
59     strncpy(thread_name, worker->thread_name, sizeof(thread_name) - 1);  in thread_loop()
64   pthread_mutex_lock(&worker->impl_->mutex_);  in thread_loop()
66     while (worker->status_ == VPX_WORKER_STATUS_OK) {  // wait in idling mode  in thread_loop()
67       pthread_cond_wait(&worker->impl_->condition_, &worker->impl_->mutex_);  in thread_loop()
69     if (worker->status_ == VPX_WORKER_STATUS_WORKING) {  in thread_loop()
70       // When worker->status_ is VPX_WORKER_STATUS_WORKING, the main thread  in thread_loop()
71       // doesn't change worker->status_ and will wait until the worker changes  in thread_loop()
72       // worker->status_ to VPX_WORKER_STATUS_OK. See change_state(). So the  in thread_loop()
73       // worker can safely call execute() without holding worker->impl_->mutex_.  in thread_loop()
74       // When the worker reacquires worker->impl_->mutex_, worker->status_ must  in thread_loop()
76       pthread_mutex_unlock(&worker->impl_->mutex_);  in thread_loop()
77       execute(worker);  in thread_loop()
78       pthread_mutex_lock(&worker->impl_->mutex_);  in thread_loop()
79       assert(worker->status_ == VPX_WORKER_STATUS_WORKING);  in thread_loop()
80       worker->status_ = VPX_WORKER_STATUS_OK;  in thread_loop()
82       pthread_cond_signal(&worker->impl_->condition_);  in thread_loop()
84       assert(worker->status_ == VPX_WORKER_STATUS_NOT_OK);  // finish the worker  in thread_loop()
88   pthread_mutex_unlock(&worker->impl_->mutex_);  in thread_loop()
93 static void change_state(VPxWorker *const worker, VPxWorkerStatus new_status) {  in change_state()  argument
97   if (worker->impl_ == NULL) return;  in change_state()
99   pthread_mutex_lock(&worker->impl_->mutex_);  in change_state()
100   if (worker->status_ >= VPX_WORKER_STATUS_OK) {  in change_state()
101     // wait for the worker to finish  in change_state()
102     while (worker->status_ != VPX_WORKER_STATUS_OK) {  in change_state()
103       pthread_cond_wait(&worker->impl_->condition_, &worker->impl_->mutex_);  in change_state()
107       worker->status_ = new_status;  in change_state()
108       pthread_cond_signal(&worker->impl_->condition_);  in change_state()
111   pthread_mutex_unlock(&worker->impl_->mutex_);  in change_state()
118 static void init(VPxWorker *const worker) {  in init()  argument
119   memset(worker, 0, sizeof(*worker));  in init()
120   worker->status_ = VPX_WORKER_STATUS_NOT_OK;  in init()
123 static int sync(VPxWorker *const worker) {  in sync()  argument
125   change_state(worker, VPX_WORKER_STATUS_OK);  in sync()
127   assert(worker->status_ <= VPX_WORKER_STATUS_OK);  in sync()
128   return !worker->had_error;  in sync()
131 static int reset(VPxWorker *const worker) {  in reset()  argument
133   worker->had_error = 0;  in reset()
134   if (worker->status_ < VPX_WORKER_STATUS_OK) {  in reset()
136     worker->impl_ = (VPxWorkerImpl *)vpx_calloc(1, sizeof(*worker->impl_));  in reset()
137     if (worker->impl_ == NULL) {  in reset()
140     if (pthread_mutex_init(&worker->impl_->mutex_, NULL)) {  in reset()
143     if (pthread_cond_init(&worker->impl_->condition_, NULL)) {  in reset()
144       pthread_mutex_destroy(&worker->impl_->mutex_);  in reset()
147     pthread_mutex_lock(&worker->impl_->mutex_);  in reset()
148     ok = !pthread_create(&worker->impl_->thread_, NULL, thread_loop, worker);  in reset()
149     if (ok) worker->status_ = VPX_WORKER_STATUS_OK;  in reset()
150     pthread_mutex_unlock(&worker->impl_->mutex_);  in reset()
152       pthread_mutex_destroy(&worker->impl_->mutex_);  in reset()
153       pthread_cond_destroy(&worker->impl_->condition_);  in reset()
155       vpx_free(worker->impl_);  in reset()
156       worker->impl_ = NULL;  in reset()
160     worker->status_ = VPX_WORKER_STATUS_OK;  in reset()
162   } else if (worker->status_ > VPX_WORKER_STATUS_OK) {  in reset()
163     ok = sync(worker);  in reset()
165   assert(!ok || (worker->status_ == VPX_WORKER_STATUS_OK));  in reset()
169 static void execute(VPxWorker *const worker) {  in execute()  argument
170   if (worker->hook != NULL) {  in execute()
171     worker->had_error |= !worker->hook(worker->data1, worker->data2);  in execute()
175 static void launch(VPxWorker *const worker) {  in launch()  argument
177   change_state(worker, VPX_WORKER_STATUS_WORKING);  in launch()
179   execute(worker);  in launch()
183 static void end(VPxWorker *const worker) {  in end()  argument
185   if (worker->impl_ != NULL) {  in end()
186     change_state(worker, VPX_WORKER_STATUS_NOT_OK);  in end()
187     pthread_join(worker->impl_->thread_, NULL);  in end()
188     pthread_mutex_destroy(&worker->impl_->mutex_);  in end()
189     pthread_cond_destroy(&worker->impl_->condition_);  in end()
190     vpx_free(worker->impl_);  in end()
191     worker->impl_ = NULL;  in end()
194   worker->status_ = VPX_WORKER_STATUS_NOT_OK;  in end()
195   assert(worker->impl_ == NULL);  in end()
197   assert(worker->status_ == VPX_WORKER_STATUS_NOT_OK);  in end()