Add opj_codec_set_threads() in public API and propagate resulting thread pool to tcd level

By default, only the main thread is used. If opj_codec_set_threads() is not used,
but the OPJ_NUM_THREADS environment variable is set, its value will be
used to initialize the number of threads. The value can be either an integer
number, or "ALL_CPUS". If OPJ_NUM_THREADS is set and this function is called,
this function will override the behaviour of the environment variable.

src/lib/openjp2/j2k.c

@@ -5944,6 +5944,32 @@ void opj_j2k_setup_decoder(opj_j2k_t *j2k, opj_dparameters_t *parameters)
         }
 }
 
+OPJ_BOOL opj_j2k_set_threads(opj_j2k_t *j2k, OPJ_UINT32 num_threads)
+{
+        if( opj_has_thread_support() )
+        {
+            opj_thread_pool_destroy(j2k->m_tp);
+            j2k->m_tp = opj_thread_pool_create((int)num_threads);
+            if( j2k->m_tp == 0 )
+            {
+                j2k->m_tp = opj_thread_pool_create(0);
+                return OPJ_FALSE;
+            }
+            return OPJ_TRUE;
+        }
+        return OPJ_FALSE;
+}
+
+static int opj_j2k_get_default_thread_count()
+{
+    const char* num_threads = getenv("OPJ_NUM_THREADS");
+    if( num_threads == NULL || !opj_has_thread_support() )
+        return 0;
+    if( strcmp(num_threads, "ALL_CPUS") == 0 )
+        return opj_get_num_cpus();
+    return atoi(num_threads);
+}
+
 /* ----------------------------------------------------------------------- */
 /* J2K encoder interface                                                       */
 /* ----------------------------------------------------------------------- */
@@ -5981,6 +6007,17 @@ opj_j2k_t* opj_j2k_create_compress(void)
                 return NULL;
         }
 
+        l_j2k->m_tp = opj_thread_pool_create(opj_j2k_get_default_thread_count());
+        if( !l_j2k->m_tp )
+        {
+            l_j2k->m_tp = opj_thread_pool_create(0);
+        }
+        if( !l_j2k->m_tp )
+        {
+            opj_j2k_destroy(l_j2k);
+            return NULL;
+        }
+
         return l_j2k;
 }
 
@@ -7486,7 +7523,7 @@ static OPJ_BOOL opj_j2k_copy_default_tcp_and_create_tcd (       opj_j2k_t * p_j2
                 return OPJ_FALSE;
         }
 
-        if ( !opj_tcd_init(p_j2k->m_tcd, l_image, &(p_j2k->m_cp)) ) {
+        if ( !opj_tcd_init(p_j2k->m_tcd, l_image, &(p_j2k->m_cp), p_j2k->m_tp) ) {
                 opj_tcd_destroy(p_j2k->m_tcd);
                 p_j2k->m_tcd = 00;
                 opj_event_msg(p_manager, EVT_ERROR, "Cannot decode tile, memory error\n");
@@ -7567,6 +7604,9 @@ void opj_j2k_destroy (opj_j2k_t *p_j2k)
         opj_image_destroy(p_j2k->m_output_image);
         p_j2k->m_output_image = NULL;
 
+        opj_thread_pool_destroy(p_j2k->m_tp);
+        p_j2k->m_tp = NULL;
+
         opj_free(p_j2k);
 }
 
@@ -8658,6 +8698,17 @@ opj_j2k_t* opj_j2k_create_decompress(void)
                 return 00;
         }
 
+        l_j2k->m_tp = opj_thread_pool_create(opj_j2k_get_default_thread_count());
+        if( !l_j2k->m_tp )
+        {
+            l_j2k->m_tp = opj_thread_pool_create(0);
+        }
+        if( !l_j2k->m_tp )
+        {
+            opj_j2k_destroy(l_j2k);
+            return NULL;
+        }
+
         return l_j2k;
 }
 
@@ -10934,7 +10985,7 @@ static OPJ_BOOL opj_j2k_create_tcd(     opj_j2k_t *p_j2k,
                 return OPJ_FALSE;
         }
 
