crypto/async/async.c

   1 /*
   2  * Copyright 2015-2016 The OpenSSL Project Authors. All Rights Reserved.
   3  *
   4  * Licensed under the OpenSSL license (the "License").  You may not use
   5  * this file except in compliance with the License.  You can obtain a copy
   6  * in the file LICENSE in the source distribution or at
   7  * https://www.openssl.org/source/license.html
   8  */
   9
  10 /*
  11  * Without this we start getting longjmp crashes because it thinks we're jumping
  12  * up the stack when in fact we are jumping to an entirely different stack. The
  13  * cost of this is not having certain buffer overrun/underrun checks etc for
  14  * this source file :-(
  15  */
  16 #undef _FORTIFY_SOURCE
  17
  18 /* This must be the first #include file */
  19 #include "async_locl.h"
  20
  21 #include <openssl/err.h>
  22 #include "internal/cryptlib_int.h"
  23 #include <string.h>
  24
  25 #define ASYNC_JOB_RUNNING   0
  26 #define ASYNC_JOB_PAUSING   1
  27 #define ASYNC_JOB_PAUSED    2
  28 #define ASYNC_JOB_STOPPING  3
  29
  30 static CRYPTO_THREAD_LOCAL ctxkey;
  31 static CRYPTO_THREAD_LOCAL poolkey;
  32
  33 static void async_free_pool_internal(async_pool *pool);
  34
  35 static async_ctx *async_ctx_new(void)
  36 {
  37     async_ctx *nctx = NULL;
  38
  39     nctx = OPENSSL_malloc(sizeof (async_ctx));
  40     if (nctx == NULL) {
  41         ASYNCerr(ASYNC_F_ASYNC_CTX_NEW, ERR_R_MALLOC_FAILURE);
  42         goto err;
  43     }
  44
  45     async_fibre_init_dispatcher(&nctx->dispatcher);
  46     nctx->currjob = NULL;
  47     nctx->blocked = 0;
  48     if (!CRYPTO_THREAD_set_local(&ctxkey, nctx))
  49         goto err;
  50
  51     return nctx;
  52 err:
  53     OPENSSL_free(nctx);
  54
  55     return NULL;
  56 }
  57
  58 async_ctx *async_get_ctx(void)
  59 {
  60     if (!OPENSSL_init_crypto(OPENSSL_INIT_ASYNC, NULL))
  61         return NULL;
  62
  63     return (async_ctx *)CRYPTO_THREAD_get_local(&ctxkey);
  64 }
  65
  66 static int async_ctx_free(void)
  67 {
  68     async_ctx *ctx;
  69
  70     ctx = async_get_ctx();
  71
  72     if (!CRYPTO_THREAD_set_local(&ctxkey, NULL))
  73         return 0;
  74
  75     OPENSSL_free(ctx);
  76
  77     return 1;
  78 }
  79
  80 static ASYNC_JOB *async_job_new(void)
  81 {
  82     ASYNC_JOB *job = NULL;
  83
  84     job = OPENSSL_zalloc(sizeof (ASYNC_JOB));
  85     if (job == NULL) {
  86         ASYNCerr(ASYNC_F_ASYNC_JOB_NEW, ERR_R_MALLOC_FAILURE);
  87         return NULL;
  88     }
  89
  90     job->status = ASYNC_JOB_RUNNING;
  91
  92     return job;
  93 }
  94
  95 static void async_job_free(ASYNC_JOB *job)
  96 {
  97     if (job != NULL) {
  98         OPENSSL_free(job->funcargs);
  99         async_fibre_free(&job->fibrectx);
 100         OPENSSL_free(job);
 101     }
 102 }
 103
 104 static ASYNC_JOB *async_get_pool_job(void) {
 105     ASYNC_JOB *job;
 106     async_pool *pool;
 107
 108     pool = (async_pool *)CRYPTO_THREAD_get_local(&poolkey);
 109     if (pool == NULL) {
 110         /*
 111          * Pool has not been initialised, so init with the defaults, i.e.
 112          * no max size and no pre-created jobs
 113          */
 114         if (ASYNC_init_thread(0, 0) == 0)
 115             return NULL;
 116         pool = (async_pool *)CRYPTO_THREAD_get_local(&poolkey);
 117     }
 118
 119     job = sk_ASYNC_JOB_pop(pool->jobs);
 120     if (job == NULL) {
 121         /* Pool is empty */
 122         if ((pool->max_size != 0) && (pool->curr_size >= pool->max_size))
 123             return NULL;
 124
 125         job = async_job_new();
 126         if (job != NULL) {
 127             if (! async_fibre_makecontext(&job->fibrectx)) {
 128                 async_job_free(job);
 129                 return NULL;
 130             }
 131             pool->curr_size++;
 132         }
 133     }
 134     return job;
 135 }
 136
 137 static void async_release_job(ASYNC_JOB *job) {
 138     async_pool *pool;
 139
 140     pool = (async_pool *)CRYPTO_THREAD_get_local(&poolkey);
 141     OPENSSL_free(job->funcargs);
 142     job->funcargs = NULL;
 143     sk_ASYNC_JOB_push(pool->jobs, job);
 144 }
 145
 146 void async_start_func(void)
 147 {
 148     ASYNC_JOB *job;
 149     async_ctx *ctx = async_get_ctx();
 150
 151     while (1) {
 152         /* Run the job */
 153         job = ctx->currjob;
 154         job->ret = job->func(job->funcargs);
 155
 156         /* Stop the job */
 157         job->status = ASYNC_JOB_STOPPING;
 158         if (!async_fibre_swapcontext(&job->fibrectx,
 159                                      &ctx->dispatcher, 1)) {
 160             /*
 161              * Should not happen. Getting here will close the thread...can't do
 162              * much about it
 163              */
 164             ASYNCerr(ASYNC_F_ASYNC_START_FUNC, ASYNC_R_FAILED_TO_SWAP_CONTEXT);
 165         }
 166     }
 167 }
 168
 169 int ASYNC_start_job(ASYNC_JOB **job, ASYNC_WAIT_CTX *wctx, int *ret,
 170                     int (*func)(void *), void *args, size_t size)
 171 {
 172     async_ctx *ctx = async_get_ctx();
 173     if (ctx == NULL)
 174         ctx = async_ctx_new();
 175     if (ctx == NULL) {
 176         return ASYNC_ERR;
 177     }
 178
 179     if (*job) {
 180         ctx->currjob = *job;
 181     }
 182
 183     for (;;) {
 184         if (ctx->currjob != NULL) {
 185             if (ctx->currjob->status == ASYNC_JOB_STOPPING) {
 186                 *ret = ctx->currjob->ret;
 187                 ctx->currjob->waitctx = NULL;
 188                 async_release_job(ctx->currjob);
 189                 ctx->currjob = NULL;
 190                 *job = NULL;
 191                 return ASYNC_FINISH;
 192             }
 193
 194             if (ctx->currjob->status == ASYNC_JOB_PAUSING) {
 195                 *job = ctx->currjob;
 196                 ctx->currjob->status = ASYNC_JOB_PAUSED;
 197                 ctx->currjob = NULL;
 198                 return ASYNC_PAUSE;
 199             }
 200
 201             if (ctx->currjob->status == ASYNC_JOB_PAUSED) {
 202                 ctx->currjob = *job;
 203                 /* Resume previous job */
 204                 if (!async_fibre_swapcontext(&ctx->dispatcher,
 205                         &ctx->currjob->fibrectx, 1)) {
 206                     ASYNCerr(ASYNC_F_ASYNC_START_JOB,
 207                              ASYNC_R_FAILED_TO_SWAP_CONTEXT);
 208                     goto err;
 209                 }
 210                 continue;
 211             }
 212
 213             /* Should not happen */
 214             ASYNCerr(ASYNC_F_ASYNC_START_JOB, ERR_R_INTERNAL_ERROR);
 215             async_release_job(ctx->currjob);
 216             ctx->currjob = NULL;
 217             *job = NULL;
 218             return ASYNC_ERR;
 219         }
 220
 221         /* Start a new job */
 222         if ((ctx->currjob = async_get_pool_job()) == NULL) {
 223             return ASYNC_NO_JOBS;
 224         }
 225
 226         if (args != NULL) {
 227             ctx->currjob->funcargs = OPENSSL_malloc(size);
 228             if (ctx->currjob->funcargs == NULL) {
 229                 ASYNCerr(ASYNC_F_ASYNC_START_JOB, ERR_R_MALLOC_FAILURE);
 230                 async_release_job(ctx->currjob);
 231                 ctx->currjob = NULL;
 232                 return ASYNC_ERR;
 233             }
 234             memcpy(ctx->currjob->funcargs, args, size);
 235         } else {
 236             ctx->currjob->funcargs = NULL;
 237         }
 238
 239         ctx->currjob->func = func;
 240         ctx->currjob->waitctx = wctx;
 241         if (!async_fibre_swapcontext(&ctx->dispatcher,
 242                 &ctx->currjob->fibrectx, 1)) {
 243             ASYNCerr(ASYNC_F_ASYNC_START_JOB, ASYNC_R_FAILED_TO_SWAP_CONTEXT);
 244             goto err;
 245         }
 246     }
 247
 248 err:
 249     async_release_job(ctx->currjob);
 250     ctx->currjob = NULL;
 251     *job = NULL;
 252     return ASYNC_ERR;
 253 }
 254
 255 int ASYNC_pause_job(void)
 256 {
 257     ASYNC_JOB *job;
 258     async_ctx *ctx = async_get_ctx();
 259
 260     if (ctx == NULL
 261             || ctx->currjob == NULL
 262             || ctx->blocked) {
 263         /*
 264          * Could be we've deliberately not been started within a job so this is
 265          * counted as success.
 266          */
 267         return 1;
 268     }
 269
 270     job = ctx->currjob;
 271     job->status = ASYNC_JOB_PAUSING;
 272
 273     if (!async_fibre_swapcontext(&job->fibrectx,
 274                                  &ctx->dispatcher, 1)) {
 275         ASYNCerr(ASYNC_F_ASYNC_PAUSE_JOB, ASYNC_R_FAILED_TO_SWAP_CONTEXT);
 276         return 0;
 277     }
 278     /* Reset counts of added and deleted fds */
 279     async_wait_ctx_reset_counts(job->waitctx);
 280
 281     return 1;
 282 }
 283
 284 static void async_empty_pool(async_pool *pool)
 285 {
 286     ASYNC_JOB *job;
 287
 288     if (!pool || !pool->jobs)
 289         return;
 290
 291     do {
 292         job = sk_ASYNC_JOB_pop(pool->jobs);
 293         async_job_free(job);
 294     } while (job);
 295 }
 296
 297 int async_init(void)
 298 {
 299     if (!CRYPTO_THREAD_init_local(&ctxkey, NULL))
 300         return 0;
 301
 302     if (!CRYPTO_THREAD_init_local(&poolkey, NULL)) {
 303         CRYPTO_THREAD_cleanup_local(&ctxkey);
 304         return 0;
 305     }
 306
 307     return 1;
 308 }
 309
 310 void async_deinit(void)
 311 {
 312     CRYPTO_THREAD_cleanup_local(&ctxkey);
 313     CRYPTO_THREAD_cleanup_local(&poolkey);
 314 }
 315
 316 int ASYNC_init_thread(size_t max_size, size_t init_size)
 317 {
 318     async_pool *pool;
 319     size_t curr_size = 0;
 320
 321     if (init_size > max_size) {
 322         ASYNCerr(ASYNC_F_ASYNC_INIT_THREAD, ASYNC_R_INVALID_POOL_SIZE);
 323         return 0;
 324     }
 325
 326     if (!OPENSSL_init_crypto(OPENSSL_INIT_ASYNC, NULL)) {
 327         return 0;
 328     }
 329     if (!ossl_init_thread_start(OPENSSL_INIT_THREAD_ASYNC)) {
 330         return 0;
 331     }
 332
 333     pool = OPENSSL_zalloc(sizeof *pool);
 334     if (pool == NULL) {
 335         ASYNCerr(ASYNC_F_ASYNC_INIT_THREAD, ERR_R_MALLOC_FAILURE);
 336         return 0;
 337     }
 338
 339     pool->jobs = sk_ASYNC_JOB_new_reserve(NULL, init_size);
 340     if (pool->jobs == NULL) {
 341         ASYNCerr(ASYNC_F_ASYNC_INIT_THREAD, ERR_R_MALLOC_FAILURE);
 342         OPENSSL_free(pool);
 343         return 0;
 344     }
 345
 346     pool->max_size = max_size;
 347
 348     /* Pre-create jobs as required */
 349     while (init_size--) {
 350         ASYNC_JOB *job;
 351         job = async_job_new();
 352         if (job == NULL || !async_fibre_makecontext(&job->fibrectx)) {
 353             /*
 354              * Not actually fatal because we already created the pool, just
 355              * skip creation of any more jobs
 356              */
 357             async_job_free(job);
 358             break;
 359         }
 360         job->funcargs = NULL;
 361         sk_ASYNC_JOB_push(pool->jobs, job); /* Cannot fail due to reserve */
 362         curr_size++;
 363     }
 364     pool->curr_size = curr_size;
 365     if (!CRYPTO_THREAD_set_local(&poolkey, pool)) {
 366         ASYNCerr(ASYNC_F_ASYNC_INIT_THREAD, ASYNC_R_FAILED_TO_SET_POOL);
 367         goto err;
 368     }
 369
 370     return 1;
 371 err:
 372     async_free_pool_internal(pool);
 373     return 0;
 374 }
 375
 376 static void async_free_pool_internal(async_pool *pool)
 377 {
 378     if (pool == NULL)
 379         return;
 380
 381     async_empty_pool(pool);
 382     sk_ASYNC_JOB_free(pool->jobs);
 383     OPENSSL_free(pool);
 384     CRYPTO_THREAD_set_local(&poolkey, NULL);
 385     async_local_cleanup();
 386     async_ctx_free();
 387 }
 388
 389 void ASYNC_cleanup_thread(void)
 390 {
 391     async_free_pool_internal((async_pool *)CRYPTO_THREAD_get_local(&poolkey));
 392 }
 393
 394 ASYNC_JOB *ASYNC_get_current_job(void)
 395 {
 396     async_ctx *ctx;
 397
 398     ctx = async_get_ctx();
 399     if (ctx == NULL)
 400         return NULL;
 401
 402     return ctx->currjob;
 403 }
 404
 405 ASYNC_WAIT_CTX *ASYNC_get_wait_ctx(ASYNC_JOB *job)
 406 {
 407     return job->waitctx;
 408 }
 409
 410 void ASYNC_block_pause(void)
 411 {
 412     async_ctx *ctx = async_get_ctx();
 413     if (ctx == NULL || ctx->currjob == NULL) {
 414         /*
 415          * We're not in a job anyway so ignore this
 416          */
 417         return;
 418     }
 419     ctx->blocked++;
 420 }
 421
 422 void ASYNC_unblock_pause(void)
 423 {
 424     async_ctx *ctx = async_get_ctx();
 425     if (ctx == NULL || ctx->currjob == NULL) {
 426         /*
 427          * We're not in a job anyway so ignore this
 428          */
 429         return;
 430     }
 431     if (ctx->blocked > 0)
 432         ctx->blocked--;
 433 }