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

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