diff --git a/src/lib/openjp2/CMakeLists.txt b/src/lib/openjp2/CMakeLists.txt index c02a9948..f45ceb34 100644 --- a/src/lib/openjp2/CMakeLists.txt +++ b/src/lib/openjp2/CMakeLists.txt @@ -9,6 +9,8 @@ include_directories( ) # Defines the source code for the library set(OPENJPEG_SRCS + ${CMAKE_CURRENT_SOURCE_DIR}/thread.c + ${CMAKE_CURRENT_SOURCE_DIR}/thread.h ${CMAKE_CURRENT_SOURCE_DIR}/bio.c ${CMAKE_CURRENT_SOURCE_DIR}/bio.h ${CMAKE_CURRENT_SOURCE_DIR}/cio.c @@ -74,6 +76,11 @@ if(OPJ_DISABLE_TPSOT_FIX) add_definitions(-DOPJ_DISABLE_TPSOT_FIX) endif() +# Special case for old i586-mingw32msvc-gcc cross compiler +if(NOT WIN32 AND CMAKE_COMPILER_IS_GNUCC AND CMAKE_C_COMPILER MATCHES ".*mingw32msvc.*" ) + set(WIN32 YES) +endif() + # Build the library if(WIN32) if(BUILD_SHARED_LIBS) @@ -143,3 +150,36 @@ if(OPJ_USE_DSYMUTIL) DEPENDS ${OPENJPEG_LIBRARY_NAME}) endif() endif() + +################################################################################# +# threading configuration +################################################################################# +set(CMAKE_THREAD_PREFER_PTHREAD TRUE) + +option(USE_THREAD "Build with thread/mutex support " ON) +if(NOT USE_THREAD) + add_definitions( -DMUTEX_stub) +endif(NOT USE_THREAD) + +find_package(Threads QUIET) + +if(USE_THREAD AND WIN32 AND NOT Threads_FOUND ) + add_definitions( -DMUTEX_win32) + set(Threads_FOUND YES) +endif() + +if(USE_THREAD AND Threads_FOUND AND CMAKE_USE_WIN32_THREADS_INIT ) + add_definitions( -DMUTEX_win32) +endif(USE_THREAD AND Threads_FOUND AND CMAKE_USE_WIN32_THREADS_INIT ) + +if(USE_THREAD AND Threads_FOUND AND CMAKE_USE_PTHREADS_INIT ) + add_definitions( -DMUTEX_pthread) +endif(USE_THREAD AND Threads_FOUND AND CMAKE_USE_PTHREADS_INIT ) + +if(USE_THREAD AND NOT Threads_FOUND) + message(FATAL_ERROR "No thread library found and thread/mutex support is required by USE_THREAD option") +endif(USE_THREAD AND NOT Threads_FOUND) + +if(USE_THREAD AND Threads_FOUND AND CMAKE_USE_PTHREADS_INIT) + TARGET_LINK_LIBRARIES(${OPENJPEG_LIBRARY_NAME} ${CMAKE_THREAD_LIBS_INIT}) +endif(USE_THREAD AND Threads_FOUND AND CMAKE_USE_PTHREADS_INIT) diff --git a/src/lib/openjp2/openjpeg.h b/src/lib/openjp2/openjpeg.h index c07e9c84..369693df 100644 --- a/src/lib/openjp2/openjpeg.h +++ b/src/lib/openjp2/openjpeg.h @@ -1554,6 +1554,19 @@ OPJ_API OPJ_BOOL OPJ_CALLCONV opj_set_MCT( opj_cparameters_t *parameters, OPJ_INT32 * p_dc_shift, OPJ_UINT32 pNbComp); +/* +========================================================== + Thread functions +========================================================== +*/ + +/** Returns if the library is built with thread support. + * OPJ_TRUE if mutex, condition, thread, thread pool are available. + */ +OPJ_API OPJ_BOOL OPJ_CALLCONV opj_has_thread_support(void); + +/** Return the number of virtual CPUs */ +OPJ_API int OPJ_CALLCONV opj_get_num_cpus(void); #ifdef __cplusplus diff --git a/src/lib/openjp2/opj_includes.h b/src/lib/openjp2/opj_includes.h index 58a5a9a9..c2cc31fa 100644 --- a/src/lib/openjp2/opj_includes.h +++ b/src/lib/openjp2/opj_includes.h @@ -182,6 +182,8 @@ static INLINE long opj_lrintf(float f) { #include "bio.h" #include "cio.h" +#include "thread.h" + #include "image.h" #include "invert.h" #include "j2k.h" diff --git a/src/lib/openjp2/thread.c b/src/lib/openjp2/thread.c new file mode 100644 index 00000000..b2f8b5b2 --- /dev/null +++ b/src/lib/openjp2/thread.c @@ -0,0 +1,959 @@ +/* + * The copyright in this software is being made available under the 2-clauses + * BSD License, included below. This software may be subject to other third + * party and contributor rights, including patent rights, and no such rights + * are granted under this license. + * + * Copyright (c) 2016, Even Rouault + * All rights reserved. + * + * Redistribution and use in source and binary forms, with or without + * modification, are permitted provided that the following conditions + * are met: + * 1. Redistributions of source code must retain the above copyright + * notice, this list of conditions and the following disclaimer. + * 2. Redistributions in binary form must reproduce the above copyright + * notice, this list of conditions and the following disclaimer in the + * documentation and/or other materials provided with the distribution. + * + * THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS `AS IS' + * AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE + * IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR PURPOSE + * ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT OWNER OR CONTRIBUTORS BE + * LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL, SPECIAL, EXEMPLARY, OR + * CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT LIMITED TO, PROCUREMENT OF + * SUBSTITUTE GOODS OR SERVICES; LOSS OF USE, DATA, OR PROFITS; OR BUSINESS + * INTERRUPTION) HOWEVER CAUSED AND ON ANY THEORY OF LIABILITY, WHETHER IN + * CONTRACT, STRICT LIABILITY, OR TORT (INCLUDING NEGLIGENCE OR OTHERWISE) + * ARISING IN ANY WAY OUT OF THE USE OF THIS SOFTWARE, EVEN IF ADVISED OF THE + * POSSIBILITY OF SUCH DAMAGE. + */ + +#include "opj_includes.h" + +#include "thread.h" +#include + +#ifdef MUTEX_win32 + +/* Some versions of x86_64-w64-mingw32-gc -m32 resolve InterlockedCompareExchange() */ +/* as __sync_val_compare_and_swap_4 but fails to link it. As this protects against */ +/* a rather unlikely race, skip it */ +#if !(defined(__MINGW32__) && defined(__i386__)) +#define HAVE_INTERLOCKED_COMPARE_EXCHANGE 1 +#endif + +#include + +OPJ_BOOL OPJ_CALLCONV opj_has_thread_support(void) +{ + return OPJ_TRUE; +} + +int OPJ_CALLCONV opj_get_num_cpus(void) +{ + SYSTEM_INFO info; + DWORD dwNum; + GetSystemInfo(&info); + dwNum = info.dwNumberOfProcessors; + if( dwNum < 1 ) + return 1; + return (int)dwNum; +} + +struct opj_mutex_t +{ + CRITICAL_SECTION cs; +}; + +opj_mutex_t* opj_mutex_create(void) +{ + opj_mutex_t* mutex = (opj_mutex_t*) opj_malloc(sizeof(opj_mutex_t)); + if( !mutex ) + return NULL; + InitializeCriticalSectionAndSpinCount(&(mutex->cs), 4000); + return mutex; +} + +void opj_mutex_lock(opj_mutex_t* mutex) +{ + EnterCriticalSection( &(mutex->cs) ); +} + +void opj_mutex_unlock(opj_mutex_t* mutex) +{ + LeaveCriticalSection( &(mutex->cs) ); +} + +void opj_mutex_destroy(opj_mutex_t* mutex) +{ + if( !mutex ) return; + DeleteCriticalSection( &(mutex->cs) ); + opj_free( mutex ); +} + +struct opj_cond_waiter_list_t +{ + HANDLE hEvent; + struct opj_cond_waiter_list_t* next; +}; +typedef struct opj_cond_waiter_list_t opj_cond_waiter_list_t; + +struct opj_cond_t +{ + opj_mutex_t *internal_mutex; + opj_cond_waiter_list_t *waiter_list; +}; + +static DWORD TLSKey = 0; +static volatile LONG inTLSLockedSection = 0; +static volatile int TLSKeyInit = OPJ_FALSE; + +opj_cond_t* opj_cond_create(void) +{ + opj_cond_t* cond = (opj_cond_t*) opj_malloc(sizeof(opj_cond_t)); + if( !cond ) + return NULL; + + /* Make sure that the TLS key is allocated in a thread-safe way */ + /* We cannot use a global mutex/critical section since its creation itself would not be */ + /* thread-safe, so use InterlockedCompareExchange trick */ + while( OPJ_TRUE ) + { + +#if HAVE_INTERLOCKED_COMPARE_EXCHANGE + if( InterlockedCompareExchange(&inTLSLockedSection, 1, 0) == 0 ) +#endif + { + if( !TLSKeyInit ) + { + TLSKey = TlsAlloc(); + TLSKeyInit = OPJ_TRUE; + } +#if HAVE_INTERLOCKED_COMPARE_EXCHANGE + InterlockedCompareExchange(&inTLSLockedSection, 0, 1); +#endif + break; + } + } + + if( TLSKey == TLS_OUT_OF_INDEXES ) + { + opj_free(cond); + return NULL; + } + cond->internal_mutex = opj_mutex_create(); + if (cond->internal_mutex == NULL) + { + opj_free(cond); + return NULL; + } + cond->waiter_list = NULL; + return cond; +} + +void opj_cond_wait(opj_cond_t* cond, opj_mutex_t* mutex) +{ + opj_cond_waiter_list_t* item; + HANDLE hEvent = (HANDLE) TlsGetValue( TLSKey ); + if (hEvent == NULL) + { + hEvent = CreateEvent(NULL, /* security attributes */ + 0, /* manual reset = no */ + 0, /* initial state = unsignaled */ + NULL /* no name */); + assert(hEvent); + + TlsSetValue( TLSKey, hEvent ); + } + + /* Insert the waiter into the waiter list of the condition */ + opj_mutex_lock(cond->internal_mutex); + + item = (opj_cond_waiter_list_t*)opj_malloc(sizeof(opj_cond_waiter_list_t)); + assert(item != NULL); + + item->hEvent = hEvent; + item->next = cond->waiter_list; + + cond->waiter_list = item; + + opj_mutex_unlock(cond->internal_mutex); + + /* Release the client mutex before waiting for the event being signaled */ + opj_mutex_unlock(mutex); + + /* Ideally we would check that we do not get WAIT_FAILED but it is hard */ + /* to report a failure. */ + WaitForSingleObject(hEvent, INFINITE); + + /* Reacquire the client mutex */ + opj_mutex_lock(mutex); +} + +void opj_cond_signal(opj_cond_t* cond) +{ + opj_cond_waiter_list_t* psIter; + + /* Signal the first registered event, and remove it from the list */ + opj_mutex_lock(cond->internal_mutex); + + psIter = cond->waiter_list; + if (psIter != NULL) + { + SetEvent(psIter->hEvent); + cond->waiter_list = psIter->next; + opj_free(psIter); + } + + opj_mutex_unlock(cond->internal_mutex); +} + +void opj_cond_destroy(opj_cond_t* cond) +{ + if( !cond ) return; + opj_mutex_destroy(cond->internal_mutex); + assert(cond->waiter_list == NULL); + opj_free(cond); +} + +struct opj_thread_t +{ + opj_thread_fn thread_fn; + void* user_data; + HANDLE hThread; +}; + +static DWORD WINAPI opj_thread_callback_adapter( void *info ) +{ + opj_thread_t* thread = (opj_thread_t*) info; + HANDLE hEvent = NULL; + + thread->thread_fn( thread->user_data ); + + /* Free the handle possible allocated by a cond */ + while( OPJ_TRUE ) + { + /* Make sure TLSKey is not being created just at that moment... */ +#if HAVE_INTERLOCKED_COMPARE_EXCHANGE + if( InterlockedCompareExchange(&inTLSLockedSection, 1, 0) == 0 ) +#endif + { + if( TLSKeyInit ) + { + hEvent = (HANDLE) TlsGetValue( TLSKey ); + } +#if HAVE_INTERLOCKED_COMPARE_EXCHANGE + InterlockedCompareExchange(&inTLSLockedSection, 0, 1); +#endif + break; + } + } + if( hEvent ) + CloseHandle(hEvent); + + return 0; +} + +opj_thread_t* opj_thread_create( opj_thread_fn thread_fn, void* user_data ) +{ + opj_thread_t* thread; + DWORD nThreadId = 0; + + assert( thread_fn ); + + thread = (opj_thread_t*) opj_malloc( sizeof(opj_thread_t) ); + if( !thread ) + return NULL; + thread->thread_fn = thread_fn; + thread->user_data = user_data; + + thread->hThread = CreateThread( NULL, 0, opj_thread_callback_adapter, thread, + 0, &nThreadId ); + + if( thread->hThread == NULL ) + { + opj_free( thread ); + return NULL; + } + return thread; +} + +void opj_thread_join( opj_thread_t* thread ) +{ + WaitForSingleObject(thread->hThread, INFINITE); + CloseHandle( thread->hThread ); + + opj_free(thread); +} + +#elif MUTEX_pthread + +#include +#include +#include + +OPJ_BOOL OPJ_CALLCONV opj_has_thread_support(void) +{ + return OPJ_TRUE; +} + +int OPJ_CALLCONV opj_get_num_cpus(void) +{ +#ifdef _SC_NPROCESSORS_ONLN + return (int)sysconf(_SC_NPROCESSORS_ONLN); +#else + return 1; +#endif +} + +struct opj_mutex_t +{ + pthread_mutex_t mutex; +}; + +opj_mutex_t* opj_mutex_create(void) +{ + opj_mutex_t* mutex = (opj_mutex_t*) opj_malloc(sizeof(opj_mutex_t)); + if( !mutex ) + return NULL; + pthread_mutex_t pthr_mutex = PTHREAD_MUTEX_INITIALIZER; + mutex->mutex = pthr_mutex; + return mutex; +} + +void opj_mutex_lock(opj_mutex_t* mutex) +{ + pthread_mutex_lock(&(mutex->mutex)); +} + +void opj_mutex_unlock(opj_mutex_t* mutex) +{ + pthread_mutex_unlock(&(mutex->mutex)); +} + +void opj_mutex_destroy(opj_mutex_t* mutex) +{ + if( !mutex ) return; + pthread_mutex_destroy(&(mutex->mutex)); + opj_free(mutex); +} + +struct opj_cond_t +{ + pthread_cond_t cond; +}; + +opj_cond_t* opj_cond_create(void) +{ + opj_cond_t* cond = (opj_cond_t*) opj_malloc(sizeof(opj_cond_t)); + if( !cond ) + return NULL; + if( pthread_cond_init(&(cond->cond), NULL) != 0 ) + { + opj_free(cond); + return NULL; + } + return cond; +} + +void opj_cond_wait(opj_cond_t* cond, opj_mutex_t* mutex) +{ + pthread_cond_wait(&(cond->cond), &(mutex->mutex)); +} + +void opj_cond_signal(opj_cond_t* cond) +{ + int ret = pthread_cond_signal(&(cond->cond)); + (void)ret; + assert(ret == 0); +} + +void opj_cond_destroy(opj_cond_t* cond) +{ + if( !cond ) return; + pthread_cond_destroy(&(cond->cond)); + opj_free(cond); +} + + +struct opj_thread_t +{ + opj_thread_fn thread_fn; + void* user_data; + pthread_t thread; +}; + +static void* opj_thread_callback_adapter( void* info ) +{ + opj_thread_t* thread = (opj_thread_t*) info; + thread->thread_fn( thread->user_data ); + return NULL; +} + +opj_thread_t* opj_thread_create( opj_thread_fn thread_fn, void* user_data ) +{ + pthread_attr_t attr; + opj_thread_t* thread; + + assert( thread_fn ); + + thread = (opj_thread_t*) opj_malloc( sizeof(opj_thread_t) ); + if( !thread ) + return NULL; + thread->thread_fn = thread_fn; + thread->user_data = user_data; + + pthread_attr_init( &attr ); + pthread_attr_setdetachstate( &attr, PTHREAD_CREATE_JOINABLE ); + if( pthread_create( &(thread->thread), &attr, + opj_thread_callback_adapter, (void *) thread ) != 0 ) + { + opj_free( thread ); + return NULL; + } + return thread; +} + +void opj_thread_join( opj_thread_t* thread ) +{ + void* status; + pthread_join( thread->thread, &status); + + opj_free(thread); +} + +#else +/* Stub implementation */ + +OPJ_BOOL OPJ_CALLCONV opj_has_thread_support(void) +{ + return OPJ_FALSE; +} + +int OPJ_CALLCONV opj_get_num_cpus(void) +{ + return 1; +} + +opj_mutex_t* opj_mutex_create(void) +{ + return NULL; +} + +void opj_mutex_lock(opj_mutex_t* mutex) +{ + (void) mutex; +} + +void opj_mutex_unlock(opj_mutex_t* mutex) +{ + (void) mutex; +} + +void opj_mutex_destroy(opj_mutex_t* mutex) +{ + (void) mutex; +} + +opj_cond_t* opj_cond_create(void) +{ + return NULL; +} + +void opj_cond_wait(opj_cond_t* cond, opj_mutex_t* mutex) +{ + (void) cond; + (void) mutex; +} + +void opj_cond_signal(opj_cond_t* cond) +{ + (void) cond; +} + +void opj_cond_destroy(opj_cond_t* cond) +{ + (void) cond; +} + +opj_thread_t* opj_thread_create( opj_thread_fn thread_fn, void* user_data ) +{ + (void) thread_fn; + (void) user_data; + return NULL; +} + +void opj_thread_join( opj_thread_t* thread ) +{ + (void) thread; +} + +#endif + +typedef struct +{ + int key; + void* value; + opj_tls_free_func opj_free_func; +} opj_tls_key_val_t; + +struct opj_tls_t +{ + opj_tls_key_val_t* key_val; + int key_val_count; +}; + +static opj_tls_t* opj_tls_new(void) +{ + return (opj_tls_t*) opj_calloc(1, sizeof(opj_tls_t)); +} + +static void opj_tls_destroy(opj_tls_t* tls) +{ + int i; + if( !tls ) return; + for(i=0;ikey_val_count;i++) + { + if( tls->key_val[i].opj_free_func ) + tls->key_val[i].opj_free_func(tls->key_val[i].value); + } + opj_free(tls->key_val); + opj_free(tls); +} + +void* opj_tls_get(opj_tls_t* tls, int key) +{ + int i; + for(i=0;ikey_val_count;i++) + { + if( tls->key_val[i].key == key ) + return tls->key_val[i].value; + } + return NULL; +} + +OPJ_BOOL opj_tls_set(opj_tls_t* tls, int key, void* value, opj_tls_free_func opj_free_func) +{ + opj_tls_key_val_t* new_key_val; + int i; + for(i=0;ikey_val_count;i++) + { + if( tls->key_val[i].key == key ) + { + if( tls->key_val[i].opj_free_func ) + tls->key_val[i].opj_free_func(tls->key_val[i].value); + tls->key_val[i].value = value; + tls->key_val[i].opj_free_func = opj_free_func; + return OPJ_TRUE; + } + } + new_key_val = (opj_tls_key_val_t*) opj_realloc( tls->key_val, + (tls->key_val_count + 1) * sizeof(opj_tls_key_val_t) ); + if( !new_key_val ) + return OPJ_FALSE; + tls->key_val = new_key_val; + new_key_val[tls->key_val_count].key = key; + new_key_val[tls->key_val_count].value = value; + new_key_val[tls->key_val_count].opj_free_func = opj_free_func; + tls->key_val_count ++; + return OPJ_TRUE; +} + + +typedef struct +{ + opj_job_fn job_fn; + void *user_data; +} opj_worker_thread_job_t; + +typedef struct +{ + opj_thread_pool_t *tp; + opj_thread_t *thread; + int marked_as_waiting; + + opj_mutex_t *mutex; + opj_cond_t *cond; +} opj_worker_thread_t; + +typedef enum +{ + OPJWTS_OK, + OPJWTS_STOP, + OPJWTS_ERROR +} opj_worker_thread_state; + +struct opj_job_list_t +{ + opj_worker_thread_job_t* job; + struct opj_job_list_t* next; +}; +typedef struct opj_job_list_t opj_job_list_t; + +struct opj_worker_thread_list_t +{ + opj_worker_thread_t* worker_thread; + struct opj_worker_thread_list_t* next; +}; +typedef struct opj_worker_thread_list_t opj_worker_thread_list_t; + +struct opj_thread_pool_t +{ + opj_worker_thread_t* worker_threads; + int worker_threads_count; + opj_cond_t* cond; + opj_mutex_t* mutex; + volatile opj_worker_thread_state state; + opj_job_list_t* job_queue; + volatile int pending_jobs_count; + opj_worker_thread_list_t* waiting_worker_thread_list; + int waiting_worker_thread_count; + opj_tls_t* tls; + int signaling_threshold; +}; + +static OPJ_BOOL opj_thread_pool_setup(opj_thread_pool_t* tp, int num_threads); +static opj_worker_thread_job_t* opj_thread_pool_get_next_job(opj_thread_pool_t* tp, + opj_worker_thread_t* worker_thread, + OPJ_BOOL signal_job_finished); + +opj_thread_pool_t* opj_thread_pool_create(int num_threads) +{ + opj_thread_pool_t* tp; + + tp = (opj_thread_pool_t*) opj_calloc(1, sizeof(opj_thread_pool_t)); + if( !tp ) + return NULL; + tp->state = OPJWTS_OK; + + if( num_threads <= 0 ) + { + tp->tls = opj_tls_new(); + if( !tp->tls ) + { + opj_free(tp); + tp = NULL; + } + return tp; + } + + tp->mutex = opj_mutex_create(); + if( !tp->mutex ) + { + opj_free(tp); + return NULL; + } + if( !opj_thread_pool_setup(tp, num_threads) ) + { + opj_thread_pool_destroy(tp); + return NULL; + } + return tp; +} + +static void opj_worker_thread_function(void* user_data) +{ + opj_worker_thread_t* worker_thread; + opj_thread_pool_t* tp; + opj_tls_t* tls; + OPJ_BOOL job_finished = OPJ_FALSE; + + worker_thread = (opj_worker_thread_t* ) user_data; + tp = worker_thread->tp; + tls = opj_tls_new(); + + while( OPJ_TRUE ) + { + opj_worker_thread_job_t* job = opj_thread_pool_get_next_job(tp, worker_thread, job_finished); + if( job == NULL ) + break; + + if( job->job_fn ) + { + job->job_fn(job->user_data, tls); + } + opj_free(job); + job_finished = OPJ_TRUE; + } + + opj_tls_destroy(tls); +} + +static OPJ_BOOL opj_thread_pool_setup(opj_thread_pool_t* tp, int num_threads) +{ + int i; + OPJ_BOOL bRet = OPJ_TRUE; + + assert( num_threads > 0 ); + + tp->cond = opj_cond_create(); + if( tp->cond == NULL ) + return OPJ_FALSE; + + tp->worker_threads = (opj_worker_thread_t*) opj_calloc( num_threads, + sizeof(opj_worker_thread_t) ); + if( tp->worker_threads == NULL ) + return OPJ_FALSE; + tp->worker_threads_count = num_threads; + + for(i=0;iworker_threads[i].tp = tp; + + tp->worker_threads[i].mutex = opj_mutex_create(); + if( tp->worker_threads[i].mutex == NULL ) + { + tp->worker_threads_count = i; + bRet = OPJ_FALSE; + break; + } + + tp->worker_threads[i].cond = opj_cond_create(); + if( tp->worker_threads[i].cond == NULL ) + { + opj_mutex_destroy(tp->worker_threads[i].mutex); + tp->worker_threads_count = i; + bRet = OPJ_FALSE; + break; + } + + tp->worker_threads[i].marked_as_waiting = OPJ_FALSE; + + tp->worker_threads[i].thread = opj_thread_create(opj_worker_thread_function, + &(tp->worker_threads[i])); + if( tp->worker_threads[i].thread == NULL ) + { + tp->worker_threads_count = i; + bRet = OPJ_FALSE; + break; + } + } + + /* Wait all threads to be started */ + /* printf("waiting for all threads to be started\n"); */ + opj_mutex_lock(tp->mutex); + while( tp->waiting_worker_thread_count < num_threads ) + { + opj_cond_wait(tp->cond, tp->mutex); + } + opj_mutex_unlock(tp->mutex); + /* printf("all threads started\n"); */ + + if( tp->state == OPJWTS_ERROR ) + bRet = OPJ_FALSE; + + return bRet; +} + +/* +void opj_waiting() +{ + printf("waiting!\n"); +} +*/ + +static opj_worker_thread_job_t* opj_thread_pool_get_next_job(opj_thread_pool_t* tp, + opj_worker_thread_t* worker_thread, + OPJ_BOOL signal_job_finished) +{ + while( OPJ_TRUE ) + { + opj_job_list_t* top_job_iter; + + opj_mutex_lock(tp->mutex); + + if( signal_job_finished ) + { + signal_job_finished = OPJ_FALSE; + tp->pending_jobs_count --; + /*printf("tp=%p, remaining jobs: %d\n", tp, tp->pending_jobs_count);*/ + if( tp->pending_jobs_count <= tp->signaling_threshold ) + opj_cond_signal(tp->cond); + } + + if( tp->state == OPJWTS_STOP ) + { + opj_mutex_unlock(tp->mutex); + return NULL; + } + top_job_iter = tp->job_queue; + if( top_job_iter ) + { + opj_worker_thread_job_t* job; + tp->job_queue = top_job_iter->next; + + job = top_job_iter->job; + opj_mutex_unlock(tp->mutex); + opj_free(top_job_iter); + return job; + } + + /* opj_waiting(); */ + if( !worker_thread->marked_as_waiting ) + { + opj_worker_thread_list_t* item; + + worker_thread->marked_as_waiting = OPJ_TRUE; + tp->waiting_worker_thread_count ++; + assert(tp->waiting_worker_thread_count <= tp->worker_threads_count); + + item= (opj_worker_thread_list_t*) opj_malloc(sizeof(opj_worker_thread_list_t)); + if( item == NULL ) + { + tp->state = OPJWTS_ERROR; + opj_cond_signal(tp->cond); + + opj_mutex_unlock(tp->mutex); + return NULL; + } + + item->worker_thread = worker_thread; + item->next = tp->waiting_worker_thread_list; + tp->waiting_worker_thread_list = item; + } + + /* printf("signaling that worker thread is ready\n"); */ + opj_cond_signal(tp->cond); + + opj_mutex_lock(worker_thread->mutex); + opj_mutex_unlock(tp->mutex); + + /* printf("waiting for job\n"); */ + opj_cond_wait( worker_thread->cond, worker_thread->mutex ); + + opj_mutex_unlock(worker_thread->mutex); + /* printf("got job\n"); */ + } +} + +OPJ_BOOL opj_thread_pool_submit_job(opj_thread_pool_t* tp, + opj_job_fn job_fn, + void* user_data) +{ + opj_worker_thread_job_t* job; + opj_job_list_t* item; + + if( tp->mutex == NULL ) + { + job_fn( user_data, tp->tls ); + return OPJ_TRUE; + } + + job = (opj_worker_thread_job_t*)opj_malloc(sizeof(opj_worker_thread_job_t)); + if( job == NULL ) + return OPJ_FALSE; + job->job_fn = job_fn; + job->user_data = user_data; + + item = (opj_job_list_t*) opj_malloc(sizeof(opj_job_list_t)); + if( item == NULL ) + { + opj_free(job); + return OPJ_FALSE; + } + item->job = job; + + opj_mutex_lock(tp->mutex); + + tp->signaling_threshold = 100 * tp->worker_threads_count; + while( tp->pending_jobs_count > tp->signaling_threshold ) + { + /* printf("%d jobs enqueued. Waiting\n", tp->pending_jobs_count); */ + opj_cond_wait(tp->cond, tp->mutex); + /* printf("...%d jobs enqueued.\n", tp->pending_jobs_count); */ + } + + item->next = tp->job_queue; + tp->job_queue = item; + tp->pending_jobs_count ++; + + if( tp->waiting_worker_thread_list ) + { + opj_worker_thread_t* worker_thread; + opj_worker_thread_list_t* next; + opj_worker_thread_list_t* to_opj_free; + + worker_thread = tp->waiting_worker_thread_list->worker_thread; + + assert( worker_thread->marked_as_waiting ); + worker_thread->marked_as_waiting = OPJ_FALSE; + + next = tp->waiting_worker_thread_list->next; + to_opj_free = tp->waiting_worker_thread_list; + tp->waiting_worker_thread_list = next; + tp->waiting_worker_thread_count --; + + opj_mutex_lock(worker_thread->mutex); + opj_mutex_unlock(tp->mutex); + opj_cond_signal(worker_thread->cond); + opj_mutex_unlock(worker_thread->mutex); + + opj_free(to_opj_free); + } + else + opj_mutex_unlock(tp->mutex); + + return OPJ_TRUE; +} + +void opj_thread_pool_wait_completion(opj_thread_pool_t* tp, int max_remaining_jobs) +{ + if( tp->mutex == NULL ) + { + return; + } + + if( max_remaining_jobs < 0 ) + max_remaining_jobs = 0; + opj_mutex_lock(tp->mutex); + tp->signaling_threshold = max_remaining_jobs; + while( tp->pending_jobs_count > max_remaining_jobs ) + { + /*printf("tp=%p, jobs before wait = %d, max_remaining_jobs = %d\n", tp, tp->pending_jobs_count, max_remaining_jobs);*/ + opj_cond_wait(tp->cond, tp->mutex); + /*printf("tp=%p, jobs after wait = %d\n", tp, tp->pending_jobs_count);*/ + } + opj_mutex_unlock(tp->mutex); +} + +int opj_thread_pool_get_thread_count(opj_thread_pool_t* tp) +{ + return tp->worker_threads_count; +} + +void opj_thread_pool_destroy(opj_thread_pool_t* tp) +{ + if( !tp ) return; + if( tp->cond ) + { + int i; + opj_thread_pool_wait_completion(tp, 0); + + tp->state = OPJWTS_STOP; + + for(i=0;iworker_threads_count;i++) + { + opj_mutex_lock(tp->worker_threads[i].mutex); + opj_cond_signal(tp->worker_threads[i].cond); + opj_mutex_unlock(tp->worker_threads[i].mutex); + opj_thread_join(tp->worker_threads[i].thread); + opj_cond_destroy(tp->worker_threads[i].cond); + opj_mutex_destroy(tp->worker_threads[i].mutex); + } + + opj_free(tp->worker_threads); + + while( tp->waiting_worker_thread_list != NULL ) + { + opj_worker_thread_list_t* next = tp->waiting_worker_thread_list->next; + opj_free( tp->waiting_worker_thread_list ); + tp->waiting_worker_thread_list = next; + } + + opj_cond_destroy(tp->cond); + } + opj_mutex_destroy(tp->mutex); + opj_tls_destroy(tp->tls); + opj_free(tp); +} diff --git a/src/lib/openjp2/thread.h b/src/lib/openjp2/thread.h new file mode 100644 index 00000000..241e6d88 --- /dev/null +++ b/src/lib/openjp2/thread.h @@ -0,0 +1,253 @@ +/* + * The copyright in this software is being made available under the 2-clauses + * BSD License, included below. This software may be subject to other third + * party and contributor rights, including patent rights, and no such rights + * are granted under this license. + * + * Copyright (c) 2016, Even Rouault + * All rights reserved. + * + * Redistribution and use in source and binary forms, with or without + * modification, are permitted provided that the following conditions + * are met: + * 1. Redistributions of source code must retain the above copyright + * notice, this list of conditions and the following disclaimer. + * 2. Redistributions in binary form must reproduce the above copyright + * notice, this list of conditions and the following disclaimer in the + * documentation and/or other materials provided with the distribution. + * + * THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS `AS IS' + * AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE + * IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR PURPOSE + * ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT OWNER OR CONTRIBUTORS BE + * LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL, SPECIAL, EXEMPLARY, OR + * CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT LIMITED TO, PROCUREMENT OF + * SUBSTITUTE GOODS OR SERVICES; LOSS OF USE, DATA, OR PROFITS; OR BUSINESS + * INTERRUPTION) HOWEVER CAUSED AND ON ANY THEORY OF LIABILITY, WHETHER IN + * CONTRACT, STRICT LIABILITY, OR TORT (INCLUDING NEGLIGENCE OR OTHERWISE) + * ARISING IN ANY WAY OUT OF THE USE OF THIS SOFTWARE, EVEN IF ADVISED OF THE + * POSSIBILITY OF SUCH DAMAGE. + */ + +#ifndef THREAD_H +#define THREAD_H + +#include "openjpeg.h" + +/** +@file thread.h +@brief Thread API + +The functions in thread.c have for goal to manage mutex, conditions, thread +creation and thread pools that accept jobs. +*/ + +/** @defgroup THREAD THREAD - Mutex, conditions, threads and thread pools */ +/*@{*/ + +/** @name Mutex */ +/*@{*/ + +/** Opaque type for a mutex */ +typedef struct opj_mutex_t opj_mutex_t; + +/** Creates a mutex. + * @return the mutex or NULL in case of error (can for example happen if the library + * is built without thread support) + */ +opj_mutex_t* opj_mutex_create(void); + +/** Lock/acquire the mutex. + * @param mutex the mutex to acquire. + */ +void opj_mutex_lock(opj_mutex_t* mutex); + +/** Unlock/release the mutex. + * @param mutex the mutex to release. + */ +void opj_mutex_unlock(opj_mutex_t* mutex); + +/** Destroy a mutex + * @param mutex the mutex to destroy. + */ +void opj_mutex_destroy(opj_mutex_t* mutex); + +/*@}*/ + +/** @name Condition */ +/*@{*/ + +/** Opaque type for a condition */ +typedef struct opj_cond_t opj_cond_t; + +/** Creates a condition. + * @return the condition or NULL in case of error (can for example happen if the library + * is built without thread support) + */ +opj_cond_t* opj_cond_create(void); + +/** Wait for the condition to be signaled. + * The semantics is the same as the POSIX pthread_cond_wait. + * The provided mutex *must* be acquired before calling this function, and + * released afterwards. + * The mutex will be released by this function while it must wait for the condition + * and reacquired afterwards. + * In some particular situations, the function might return even if the condition is not signaled + * with opj_cond_signal(), hence the need to check with an application level + * mechanism. + * + * Waiting thread : + * \code + * opj_mutex_lock(mutex); + * while( !some_application_level_condition ) + * { + * opj_cond_wait(cond, mutex); + * } + * opj_mutex_unlock(mutex); + * \endcode + * + * Signaling thread : + * \code + * opj_mutex_lock(mutex); + * some_application_level_condition = TRUE; + * opj_cond_signal(cond); + * opj_mutex_unlock(mutex); + * \endcode + * + * @param cond the condition to wait. + * @param mutex the mutex (in acquired state before calling this function) + */ +void opj_cond_wait(opj_cond_t* cond, opj_mutex_t* mutex); + +/** Signal waiting threads on a condition. + * One of the thread waiting with opj_cond_wait() will be waken up. + * It is strongly advised that this call is done with the mutex that is used + * by opj_cond_wait(), in a acquired state. + * @param cond the condition to signal. + */ +void opj_cond_signal(opj_cond_t* cond); + +/** Destroy a condition + * @param cond the condition to destroy. + */ +void opj_cond_destroy(opj_cond_t* cond); + +/*@}*/ + +/** @name Thread */ +/*@{*/ + +/** Opaque type for a thread handle */ +typedef struct opj_thread_t opj_thread_t; + +/** User function to execute in a thread + * @param user_data user data provided with opj_thread_create() + */ +typedef void (*opj_thread_fn)(void* user_data); + +/** Creates a new thread. + * @param thread_fn Function to run in the new thread. + * @param user_data user data provided to the thread function. Might be NULL. + * @return a thread handle or NULL in case of failure (can for example happen if the library + * is built without thread support) + */ +opj_thread_t* opj_thread_create( opj_thread_fn thread_fn, void* user_data ); + +/** Wait for a thread to be finished and release associated resources to the + * thread handle. + * @param thread the thread to wait for being finished. + */ +void opj_thread_join( opj_thread_t* thread ); + +/*@}*/ + +/** @name Thread local storage */ +/*@{*/ +/** Opaque type for a thread local storage */ +typedef struct opj_tls_t opj_tls_t; + +/** Get a thread local value corresponding to the provided key. + * @param tls thread local storage handle + * @param key key whose value to retrieve. + * @return value associated with the key, or NULL is missing. + */ +void* opj_tls_get(opj_tls_t* tls, int key); + +/** Type of the function used to free a TLS value */ +typedef void (*opj_tls_free_func)(void* value); + +/** Set a thread local value corresponding to the provided key. + * @param tls thread local storage handle + * @param key key whose value to set. + * @param value value to set (may be NULL). + * @param free_func function to call currently installed value. + * @return OPJ_TRUE if successful. + */ +OPJ_BOOL opj_tls_set(opj_tls_t* tls, int key, void* value, opj_tls_free_func free_func); + +/*@}*/ + +/** @name Thread pool */ +/*@{*/ + +/** Opaque type for a thread pool */ +typedef struct opj_thread_pool_t opj_thread_pool_t; + +/** Create a new thread pool. + * num_thread must nominally be >= 1 to create a real thread pool. If num_threads + * is negative or null, then a dummy thread pool will be created. All functions + * operating on the thread pool will work, but job submission will be run + * synchronously in the calling thread. + * + * @param num_threads the number of threads to allocate for this thread pool. + * @return a thread pool handle, or NULL in case of failure (can for example happen if the library + * is built without thread support) + */ +opj_thread_pool_t* opj_thread_pool_create(int num_threads); + +/** User function to execute in a thread + * @param user_data user data provided with opj_thread_create() + * @param tls handle to thread local storage + */ +typedef void (*opj_job_fn)(void* user_data, opj_tls_t* tls); + + +/** Submit a new job to be run by one of the thread in the thread pool. + * The job ( thread_fn, user_data ) will be added in the queue of jobs managed + * by the thread pool, and run by the first thread that is no longer busy. + * + * @param tp the thread pool handle. + * @param job_fn Function to run. Must not be NULL. + * @param user_data User data provided to thread_fn. + * @return OPJ_TRUE if the job was successfully submitted. + */ +OPJ_BOOL opj_thread_pool_submit_job(opj_thread_pool_t* tp, opj_job_fn job_fn, void* user_data); + +/** Wait that no more than max_remaining_jobs jobs are remaining in the queue of + * the thread pool. The aim of this function is to avoid submitting too many + * jobs while the thread pool cannot cope fast enough with them, which would + * result potentially in out-of-memory situations with too many job descriptions + * being queued. + * + * @param tp the thread pool handle + * @param max_remaining_jobs maximum number of jobs allowed to be queued without waiting. + */ +void opj_thread_pool_wait_completion(opj_thread_pool_t* tp, int max_remaining_jobs); + +/** Return the number of threads associated with the thread pool. + * + * @param tp the thread pool handle. + * @return number of threads associated with the thread pool. + */ +int opj_thread_pool_get_thread_count(opj_thread_pool_t* tp); + +/** Destroy a thread pool. + * @param tp the thread pool handle. + */ +void opj_thread_pool_destroy(opj_thread_pool_t* tp); + +/*@}*/ + +/*@}*/ + +#endif /* THREAD_H */