-        if (!opj_tcd_init(p_j2k->m_tcd,p_j2k->m_private_image,&p_j2k->m_cp)) {
+        if (!opj_tcd_init(p_j2k->m_tcd,p_j2k->m_private_image,&p_j2k->m_cp, p_j2k->m_tp)) {
                 opj_tcd_destroy(p_j2k->m_tcd);
                 p_j2k->m_tcd = 00;
                 return OPJ_FALSE;

src/lib/openjp2/j2k.h

@@ -589,6 +589,12 @@ typedef struct opj_j2k
 
 	/** the current tile coder/decoder **/
 	struct opj_tcd *	m_tcd;
+
+    /** Number of threads to use */
+    int m_num_threads;
+
+    /** Thread pool */
+    opj_thread_pool_t* m_tp;
 }
 opj_j2k_t;
 
@@ -607,6 +613,8 @@ Decoding parameters are returned in j2k->cp.
 */
 void opj_j2k_setup_decoder(opj_j2k_t *j2k, opj_dparameters_t *parameters);
 
+OPJ_BOOL opj_j2k_set_threads(opj_j2k_t *j2k, OPJ_UINT32 num_threads);
+
 /**
  * Creates a J2K compression structure
  *

src/lib/openjp2/jp2.c

@@ -1767,6 +1767,11 @@ void opj_jp2_setup_decoder(opj_jp2_t *jp2, opj_dparameters_t *parameters)
     jp2->ignore_pclr_cmap_cdef = parameters->flags & OPJ_DPARAMETERS_IGNORE_PCLR_CMAP_CDEF_FLAG;
 }
 
+OPJ_BOOL opj_jp2_set_threads(opj_jp2_t *jp2, OPJ_UINT32 num_threads)
+{
+     return opj_j2k_set_threads(jp2->j2k, num_threads);
+}
+
 /* ----------------------------------------------------------------------- */
 /* JP2 encoder interface                                             */
 /* ----------------------------------------------------------------------- */

src/lib/openjp2/jp2.h

@@ -243,6 +243,8 @@ Decoding parameters are returned in jp2->j2k->cp.
 */
 void opj_jp2_setup_decoder(opj_jp2_t *jp2, opj_dparameters_t *parameters);
 
+OPJ_BOOL opj_jp2_set_threads(opj_jp2_t *jp2, OPJ_UINT32 num_threads);
+
 /**
  * Decode an image from a JPEG-2000 file stream
  * @param jp2 JP2 decompressor handle

src/lib/openjp2/openjpeg.c

@@ -239,6 +239,9 @@ opj_codec_t* OPJ_CALLCONV opj_create_decompress(OPJ_CODEC_FORMAT p_format)
 									OPJ_UINT32 res_factor,
 									struct opj_event_mgr * p_manager)) opj_j2k_set_decoded_resolution_factor;
 
+            l_codec->opj_set_threads = 
+                    (OPJ_BOOL (*) ( void * p_codec, OPJ_UINT32 num_threads )) opj_j2k_set_threads;
+
 			l_codec->m_codec = opj_j2k_create_decompress();
 
 			if (! l_codec->m_codec) {
@@ -315,6 +318,9 @@ opj_codec_t* OPJ_CALLCONV opj_create_decompress(OPJ_CODEC_FORMAT p_format)
 						    		OPJ_UINT32 res_factor,
 							    	opj_event_mgr_t * p_manager)) opj_jp2_set_decoded_resolution_factor;
 
+            l_codec->opj_set_threads = 
+                    (OPJ_BOOL (*) ( void * p_codec, OPJ_UINT32 num_threads )) opj_jp2_set_threads;
+
 			l_codec->m_codec = opj_jp2_create(OPJ_TRUE);
 
 			if (! l_codec->m_codec) {
@@ -354,6 +360,18 @@ void OPJ_CALLCONV opj_set_default_decoder_parameters(opj_dparameters_t *paramete
 	}
 }
 
+
+OPJ_API OPJ_CALLCONV opj_codec_set_threads(opj_codec_t *p_codec,
+                                                    int num_threads)
+{
+  if (p_codec ) { 
+        opj_codec_private_t * l_codec = (opj_codec_private_t *) p_codec;
+
+        return l_codec->opj_set_threads(l_codec->m_codec, num_threads);
+    }
+    return OPJ_FALSE;
+}
+
 OPJ_BOOL OPJ_CALLCONV opj_setup_decoder(opj_codec_t *p_codec,
                                         opj_dparameters_t *parameters 
 										)

src/lib/openjp2/openjpeg.h

@@ -1262,6 +1262,25 @@ OPJ_API void OPJ_CALLCONV opj_set_default_decoder_parameters(opj_dparameters_t *
 OPJ_API OPJ_BOOL OPJ_CALLCONV opj_setup_decoder(opj_codec_t *p_codec,
 												opj_dparameters_t *parameters );
 
+/**
+ * Allocates worker threads for the compressor/decompressor.
+ *
+ * By default, only the main thread is used. If this function is not used,
+ * but the OPJ_NUM_THREADS environment variable is set, its value will be
+ * used to initialize the number of threads. The value can be either an integer
+ * number, or "ALL_CPUS". If OPJ_NUM_THREADS is set and this function is called,
+ * this function will override the behaviour of the environment variable.
+ *
+ * Note: currently only has effect on the decompressor.
+ *
+ * @param p_codec       decompressor handler
+ * @param num_threads   number of threads.
+ *
+ * @return OPJ_TRUE     if the decoder is correctly set
+ */
+OPJ_API OPJ_BOOL OPJ_CALLCONV opj_codec_set_threads(opj_codec_t *p_codec,
+                                                    int num_threads);
+
 /**
  * Decodes an image header.
  *

src/lib/openjp2/opj_codec.h

@@ -113,6 +113,7 @@ typedef struct opj_codec_private
             OPJ_BOOL (*opj_set_decoded_resolution_factor) ( void * p_codec,
                                                             OPJ_UINT32 res_factor,
                                                             opj_event_mgr_t * p_manager);
+
         } m_decompression;
 
         /**
@@ -157,6 +158,9 @@ typedef struct opj_codec_private
     void (*opj_dump_codec) (void * p_codec, OPJ_INT32 info_flag, FILE* output_stream);
     opj_codestream_info_v2_t* (*opj_get_codec_info)(void* p_codec);
     opj_codestream_index_t* (*opj_get_codec_index)(void* p_codec);
+
+    /** Set number of threads */
+    OPJ_BOOL (*opj_set_threads) ( void * p_codec, OPJ_UINT32 num_threads );
 }
 opj_codec_private_t;
 

