1 /* thread_pool_internal.h -- Internal API for the thread pool. 2 3 Copyright (c) 2013-2016 Genome Research Ltd. 4 5 Author: James Bonfield <jkb@sanger.ac.uk> 6 7 Permission is hereby granted, free of charge, to any person obtaining a copy 8 of this software and associated documentation files (the "Software"), to deal 9 in the Software without restriction, including without limitation the rights 10 to use, copy, modify, merge, publish, distribute, sublicense, and/or sell 11 copies of the Software, and to permit persons to whom the Software is 12 furnished to do so, subject to the following conditions: 13 14 The above copyright notice and this permission notice shall be included in 15 all copies or substantial portions of the Software. 16 17 THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR 18 IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, 19 FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL 20 THE AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER 21 LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING 22 FROM, OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER 23 DEALINGS IN THE SOFTWARE. */ 24 25 /* 26 * This file implements a thread pool for multi-threading applications. 27 * It consists of two distinct interfaces: thread pools an thread job queues. 28 * 29 * The pool of threads is given a function pointer and void* data to pass in. 30 * This means the pool can run jobs of multiple types, albeit first come 31 * first served with no job scheduling except to pick tasks from 32 * queues that have room to store the result. 33 * 34 * Upon completion, the return value from the function pointer is 35 * added to back to the queue if the result is required. We may have 36 * multiple queues in use for the one pool. 37 * 38 * To see example usage, please look at the #ifdef TEST_MAIN code in 39 * thread_pool.c. 40 */ 41 42 #ifndef THREAD_POOL_INTERNAL_H 43 #define THREAD_POOL_INTERNAL_H 44 45 #include <pthread.h> 46 #include <stdint.h> 47 #include "htslib/thread_pool.h" 48 49 #ifdef __cplusplus 50 extern "C" { 51 #endif 52 53 /* 54 * An input job, before execution. 55 */ 56 typedef struct hts_tpool_job { 57 void *(*func)(void *arg); 58 void *arg; 59 void (*job_cleanup)(void *arg); 60 void (*result_cleanup)(void *data); 61 struct hts_tpool_job *next; 62 63 struct hts_tpool *p; 64 struct hts_tpool_process *q; 65 uint64_t serial; 66 } hts_tpool_job; 67 68 /* 69 * An output, after job has executed. 70 */ 71 struct hts_tpool_result { 72 struct hts_tpool_result *next; 73 void (*result_cleanup)(void *data); 74 uint64_t serial; // sequential number for ordering 75 void *data; // result itself 76 }; 77 78 /* 79 * A per-thread worker struct. 80 */ 81 typedef struct { 82 struct hts_tpool *p; 83 int idx; 84 pthread_t tid; 85 pthread_cond_t pending_c; // when waiting for a job 86 } hts_tpool_worker; 87 88 /* 89 * An IO queue consists of a queue of jobs to execute 90 * (the "input" side) and a queue of job results post- 91 * execution (the "output" side). 92 * 93 * We have size limits to prevent either queue from 94 * growing too large and serial numbers to ensure 95 * sequential consumption of the output. 96 * 97 * The thread pool may have many hetergeneous tasks, each 98 * using its own io_queue mixed into the same thread pool. 99 */ 100 struct hts_tpool_process { 101 struct hts_tpool *p; // thread pool 102 hts_tpool_job *input_head; // input list 103 hts_tpool_job *input_tail; 104 hts_tpool_result *output_head; // output list 105 hts_tpool_result *output_tail; 106 int qsize; // max size of i/o queues 107 uint64_t next_serial; // next serial for output 108 uint64_t curr_serial; // current serial (next input) 109 110 int no_more_input; // disable dispatching of more jobs 111 int n_input; // no. items in input queue; was njobs 112 int n_output; // no. items in output queue 113 int n_processing; // no. items being processed (executing) 114 115 int shutdown; // true if pool is being destroyed 116 int in_only; // if true, don't queue result up. 117 int wake_dispatch; // unblocks waiting dispatchers 118 119 int ref_count; // used to track safe destruction 120 121 pthread_cond_t output_avail_c; // Signalled on each new output 122 pthread_cond_t input_not_full_c; // Input queue is no longer full 123 pthread_cond_t input_empty_c; // Input queue has become empty 124 pthread_cond_t none_processing_c;// n_processing has hit zero 125 126 struct hts_tpool_process *next, *prev;// to form circular linked list. 127 }; 128 129 /* 130 * The single pool structure itself. 131 * 132 * This knows nothing about the nature of the jobs or where their 133 * output is going, but it maintains a list of queues associated with 134 * this pool from which the jobs are taken. 135 */ 136 struct hts_tpool { 137 int nwaiting; // how many workers waiting for new jobs 138 int njobs; // how many total jobs are waiting in all queues 139 int shutdown; // true if pool is being destroyed 140 141 // I/O queues to check for jobs in and to put results. 142 // Forms a circular linked list. (q_head may be amended 143 // to point to the most recently updated.) 144 hts_tpool_process *q_head; 145 146 // threads 147 int tsize; // maximum number of jobs 148 hts_tpool_worker *t; 149 // array of worker IDs free 150 int *t_stack, t_stack_top; 151 152 // A single mutex used when updating this and any associated structure. 153 pthread_mutex_t pool_m; 154 155 // Tracking of average number of running jobs. 156 // This can be used to dampen any hysteresis caused by bursty 157 // input availability. 158 int n_count, n_running; 159 160 // Debugging to check wait time. 161 // FIXME: should we just delete these and cull the associated code? 162 long long total_time, wait_time; 163 }; 164 165 #ifdef __cplusplus 166 } 167 #endif 168 169 #endif 170