src/misc/background_worker.c

   1 /*****************************************************************************
   2  * Copyright (C) 2017 VLC authors and VideoLAN
   3  *
   4  * This program is free software; you can redistribute it and/or modify it
   5  * under the terms of the GNU Lesser General Public License as published by
   6  * the Free Software Foundation; either version 2.1 of the License, or
   7  * (at your option) any later version.
   8  *
   9  * This program is distributed in the hope that it will be useful,
  10  * but WITHOUT ANY WARRANTY; without even the implied warranty of
  11  * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
  12  * GNU Lesser General Public License for more details.
  13  *
  14  * You should have received a copy of the GNU Lesser General Public License
  15  * along with this program; if not, write to the Free Software Foundation,
  16  * Inc., 51 Franklin Street, Fifth Floor, Boston MA 02110-1301, USA.
  17  *****************************************************************************/
  18
  19 #ifdef HAVE_CONFIG_H
  20 #  include "config.h"
  21 #endif
  22
  23 #include <assert.h>
  24 #include <vlc_common.h>
  25 #include <vlc_atomic.h>
  26 #include <vlc_list.h>
  27 #include <vlc_threads.h>
  28
  29 #include "libvlc.h"
  30 #include "background_worker.h"
  31
  32 struct task {
  33     struct vlc_list node;
  34     void* id; /**< id associated with entity */
  35     void* entity; /**< the entity to process */
  36     vlc_tick_t timeout; /**< timeout duration in vlc_tick_t */
  37 };
  38
  39 struct background_worker;
  40
  41 struct background_thread {
  42     struct background_worker *owner;
  43     vlc_cond_t probe_cancel_wait; /**< wait for probe request or cancelation */
  44     bool probe; /**< true if a probe is requested */
  45     bool cancel; /**< true if a cancel is requested */
  46     struct task *task; /**< current task */
  47     struct vlc_list node;
  48 };
  49
  50 struct background_worker {
  51     void* owner;
  52     struct background_worker_config conf;
  53
  54     vlc_mutex_t lock;
  55
  56     int uncompleted; /**< number of tasks requested but not completed */
  57     int nthreads; /**< number of threads in the threads list */
  58     struct vlc_list threads; /**< list of active background_thread instances */
  59
  60     struct vlc_list queue; /**< queue of tasks */
  61     vlc_cond_t queue_wait; /**< wait for the queue to be non-empty */
  62
  63     vlc_cond_t nothreads_wait; /**< wait for nthreads == 0 */
  64     bool closing; /**< true if background worker deletion is requested */
  65 };
  66
  67 static struct task *task_Create(struct background_worker *worker, void *id,
  68                                 void *entity, int timeout)
  69 {
  70     struct task *task = malloc(sizeof(*task));
  71     if (unlikely(!task))
  72         return NULL;
  73
  74     task->id = id;
  75     task->entity = entity;
  76     task->timeout = timeout < 0 ? worker->conf.default_timeout : VLC_TICK_FROM_MS(timeout);
  77     worker->conf.pf_hold(task->entity);
  78     return task;
  79 }
  80
  81 static void task_Destroy(struct background_worker *worker, struct task *task)
  82 {
  83     worker->conf.pf_release(task->entity);
  84     free(task);
  85 }
  86
  87 static struct task *QueueTake(struct background_worker *worker, int timeout_ms)
  88 {
  89     vlc_assert_locked(&worker->lock);
  90
  91     vlc_tick_t deadline = vlc_tick_now() + VLC_TICK_FROM_MS(timeout_ms);
  92     bool timeout = false;
  93     while (!timeout && !worker->closing && vlc_list_is_empty(&worker->queue))
  94         timeout = vlc_cond_timedwait(&worker->queue_wait,
  95                                      &worker->lock, deadline) != 0;
  96
  97     if (worker->closing || timeout)
  98         return NULL;
  99
 100     struct task *task = vlc_list_first_entry_or_null(&worker->queue,
 101                                                      struct task, node);
 102     assert(task);
 103     vlc_list_remove(&task->node);
 104
 105     return task;
 106 }
 107
 108 static void QueuePush(struct background_worker *worker, struct task *task)
 109 {
 110     vlc_assert_locked(&worker->lock);
 111     vlc_list_append(&task->node, &worker->queue);
 112     vlc_cond_signal(&worker->queue_wait);
 113 }
 114
 115 static void QueueRemoveAll(struct background_worker *worker, void *id)
 116 {
 117     vlc_assert_locked(&worker->lock);
 118     struct task *task;
 119     vlc_list_foreach(task, &worker->queue, node)
 120     {
 121         if (!id || task->id == id)
 122         {
 123             vlc_list_remove(&task->node);
 124             task_Destroy(worker, task);
 125         }
 126     }
 127 }
 128
 129 static struct background_thread *
 130 background_thread_Create(struct background_worker *owner)
 131 {
 132     struct background_thread *thread = malloc(sizeof(*thread));
 133     if (!thread)
 134         return NULL;
 135
 136     vlc_cond_init(&thread->probe_cancel_wait);
 137     thread->probe = false;
 138     thread->cancel = false;
 139     thread->task = NULL;
 140     thread->owner = owner;
 141     return thread;
 142 }
 143
 144 static void background_thread_Destroy(struct background_thread *thread)
 145 {
 146     vlc_cond_destroy(&thread->probe_cancel_wait);
 147     free(thread);
 148 }
 149
 150 static struct background_worker *background_worker_Create(void *owner,
 151                                          struct background_worker_config *conf)
 152 {
 153     struct background_worker* worker = malloc(sizeof(*worker));
 154     if (unlikely(!worker))
 155         return NULL;
 156
 157     worker->conf = *conf;
 158     worker->owner = owner;
 159
 160     vlc_mutex_init(&worker->lock);
 161     worker->uncompleted = 0;
 162     worker->nthreads = 0;
 163     vlc_list_init(&worker->threads);
 164     vlc_list_init(&worker->queue);
 165     vlc_cond_init(&worker->queue_wait);
 166     vlc_cond_init(&worker->nothreads_wait);
 167     worker->closing = false;
 168     return worker;
 169 }
 170
 171 static void background_worker_Destroy(struct background_worker *worker)
 172 {
 173     vlc_cond_destroy(&worker->queue_wait);
 174     vlc_mutex_destroy(&worker->lock);
 175     free(worker);
 176 }
 177
 178 static void TerminateTask(struct background_thread *thread, struct task *task)
 179 {
 180     struct background_worker *worker = thread->owner;
 181     task_Destroy(worker, task);
 182
 183     vlc_mutex_lock(&worker->lock);
 184     thread->task = NULL;
 185     worker->uncompleted--;
 186     assert(worker->uncompleted >= 0);
 187     vlc_mutex_unlock(&worker->lock);
 188 }
 189
 190 static void RemoveThread(struct background_thread *thread)
 191 {
 192     struct background_worker *worker = thread->owner;
 193
 194     vlc_mutex_lock(&worker->lock);
 195
 196     vlc_list_remove(&thread->node);
 197     worker->nthreads--;
 198     assert(worker->nthreads >= 0);
 199     if (!worker->nthreads)
 200         vlc_cond_signal(&worker->nothreads_wait);
 201
 202     vlc_mutex_unlock(&worker->lock);
 203
 204     background_thread_Destroy(thread);
 205 }
 206
 207 static void* Thread( void* data )
 208 {
 209     struct background_thread *thread = data;
 210     struct background_worker *worker = thread->owner;
 211
 212     for (;;)
 213     {
 214         vlc_mutex_lock(&worker->lock);
 215         struct task *task = QueueTake(worker, 5000);
 216         if (!task)
 217         {
 218             vlc_mutex_unlock(&worker->lock);
 219             /* terminate this thread */
 220             break;
 221         }
 222
 223         thread->task = task;
 224         thread->cancel = false;
 225         thread->probe = false;
 226         vlc_tick_t deadline;
 227         if (task->timeout > 0)
 228             deadline = vlc_tick_now() + task->timeout;
 229         else
 230             deadline = INT64_MAX; /* no deadline */
 231         vlc_mutex_unlock(&worker->lock);
 232
 233         void *handle;
 234         if (worker->conf.pf_start(worker->owner, task->entity, &handle))
 235         {
 236             TerminateTask(thread, task);
 237             continue;
 238         }
 239
 240         for (;;)
 241         {
 242             vlc_mutex_lock(&worker->lock);
 243             bool timeout = false;
 244             while (!timeout && !thread->probe && !thread->cancel)
 245                 /* any non-zero return value means timeout */
 246                 timeout = vlc_cond_timedwait(&thread->probe_cancel_wait,
 247                                              &worker->lock, deadline) != 0;
 248
 249             bool cancel = thread->cancel;
 250             thread->cancel = false;
 251             thread->probe = false;
 252             vlc_mutex_unlock(&worker->lock);
 253
 254             if (timeout || cancel
 255                     || worker->conf.pf_probe(worker->owner, handle))
 256             {
 257                 worker->conf.pf_stop(worker->owner, handle);
 258                 TerminateTask(thread, task);
 259                 break;
 260             }
 261         }
 262     }
 263
 264     RemoveThread(thread);
 265
 266     return NULL;
 267 }
 268
 269 static bool SpawnThread(struct background_worker *worker)
 270 {
 271     vlc_assert_locked(&worker->lock);
 272
 273     struct background_thread *thread = background_thread_Create(worker);
 274     if (!thread)
 275         return false;
 276
 277     if (vlc_clone_detach(NULL, Thread, thread, VLC_THREAD_PRIORITY_LOW))
 278     {
 279         free(thread);
 280         return false;
 281     }
 282     worker->nthreads++;
 283     vlc_list_append(&thread->node, &worker->threads);
 284
 285     return true;
 286 }
 287
 288 struct background_worker* background_worker_New( void* owner,
 289     struct background_worker_config* conf )
 290 {
 291     return background_worker_Create(owner, conf);
 292 }
 293
 294 int background_worker_Push( struct background_worker* worker, void* entity,
 295                         void* id, int timeout )
 296 {
 297     struct task *task = task_Create(worker, id, entity, timeout);
 298     if (unlikely(!task))
 299         return VLC_ENOMEM;
 300
 301     vlc_mutex_lock(&worker->lock);
 302     QueuePush(worker, task);
 303     if (++worker->uncompleted > worker->nthreads
 304             && worker->nthreads < worker->conf.max_threads)
 305         SpawnThread(worker);
 306     vlc_mutex_unlock(&worker->lock);
 307
 308     return VLC_SUCCESS;
 309 }
 310
 311 static void BackgroundWorkerCancelLocked(struct background_worker *worker,
 312                                          void *id)
 313 {
 314     vlc_assert_locked(&worker->lock);
 315
 316     QueueRemoveAll(worker, id);
 317
 318     struct background_thread *thread;
 319     vlc_list_foreach(thread, &worker->threads, node)
 320     {
 321         if (!id || (thread->task && thread->task->id == id && !thread->cancel))
 322         {
 323             thread->cancel = true;
 324             vlc_cond_signal(&thread->probe_cancel_wait);
 325         }
 326     }
 327 }
 328
 329 void background_worker_Cancel( struct background_worker* worker, void* id )
 330 {
 331     vlc_mutex_lock(&worker->lock);
 332     BackgroundWorkerCancelLocked(worker, id);
 333     vlc_mutex_unlock(&worker->lock);
 334 }
 335
 336 void background_worker_RequestProbe( struct background_worker* worker )
 337 {
 338     vlc_mutex_lock(&worker->lock);
 339
 340     struct background_thread *thread;
 341     vlc_list_foreach(thread, &worker->threads, node)
 342     {
 343         thread->probe = true;
 344         vlc_cond_signal(&thread->probe_cancel_wait);
 345     }
 346
 347     vlc_mutex_unlock(&worker->lock);
 348 }
 349
 350 void background_worker_Delete( struct background_worker* worker )
 351 {
 352     vlc_mutex_lock(&worker->lock);
 353
 354     worker->closing = true;
 355     BackgroundWorkerCancelLocked(worker, NULL);
 356     /* closing is now true, this will wake up any QueueTake() */
 357     vlc_cond_broadcast(&worker->queue_wait);
 358
 359     while (worker->nthreads)
 360         vlc_cond_wait(&worker->nothreads_wait, &worker->lock);
 361
 362     vlc_mutex_unlock(&worker->lock);
 363
 364     /* no threads use the worker anymore, we can destroy it */
 365     background_worker_Destroy(worker);
 366 }