28747907ce52c2f028a0ee69116f6213b2091e81
[openssl.git] / 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(*nctx));
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(*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 }