Implement local thread pools

Implement the ASYNC_JOB as a local thread pool. Remove the API support
for global pools.

Reviewed-by: Rich Salz <rsalz@openssl.org>

apps/s_server.c

@@ -1661,7 +1661,7 @@ int s_server_main(int argc, char *argv[])
 
     if (async) {
         SSL_CTX_set_mode(ctx, SSL_MODE_ASYNC);
-        ASYNC_init_pool(0, 0, 0);
+        ASYNC_init_pool(0, 0);
     }
 
 #ifndef OPENSSL_NO_SRTP

crypto/async/arch/async_posix.c

@@ -65,6 +65,10 @@ __thread ASYNC_CTX *sysvctx;
 
 #define STACKSIZE       32768
 
+__thread size_t pool_max_size = 0;
+__thread size_t pool_curr_size = 0;
+__thread STACK_OF(ASYNC_JOB) *pool = NULL;
+
 int ASYNC_FIBRE_init(ASYNC_FIBRE *fibre)
 {
     void *stack = NULL;
@@ -111,4 +115,42 @@ int async_read1(int fd, void *buf)
     return 0;
 }
 
+STACK_OF(ASYNC_JOB) *async_get_pool(void)
+{
+    return pool;
+}
+
+void async_set_pool(STACK_OF(ASYNC_JOB) *poolin, size_t curr_size,
+                    size_t max_size)
+{
+    pool = poolin;
+    pool_curr_size = curr_size;
+    pool_max_size = max_size;
+}
+
+void async_increment_pool_size(void)
+{
+    pool_curr_size++;
+}
+
+void async_release_job_to_pool(ASYNC_JOB *job)
+{
+    sk_ASYNC_JOB_push(pool, job);
+}
+
+size_t async_pool_max_size(void)
+{
+    return pool_max_size;
+}
+
+void async_release_pool(void)
+{
+    sk_ASYNC_JOB_free(pool);
+}
+
+int async_pool_can_grow(void)
+{
+    return (pool_max_size == 0) || (pool_curr_size < pool_max_size);
+}
+
 #endif

crypto/async/arch/async_posix.h

@@ -103,6 +103,5 @@ int async_pipe(int *pipefds);
 int async_write1(int fd, const void *buf);
 int async_read1(int fd, void *buf);
 
-
 # endif
 #endif

crypto/async/async.c

@@ -51,7 +51,6 @@
  * ====================================================================
  */
 
-#include <openssl/crypto.h>
 #include <openssl/async.h>
 #include <string.h>
 #include "async_locl.h"
@@ -61,13 +60,6 @@
 #define ASYNC_JOB_PAUSED    2
 #define ASYNC_JOB_STOPPING  3
 
-static size_t pool_max_size = 0;
-static size_t curr_size = 0;
-
-DECLARE_STACK_OF(ASYNC_JOB)
-static STACK_OF(ASYNC_JOB) *pool = NULL;
-
-
 static ASYNC_CTX *ASYNC_CTX_new(void)
 {
     ASYNC_CTX *nctx = NULL;
@@ -139,27 +131,29 @@ static void ASYNC_JOB_free(ASYNC_JOB *job)
 
 static ASYNC_JOB *async_get_pool_job(void) {
     ASYNC_JOB *job;
+    STACK_OF(ASYNC_JOB) *pool;
 
+    pool = async_get_pool();
     if (pool == NULL) {
         /*
          * Pool has not been initialised, so init with the defaults, i.e.
          * global pool, with no max size and no pre-created jobs
          */
-        if (ASYNC_init_pool(0, 0, 0) == 0)
+        if (ASYNC_init_pool(0, 0) == 0)
             return NULL;
+        pool = async_get_pool();
     }
 
     job = sk_ASYNC_JOB_pop(pool);
     if (job == NULL) {
         /* Pool is empty */
-        if (pool_max_size && curr_size >= pool_max_size) {
+        if (!async_pool_can_grow())
-            /* Pool is at max size. We cannot continue */
             return NULL;
-        }
+
         job = ASYNC_JOB_new();
         if (job) {
             ASYNC_FIBRE_makecontext(&job->fibrectx);
-            curr_size++;
+            async_increment_pool_size();
         }
     }
     return job;
@@ -170,7 +164,7 @@ static void async_release_job(ASYNC_JOB *job) {
         OPENSSL_free(job->funcargs);
     job->funcargs = NULL;
     /* Ignore error return */
-    sk_ASYNC_JOB_push(pool, job);
+    async_release_job_to_pool(job);
 }
 
 void ASYNC_start_func(void)
@@ -301,14 +295,14 @@ int ASYNC_in_job(void)
     return 0;
 }
 
-int ASYNC_init_pool(unsigned int local, size_t max_size, size_t init_size)
+int ASYNC_init_pool(size_t max_size, size_t init_size)
 {
-    if (local != 0) {
+    STACK_OF(ASYNC_JOB) *pool;
-        /* We only support a global pool so far */
+    size_t curr_size = 0;
-        return 0;
+
-    }
+    if (init_size > max_size)
+        return 0;
 
-    pool_max_size = max_size;
     pool = sk_ASYNC_JOB_new_null();
     if (pool == NULL) {
         return 0;
@@ -332,18 +326,24 @@ int ASYNC_init_pool(unsigned int local, size_t max_size, size_t init_size)
         }
     }
 
+    async_set_pool(pool, curr_size, max_size);
+
     return 1;
 }
 
 void ASYNC_free_pool(void)
 {
     ASYNC_JOB *job;
+    STACK_OF(ASYNC_JOB) *pool;
 
+    pool = async_get_pool();
+    if (pool == NULL)
+        return;
     do {
         job = sk_ASYNC_JOB_pop(pool);
         ASYNC_JOB_free(job);
     } while (job);
-    sk_ASYNC_JOB_free(pool);
+    async_release_pool();
 }
 
 ASYNC_JOB *ASYNC_get_current_job(void)

crypto/async/async_locl.h

@@ -52,6 +52,7 @@
  */
 
 #include <openssl/async.h>
+#include <openssl/crypto.h>
 
 typedef struct async_ctx_st ASYNC_CTX;
 
@@ -75,4 +76,14 @@ struct async_job_st {
     int wake_fd;
 };
 
+DECLARE_STACK_OF(ASYNC_JOB)
+
 void ASYNC_start_func(void);
+STACK_OF(ASYNC_JOB) *async_get_pool(void);
+void async_set_pool(STACK_OF(ASYNC_JOB) *poolin, size_t curr_size,
+                    size_t max_size);
+void async_increment_pool_size(void);
+void async_release_job_to_pool(ASYNC_JOB *job);
+size_t async_pool_max_size(void);
+void async_release_pool(void);
+int async_pool_can_grow(void);

include/openssl/async.h

@@ -67,7 +67,7 @@ typedef struct async_job_st ASYNC_JOB;
 #define ASYNC_PAUSE    2
 #define ASYNC_FINISH   3
 
-int ASYNC_init_pool(unsigned int local, size_t max_size, size_t init_size);
+int ASYNC_init_pool(size_t max_size, size_t init_size);
 void ASYNC_free_pool(void);
 
 int ASYNC_start_job(ASYNC_JOB **job, int *ret, int (*func)(void *),