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_threads.h>
  26 #include <vlc_arrays.h>
  27
  28 #include "libvlc.h"
  29 #include "background_worker.h"
  30
  31 struct bg_queued_item {
  32     void* id; /**< id associated with entity */
  33     void* entity; /**< the entity to process */
  34     int timeout; /**< timeout duration in microseconds */
  35 };
  36
  37 struct background_worker {
  38     void* owner;
  39     struct background_worker_config conf;
  40
  41     vlc_mutex_t lock; /**< acquire to inspect members that follow */
  42     struct {
  43         bool probe_request; /**< true if a probe is requested */
  44         vlc_cond_t wait; /**< wait for update in terms of head */
  45         vlc_cond_t worker_wait; /**< wait for probe request or cancelation */
  46         mtime_t deadline; /**< deadline of the current task */
  47         void* id; /**< id of the current task */
  48         bool active; /**< true if there is an active thread */
  49     } head;
  50
  51     struct {
  52         vlc_cond_t wait; /**< wait for update in terms of tail */
  53         vlc_array_t data; /**< queue of pending entities to process */
  54     } tail;
  55 };
  56
  57 static void* Thread( void* data )
  58 {
  59     struct background_worker* worker = data;
  60
  61     for( ;; )
  62     {
  63         struct bg_queued_item* item = NULL;
  64         void* handle;
  65
  66         vlc_mutex_lock( &worker->lock );
  67         for( ;; )
  68         {
  69             if( vlc_array_count( &worker->tail.data ) )
  70             {
  71                 item = vlc_array_item_at_index( &worker->tail.data, 0 );
  72                 handle = NULL;
  73
  74                 vlc_array_remove( &worker->tail.data, 0 );
  75             }
  76
  77             if( worker->head.deadline == VLC_TS_0 && item == NULL )
  78                 worker->head.active = false;
  79             worker->head.id = item ? item->id : NULL;
  80             vlc_cond_broadcast( &worker->head.wait );
  81
  82             if( item )
  83             {
  84                 if( item->timeout > 0 )
  85                     worker->head.deadline = mdate() + item->timeout * 1000;
  86                 else
  87                     worker->head.deadline = INT64_MAX;
  88             }
  89             else if( worker->head.deadline != VLC_TS_0 )
  90             {
  91                 /* Wait 1 seconds for new inputs before terminating */
  92                 mtime_t deadline = mdate() + INT64_C(1000000);
  93                 int ret = vlc_cond_timedwait( &worker->tail.wait,
  94                                               &worker->lock, deadline );
  95                 if( ret != 0 )
  96                 {
  97                     /* Timeout: if there is still no items, the thread will be
  98                      * terminated at next loop iteration (active = false). */
  99                     worker->head.deadline = VLC_TS_0;
 100                 }
 101                 continue;
 102             }
 103             break;
 104         }
 105
 106         if( !worker->head.active )
 107         {
 108             vlc_mutex_unlock( &worker->lock );
 109             break;
 110         }
 111         vlc_mutex_unlock( &worker->lock );
 112
 113         assert( item != NULL );
 114
 115         if( worker->conf.pf_start( worker->owner, item->entity, &handle ) )
 116         {
 117             worker->conf.pf_release( item->entity );
 118             free( item );
 119             continue;
 120         }
 121
 122         for( ;; )
 123         {
 124             vlc_mutex_lock( &worker->lock );
 125
 126             bool const b_timeout = worker->head.deadline <= mdate();
 127             worker->head.probe_request = false;
 128
 129             vlc_mutex_unlock( &worker->lock );
 130
 131             if( b_timeout ||
 132                 worker->conf.pf_probe( worker->owner, handle ) )
 133             {
 134                 worker->conf.pf_stop( worker->owner, handle );
 135                 worker->conf.pf_release( item->entity );
 136                 free( item );
 137                 break;
 138             }
 139
 140             vlc_mutex_lock( &worker->lock );
 141             if( worker->head.probe_request == false &&
 142                 worker->head.deadline > mdate() )
 143             {
 144                 vlc_cond_timedwait( &worker->head.worker_wait, &worker->lock,
 145                                      worker->head.deadline );
 146             }
 147             vlc_mutex_unlock( &worker->lock );
 148         }
 149     }
 150
 151     return NULL;
 152 }
 153
 154 static void BackgroundWorkerCancel( struct background_worker* worker, void* id)
 155 {
 156     vlc_mutex_lock( &worker->lock );
 157     for( size_t i = 0; i < vlc_array_count( &worker->tail.data ); )
 158     {
 159         struct bg_queued_item* item =
 160             vlc_array_item_at_index( &worker->tail.data, i );
 161
 162         if( id == NULL || item->id == id )
 163         {
 164             vlc_array_remove( &worker->tail.data, i );
 165             worker->conf.pf_release( item->entity );
 166             free( item );
 167             continue;
 168         }
 169
 170         ++i;
 171     }
 172
 173     while( ( id == NULL && worker->head.active )
 174         || ( id != NULL && worker->head.id == id ) )
 175     {
 176         worker->head.deadline = VLC_TS_0;
 177         vlc_cond_signal( &worker->head.worker_wait );
 178         vlc_cond_signal( &worker->tail.wait );
 179         vlc_cond_wait( &worker->head.wait, &worker->lock );
 180     }
 181     vlc_mutex_unlock( &worker->lock );
 182 }
 183
 184 struct background_worker* background_worker_New( void* owner,
 185     struct background_worker_config* conf )
 186 {
 187     struct background_worker* worker = malloc( sizeof *worker );
 188
 189     if( unlikely( !worker ) )
 190         return NULL;
 191
 192     worker->conf = *conf;
 193     worker->owner = owner;
 194     worker->head.id = NULL;
 195     worker->head.active = false;
 196     worker->head.deadline = VLC_TS_INVALID;
 197
 198     vlc_mutex_init( &worker->lock );
 199     vlc_cond_init( &worker->head.wait );
 200     vlc_cond_init( &worker->head.worker_wait );
 201
 202     vlc_array_init( &worker->tail.data );
 203     vlc_cond_init( &worker->tail.wait );
 204
 205     return worker;
 206 }
 207
 208 int background_worker_Push( struct background_worker* worker, void* entity,
 209                         void* id, int timeout )
 210 {
 211     struct bg_queued_item* item = malloc( sizeof( *item ) );
 212
 213     if( unlikely( !item ) )
 214         return VLC_EGENERIC;
 215
 216     item->id = id;
 217     item->entity = entity;
 218     item->timeout = timeout < 0 ? worker->conf.default_timeout : timeout;
 219
 220     vlc_mutex_lock( &worker->lock );
 221     int i_ret = vlc_array_append( &worker->tail.data, item );
 222     vlc_cond_signal( &worker->tail.wait );
 223     if( i_ret != 0 )
 224     {
 225         free( item );
 226         return VLC_EGENERIC;
 227     }
 228
 229     if( worker->head.active == false )
 230     {
 231         worker->head.probe_request = false;
 232         worker->head.active =
 233             !vlc_clone_detach( NULL, Thread, worker, VLC_THREAD_PRIORITY_LOW );
 234     }
 235
 236     if( worker->head.active )
 237         worker->conf.pf_hold( item->entity );
 238
 239     int ret = worker->head.active ? VLC_SUCCESS : VLC_EGENERIC;
 240     vlc_mutex_unlock( &worker->lock );
 241
 242     return ret;
 243 }
 244
 245 void background_worker_Cancel( struct background_worker* worker, void* id )
 246 {
 247     BackgroundWorkerCancel( worker, id );
 248 }
 249
 250 void background_worker_RequestProbe( struct background_worker* worker )
 251 {
 252     vlc_mutex_lock( &worker->lock );
 253     worker->head.probe_request = true;
 254     vlc_cond_signal( &worker->head.worker_wait );
 255     vlc_mutex_unlock( &worker->lock );
 256 }
 257
 258 void background_worker_Delete( struct background_worker* worker )
 259 {
 260     BackgroundWorkerCancel( worker, NULL );
 261     vlc_array_clear( &worker->tail.data );
 262     vlc_mutex_destroy( &worker->lock );
 263     vlc_cond_destroy( &worker->head.wait );
 264     vlc_cond_destroy( &worker->head.worker_wait );
 265     vlc_cond_destroy( &worker->tail.wait );
 266     free( worker );
 267 }