git.sesse.net Git - x264/blob - common/threadpool.c

   1 /*****************************************************************************
   2  * threadpool.c: thread pooling
   3  *****************************************************************************
   4  * Copyright (C) 2010-2015 x264 project
   5  *
   6  * Authors: Steven Walters <kemuri9@gmail.com>
   7  *
   8  * This program is free software; you can redistribute it and/or modify
   9  * it under the terms of the GNU General Public License as published by
  10  * the Free Software Foundation; either version 2 of the License, or
  11  * (at your option) any later version.
  12  *
  13  * This program is distributed in the hope that it will be useful,
  14  * but WITHOUT ANY WARRANTY; without even the implied warranty of
  15  * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
  16  * GNU General Public License for more details.
  17  *
  18  * You should have received a copy of the GNU General Public License
  19  * along with this program; if not, write to the Free Software
  20  * Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston, MA  02111, USA.
  21  *
  22  * This program is also available under a commercial proprietary license.
  23  * For more information, contact us at licensing@x264.com.
  24  *****************************************************************************/
  25
  26 #include "common.h"
  27
  28 typedef struct
  29 {
  30     void *(*func)(void *);
  31     void *arg;
  32     void *ret;
  33 } x264_threadpool_job_t;
  34
  35 struct x264_threadpool_t
  36 {
  37     int            exit;
  38     int            threads;
  39     x264_pthread_t *thread_handle;
  40     void           (*init_func)(void *);
  41     void           *init_arg;
  42
  43     /* requires a synchronized list structure and associated methods,
  44        so use what is already implemented for frames */
  45     x264_sync_frame_list_t uninit; /* list of jobs that are awaiting use */
  46     x264_sync_frame_list_t run;    /* list of jobs that are queued for processing by the pool */
  47     x264_sync_frame_list_t done;   /* list of jobs that have finished processing */
  48 };
  49
  50 static void *x264_threadpool_thread( x264_threadpool_t *pool )
  51 {
  52     if( pool->init_func )
  53         pool->init_func( pool->init_arg );
  54
  55     while( !pool->exit )
  56     {
  57         x264_threadpool_job_t *job = NULL;
  58         x264_pthread_mutex_lock( &pool->run.mutex );
  59         while( !pool->exit && !pool->run.i_size )
  60             x264_pthread_cond_wait( &pool->run.cv_fill, &pool->run.mutex );
  61         if( pool->run.i_size )
  62         {
  63             job = (void*)x264_frame_shift( pool->run.list );
  64             pool->run.i_size--;
  65         }
  66         x264_pthread_mutex_unlock( &pool->run.mutex );
  67         if( !job )
  68             continue;
  69         job->ret = (void*)x264_stack_align( job->func, job->arg ); /* execute the function */
  70         x264_sync_frame_list_push( &pool->done, (void*)job );
  71     }
  72     return NULL;
  73 }
  74
  75 int x264_threadpool_init( x264_threadpool_t **p_pool, int threads,
  76                           void (*init_func)(void *), void *init_arg )
  77 {
  78     if( threads <= 0 )
  79         return -1;
  80
  81     x264_threadpool_t *pool;
  82     CHECKED_MALLOCZERO( pool, sizeof(x264_threadpool_t) );
  83     *p_pool = pool;
  84
  85     pool->init_func = init_func;
  86     pool->init_arg  = init_arg;
  87     pool->threads   = threads;
  88
  89     CHECKED_MALLOC( pool->thread_handle, pool->threads * sizeof(x264_pthread_t) );
  90
  91     if( x264_sync_frame_list_init( &pool->uninit, pool->threads ) ||
  92         x264_sync_frame_list_init( &pool->run, pool->threads ) ||
  93         x264_sync_frame_list_init( &pool->done, pool->threads ) )
  94         goto fail;
  95
  96     for( int i = 0; i < pool->threads; i++ )
  97     {
  98        x264_threadpool_job_t *job;
  99        CHECKED_MALLOC( job, sizeof(x264_threadpool_job_t) );
 100        x264_sync_frame_list_push( &pool->uninit, (void*)job );
 101     }
 102     for( int i = 0; i < pool->threads; i++ )
 103         if( x264_pthread_create( pool->thread_handle+i, NULL, (void*)x264_threadpool_thread, pool ) )
 104             goto fail;
 105
 106     return 0;
 107 fail:
 108     return -1;
 109 }
 110
 111 void x264_threadpool_run( x264_threadpool_t *pool, void *(*func)(void *), void *arg )
 112 {
 113     x264_threadpool_job_t *job = (void*)x264_sync_frame_list_pop( &pool->uninit );
 114     job->func = func;
 115     job->arg  = arg;
 116     x264_sync_frame_list_push( &pool->run, (void*)job );
 117 }
 118
 119 void *x264_threadpool_wait( x264_threadpool_t *pool, void *arg )
 120 {
 121     x264_threadpool_job_t *job = NULL;
 122
 123     x264_pthread_mutex_lock( &pool->done.mutex );
 124     while( !job )
 125     {
 126         for( int i = 0; i < pool->done.i_size; i++ )
 127         {
 128             x264_threadpool_job_t *t = (void*)pool->done.list[i];
 129             if( t->arg == arg )
 130             {
 131                 job = (void*)x264_frame_shift( pool->done.list+i );
 132                 pool->done.i_size--;
 133             }
 134         }
 135         if( !job )
 136             x264_pthread_cond_wait( &pool->done.cv_fill, &pool->done.mutex );
 137     }
 138     x264_pthread_mutex_unlock( &pool->done.mutex );
 139
 140     void *ret = job->ret;
 141     x264_sync_frame_list_push( &pool->uninit, (void*)job );
 142     return ret;
 143 }
 144
 145 static void x264_threadpool_list_delete( x264_sync_frame_list_t *slist )
 146 {
 147     for( int i = 0; slist->list[i]; i++ )
 148     {
 149         x264_free( slist->list[i] );
 150         slist->list[i] = NULL;
 151     }
 152     x264_sync_frame_list_delete( slist );
 153 }
 154
 155 void x264_threadpool_delete( x264_threadpool_t *pool )
 156 {
 157     x264_pthread_mutex_lock( &pool->run.mutex );
 158     pool->exit = 1;
 159     x264_pthread_cond_broadcast( &pool->run.cv_fill );
 160     x264_pthread_mutex_unlock( &pool->run.mutex );
 161     for( int i = 0; i < pool->threads; i++ )
 162         x264_pthread_join( pool->thread_handle[i], NULL );
 163
 164     x264_threadpool_list_delete( &pool->uninit );
 165     x264_threadpool_list_delete( &pool->run );
 166     x264_threadpool_list_delete( &pool->done );
 167     x264_free( pool->thread_handle );
 168     x264_free( pool );
 169 }