diff options
Diffstat (limited to 'threadpool.c')
| -rw-r--r-- | threadpool.c | 422 |
1 files changed, 422 insertions, 0 deletions
diff --git a/threadpool.c b/threadpool.c new file mode 100644 index 000000000000..a973a143e9e2 --- /dev/null +++ b/threadpool.c @@ -0,0 +1,422 @@ +/* + * Copyright 2016 Jakub Klama <jceel@FreeBSD.org> + * All rights reserved + * + * Redistribution and use in source and binary forms, with or without + * modification, are permitted providing 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 AUTHOR ``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 AUTHOR 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 <errno.h> +#include <stdlib.h> +#include <pthread.h> +#if defined(__FreeBSD__) +#include <pthread_np.h> +#endif +#include <sys/queue.h> +#include "lib9p.h" +#include "threadpool.h" + +static void l9p_threadpool_rflush(struct l9p_threadpool *tp, + struct l9p_request *req); + +static void * +l9p_responder(void *arg) +{ + struct l9p_threadpool *tp; + struct l9p_worker *worker = arg; + struct l9p_request *req; + + tp = worker->ltw_tp; + for (;;) { + /* get next reply to send */ + pthread_mutex_lock(&tp->ltp_mtx); + while (STAILQ_EMPTY(&tp->ltp_replyq) && !worker->ltw_exiting) + pthread_cond_wait(&tp->ltp_reply_cv, &tp->ltp_mtx); + if (worker->ltw_exiting) { + pthread_mutex_unlock(&tp->ltp_mtx); + break; + } + + /* off reply queue */ + req = STAILQ_FIRST(&tp->ltp_replyq); + STAILQ_REMOVE_HEAD(&tp->ltp_replyq, lr_worklink); + + /* request is now in final glide path, can't be Tflush-ed */ + req->lr_workstate = L9P_WS_REPLYING; + + /* any flushers waiting for this request can go now */ + if (req->lr_flushstate != L9P_FLUSH_NONE) + l9p_threadpool_rflush(tp, req); + + pthread_mutex_unlock(&tp->ltp_mtx); + + /* send response */ + l9p_respond(req, false, true); + } + return (NULL); +} + +static void * +l9p_worker(void *arg) +{ + struct l9p_threadpool *tp; + struct l9p_worker *worker = arg; + struct l9p_request *req; + + tp = worker->ltw_tp; + pthread_mutex_lock(&tp->ltp_mtx); + for (;;) { + while (STAILQ_EMPTY(&tp->ltp_workq) && !worker->ltw_exiting) + pthread_cond_wait(&tp->ltp_work_cv, &tp->ltp_mtx); + if (worker->ltw_exiting) + break; + + /* off work queue; now work-in-progress, by us */ + req = STAILQ_FIRST(&tp->ltp_workq); + STAILQ_REMOVE_HEAD(&tp->ltp_workq, lr_worklink); + req->lr_workstate = L9P_WS_INPROGRESS; + req->lr_worker = worker; + pthread_mutex_unlock(&tp->ltp_mtx); + + /* actually try the request */ + req->lr_error = l9p_dispatch_request(req); + + /* move to responder queue, updating work-state */ + pthread_mutex_lock(&tp->ltp_mtx); + req->lr_workstate = L9P_WS_RESPQUEUED; + req->lr_worker = NULL; + STAILQ_INSERT_TAIL(&tp->ltp_replyq, req, lr_worklink); + + /* signal the responder */ + pthread_cond_signal(&tp->ltp_reply_cv); + } + pthread_mutex_unlock(&tp->ltp_mtx); + return (NULL); +} + +/* + * Just before finally replying to a request that got touched by + * a Tflush request, we enqueue its flushers (requests of type + * Tflush, which are now on the flushee's lr_flushq) onto the + * response queue. + */ +static void +l9p_threadpool_rflush(struct l9p_threadpool *tp, struct l9p_request *req) +{ + struct l9p_request *flusher; + + /* + * https://swtch.com/plan9port/man/man9/flush.html says: + * + * "Should multiple Tflushes be received for a pending + * request, they must be answered in order. A Rflush for + * any of the multiple Tflushes implies an answer for all + * previous ones. Therefore, should a server receive a + * request and then multiple flushes for that request, it + * need respond only to the last flush." This means + * we could march through the queue of flushers here, + * marking all but the last one as "to be dropped" rather + * than "to be replied-to". + * + * However, we'll leave that for later, if ever -- it + * should be harmless to respond to each, in order. + */ + STAILQ_FOREACH(flusher, &req->lr_flushq, lr_flushlink) { + flusher->lr_workstate = L9P_WS_RESPQUEUED; +#ifdef notdef + if (not the last) { + flusher->lr_flushstate = L9P_FLUSH_NOT_RUN; + /* or, flusher->lr_drop = true ? */ + } +#endif + STAILQ_INSERT_TAIL(&tp->ltp_replyq, flusher, lr_worklink); + } +} + +int +l9p_threadpool_init(struct l9p_threadpool *tp, int size) +{ + struct l9p_worker *worker; +#if defined(__FreeBSD__) + char threadname[16]; +#endif + int error; + int i, nworkers, nresponders; + + if (size <= 0) + return (EINVAL); + error = pthread_mutex_init(&tp->ltp_mtx, NULL); + if (error) + return (error); + error = pthread_cond_init(&tp->ltp_work_cv, NULL); + if (error) + goto fail_work_cv; + error = pthread_cond_init(&tp->ltp_reply_cv, NULL); + if (error) + goto fail_reply_cv; + + STAILQ_INIT(&tp->ltp_workq); + STAILQ_INIT(&tp->ltp_replyq); + LIST_INIT(&tp->ltp_workers); + + nresponders = 0; + nworkers = 0; + for (i = 0; i <= size; i++) { + worker = calloc(1, sizeof(struct l9p_worker)); + worker->ltw_tp = tp; + worker->ltw_responder = i == 0; + error = pthread_create(&worker->ltw_thread, NULL, + worker->ltw_responder ? l9p_responder : l9p_worker, + (void *)worker); + if (error) { + free(worker); + break; + } + if (worker->ltw_responder) + nresponders++; + else + nworkers++; + +#if defined(__FreeBSD__) + if (worker->ltw_responder) { + pthread_set_name_np(worker->ltw_thread, "9p-responder"); + } else { + sprintf(threadname, "9p-worker:%d", i - 1); + pthread_set_name_np(worker->ltw_thread, threadname); + } +#endif + + LIST_INSERT_HEAD(&tp->ltp_workers, worker, ltw_link); + } + if (nresponders == 0 || nworkers == 0) { + /* need the one responder, and at least one worker */ + l9p_threadpool_shutdown(tp); + return (error); + } + return (0); + + /* + * We could avoid these labels by having multiple destroy + * paths (one for each error case), or by having booleans + * for which variables were initialized. Neither is very + * appealing... + */ +fail_reply_cv: + pthread_cond_destroy(&tp->ltp_work_cv); +fail_work_cv: + pthread_mutex_destroy(&tp->ltp_mtx); + + return (error); +} + +/* + * Run a request, usually by queueing it. + */ +void +l9p_threadpool_run(struct l9p_threadpool *tp, struct l9p_request *req) +{ + + /* + * Flush requests must be handled specially, since they + * can cancel / kill off regular requests. (But we can + * run them through the regular dispatch mechanism.) + */ + if (req->lr_req.hdr.type == L9P_TFLUSH) { + /* not on a work queue yet so we can touch state */ + req->lr_workstate = L9P_WS_IMMEDIATE; + (void) l9p_dispatch_request(req); + } else { + pthread_mutex_lock(&tp->ltp_mtx); + req->lr_workstate = L9P_WS_NOTSTARTED; + STAILQ_INSERT_TAIL(&tp->ltp_workq, req, lr_worklink); + pthread_cond_signal(&tp->ltp_work_cv); + pthread_mutex_unlock(&tp->ltp_mtx); + } +} + +/* + * Run a Tflush request. Called via l9p_dispatch_request() since + * it has some debug code in it, but not called from worker thread. + */ +int +l9p_threadpool_tflush(struct l9p_request *req) +{ + struct l9p_connection *conn; + struct l9p_threadpool *tp; + struct l9p_request *flushee; + uint16_t oldtag; + enum l9p_flushstate nstate; + + /* + * Find what we're supposed to flush (the flushee, as it were). + */ + req->lr_error = 0; /* Tflush always succeeds */ + conn = req->lr_conn; + tp = &conn->lc_tp; + oldtag = req->lr_req.tflush.oldtag; + ht_wrlock(&conn->lc_requests); + flushee = ht_find_locked(&conn->lc_requests, oldtag); + if (flushee == NULL) { + /* + * Nothing to flush! The old request must have + * been done and gone already. Just queue this + * Tflush for a success reply. + */ + ht_unlock(&conn->lc_requests); + pthread_mutex_lock(&tp->ltp_mtx); + goto done; + } + + /* + * Found the original request. We'll need to inspect its + * work-state to figure out what to do. + */ + pthread_mutex_lock(&tp->ltp_mtx); + ht_unlock(&conn->lc_requests); + + switch (flushee->lr_workstate) { + + case L9P_WS_NOTSTARTED: + /* + * Flushee is on work queue, but not yet being + * handled by a worker. + * + * The documentation -- see + * http://ericvh.github.io/9p-rfc/rfc9p2000.html + * https://swtch.com/plan9port/man/man9/flush.html + * -- says that "the server should answer the + * flush message immediately". However, Linux + * sends flush requests for operations that + * must finish, such as Tclunk, and it's not + * possible to *answer* the flush request until + * it has been handled (if necessary) or aborted + * (if allowed). + * + * We therefore now just the original request + * and let the request-handler do whatever is + * appropriate. NOTE: we could have a table of + * "requests that can be aborted without being + * run" vs "requests that must be run to be + * aborted", but for now that seems like an + * unnecessary complication. + */ + nstate = L9P_FLUSH_REQUESTED_PRE_START; + break; + + case L9P_WS_IMMEDIATE: + /* + * This state only applies to Tflush requests, and + * flushing a Tflush is illegal. But we'll do nothing + * special here, which will make us act like a flush + * request for the flushee that arrived too late to + * do anything about the flushee. + */ + nstate = L9P_FLUSH_REQUESTED_POST_START; + break; + + case L9P_WS_INPROGRESS: + /* + * Worker thread flushee->lr_worker is working on it. + * Kick it to get it out of blocking system calls. + * (This requires that it carefully set up some + * signal handlers, and may be FreeBSD-dependent, + * it probably cannot be handled this way on MacOS.) + */ +#ifdef notyet + pthread_kill(...); +#endif + nstate = L9P_FLUSH_REQUESTED_POST_START; + break; + + case L9P_WS_RESPQUEUED: + /* + * The flushee is already in the response queue. + * We'll just mark it as having had some flush + * action applied. + */ + nstate = L9P_FLUSH_TOOLATE; + break; + + case L9P_WS_REPLYING: + /* + * Although we found the flushee, it's too late to + * make us depend on it: it's already heading out + * the door as a reply. + * + * We don't want to do anything to the flushee. + * Instead, we want to work the same way as if + * we had never found the tag. + */ + goto done; + } + + /* + * Now add us to the list of Tflush-es that are waiting + * for the flushee (creating the list if needed, i.e., if + * this is the first Tflush for the flushee). We (req) + * will get queued for reply later, when the responder + * processes the flushee and calls l9p_threadpool_rflush(). + */ + if (flushee->lr_flushstate == L9P_FLUSH_NONE) + STAILQ_INIT(&flushee->lr_flushq); + flushee->lr_flushstate = nstate; + STAILQ_INSERT_TAIL(&flushee->lr_flushq, req, lr_flushlink); + + pthread_mutex_unlock(&tp->ltp_mtx); + + return (0); + +done: + /* + * This immediate op is ready to be replied-to now, so just + * stick it onto the reply queue. + */ + req->lr_workstate = L9P_WS_RESPQUEUED; + STAILQ_INSERT_TAIL(&tp->ltp_replyq, req, lr_worklink); + pthread_mutex_unlock(&tp->ltp_mtx); + pthread_cond_signal(&tp->ltp_reply_cv); + return (0); +} + +int +l9p_threadpool_shutdown(struct l9p_threadpool *tp) +{ + struct l9p_worker *worker, *tmp; + + LIST_FOREACH_SAFE(worker, &tp->ltp_workers, ltw_link, tmp) { + pthread_mutex_lock(&tp->ltp_mtx); + worker->ltw_exiting = true; + if (worker->ltw_responder) + pthread_cond_signal(&tp->ltp_reply_cv); + else + pthread_cond_broadcast(&tp->ltp_work_cv); + pthread_mutex_unlock(&tp->ltp_mtx); + pthread_join(worker->ltw_thread, NULL); + LIST_REMOVE(worker, ltw_link); + free(worker); + } + pthread_cond_destroy(&tp->ltp_reply_cv); + pthread_cond_destroy(&tp->ltp_work_cv); + pthread_mutex_destroy(&tp->ltp_mtx); + + return (0); +} |