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

   1 /*****************************************************************************
   2  * threadpool.c: thread pooling
   3  *****************************************************************************
   4  * Copyright (C) 2010-2012 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 = job->func( job->arg ); /* execute the function */
  70         x264_sync_frame_list_push( &pool->done, (void*)job );
  71     }
  72 }
  73
  74 int x264_threadpool_init( x264_threadpool_t **p_pool, int threads,
  75                           void (*init_func)(void *), void *init_arg )
  76 {
  77     if( threads <= 0 )
  78         return -1;
  79
  80     x264_threadpool_t *pool;
  81     CHECKED_MALLOCZERO( pool, sizeof(x264_threadpool_t) );
  82     *p_pool = pool;
  83
  84     pool->init_func = init_func;
  85     pool->init_arg  = init_arg;
  86     pool->threads   = X264_MIN( threads, X264_THREAD_MAX );
  87
  88     CHECKED_MALLOC( pool->thread_handle, pool->threads * sizeof(x264_pthread_t) );
  89
  90     if( x264_sync_frame_list_init( &pool->uninit, pool->threads ) ||
  91         x264_sync_frame_list_init( &pool->run, pool->threads ) ||
  92         x264_sync_frame_list_init( &pool->done, pool->threads ) )
  93         goto fail;
  94
  95     for( int i = 0; i < pool->threads; i++ )
  96     {
  97        x264_threadpool_job_t *job;
  98        CHECKED_MALLOC( job, sizeof(x264_threadpool_job_t) );
  99        x264_sync_frame_list_push( &pool->uninit, (void*)job );
 100     }
 101     for( int i = 0; i < pool->threads; i++ )
 102         if( x264_pthread_create( pool->thread_handle+i, NULL, (void*)x264_threadpool_thread, pool ) )
 103             goto fail;
 104
 105     return 0;
 106 fail:
 107     return -1;
 108 }
 109
 110 void x264_threadpool_run( x264_threadpool_t *pool, void *(*func)(void *), void *arg )
 111 {
 112     x264_threadpool_job_t *job = (void*)x264_sync_frame_list_pop( &pool->uninit );
 113     job->func = func;
 114     job->arg  = arg;
 115     x264_sync_frame_list_push( &pool->run, (void*)job );
 116 }
 117
 118 void *x264_threadpool_wait( x264_threadpool_t *pool, void *arg )
 119 {
 120     x264_threadpool_job_t *job = NULL;
 121
 122     x264_pthread_mutex_lock( &pool->done.mutex );
 123     while( !job )
 124     {
 125         for( int i = 0; i < pool->done.i_size; i++ )
 126         {
 127             x264_threadpool_job_t *t = (void*)pool->done.list[i];
 128             if( t->arg == arg )
 129             {
 130                 job = (void*)x264_frame_shift( pool->done.list+i );
 131                 pool->done.i_size--;
 132             }
 133         }
 134         if( !job )
 135             x264_pthread_cond_wait( &pool->done.cv_fill, &pool->done.mutex );
 136     }
 137     x264_pthread_mutex_unlock( &pool->done.mutex );
 138
 139     void *ret = job->ret;
 140     x264_sync_frame_list_push( &pool->uninit, (void*)job );
 141     return ret;
 142 }
 143
 144 static void x264_threadpool_list_delete( x264_sync_frame_list_t *slist )
 145 {
 146     for( int i = 0; slist->list[i]; i++ )
 147     {
 148         x264_free( slist->list[i] );
 149         slist->list[i] = NULL;
 150     }
 151     x264_sync_frame_list_delete( slist );
 152 }
 153
 154 void x264_threadpool_delete( x264_threadpool_t *pool )
 155 {
 156     x264_pthread_mutex_lock( &pool->run.mutex );
 157     pool->exit = 1;
 158     x264_pthread_cond_broadcast( &pool->run.cv_fill );
 159     x264_pthread_mutex_unlock( &pool->run.mutex );
 160     for( int i = 0; i < pool->threads; i++ )
 161         x264_pthread_join( pool->thread_handle[i], NULL );
 162
 163     x264_threadpool_list_delete( &pool->uninit );
 164     x264_threadpool_list_delete( &pool->run );
 165     x264_threadpool_list_delete( &pool->done );
 166     x264_free( pool->thread_handle );
 167     x264_free( pool );
 168 }