Add threading and thread pool API

5 files changed, 1267 insertions, 0 deletions

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 <assert.h>
+
+#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 <windows.h>
+
+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 <pthread.h>
+#include <stdlib.h>
+#include <unistd.h>
+
+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;i<tls->key_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;i<tls->key_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;i<tls->key_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;i<num_threads;i++)
+    {
+        tp->worker_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;i<tp->worker_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 */