src/lib/openjp2/tcd.c

@@ -580,7 +580,8 @@ OPJ_BOOL opj_tcd_rateallocate(  opj_tcd_t *tcd,
 
 OPJ_BOOL opj_tcd_init( opj_tcd_t *p_tcd,
                                            opj_image_t * p_image,
-                                           opj_cp_t * p_cp )
+                                           opj_cp_t * p_cp,
+                       opj_thread_pool_t* p_tp )
 {
         p_tcd->image = p_image;
         p_tcd->cp = p_cp;
@@ -597,6 +598,7 @@ OPJ_BOOL opj_tcd_init( opj_tcd_t *p_tcd,
 
         p_tcd->tcd_image->tiles->numcomps = p_image->numcomps;
         p_tcd->tp_pos = p_cp->m_specific_param.m_enc.m_tp_pos;
+        p_tcd->thread_pool = p_tp;
 
         return OPJ_TRUE;
 }

src/lib/openjp2/tcd.h

@@ -220,6 +220,8 @@ typedef struct opj_tcd
 	OPJ_UINT32 tcd_tileno;
 	/** tell if the tcd is a decoder. */
 	OPJ_UINT32 m_is_decoder : 1;
+    /** Thread pool */
+    opj_thread_pool_t* thread_pool;
 } opj_tcd_t;
 
 /** @name Exported functions */
@@ -249,12 +251,14 @@ void opj_tcd_destroy(opj_tcd_t *tcd);
  * @param	p_tcd		TCD handle.
  * @param	p_image		raw image.
  * @param	p_cp		coding parameters.
+ * @param   p_tp        thread pool
  *
  * @return true if the encoding values could be set (false otherwise).
 */
 OPJ_BOOL opj_tcd_init(	opj_tcd_t *p_tcd,
 						opj_image_t * p_image,
-						opj_cp_t * p_cp );
+						opj_cp_t * p_cp,
+                        opj_thread_pool_t* p_tp);
 
 /**
  * Allocates memory for decoding a specific tile.