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 struct hts_tpool_job *next; 60 61 struct hts_tpool *p; 62 struct hts_tpool_process *q; 63 uint64_t serial; 64 } hts_tpool_job; 65 66 /* 67 * An output, after job has executed. 68 */ 69 struct hts_tpool_result { 70 struct hts_tpool_result *next; 71 uint64_t serial; // sequential number for ordering 72 void *data; // result itself 73 }; 74 75 /* 76 * A per-thread worker struct. 77 */ 78 typedef struct { 79 struct hts_tpool *p; 80 int idx; 81 pthread_t tid; 82 pthread_cond_t pending_c; // when waiting for a job 83 } hts_tpool_worker; 84 85 /* 86 * An IO queue consists of a queue of jobs to execute 87 * (the "input" side) and a queue of job results post- 88 * execution (the "output" side). 89 * 90 * We have size limits to prevent either queue from 91 * growing too large and serial numbers to ensure 92 * sequential consumption of the output. 93 * 94 * The thread pool may have many hetergeneous tasks, each 95 * using its own io_queue mixed into the same thread pool. 96 */ 97 struct hts_tpool_process { 98 struct hts_tpool *p; // thread pool 99 hts_tpool_job *input_head; // input list 100 hts_tpool_job *input_tail; 101 hts_tpool_result *output_head; // output list 102 hts_tpool_result *output_tail; 103 int qsize; // max size of i/o queues 104 uint64_t next_serial; // next serial for output 105 uint64_t curr_serial; // current serial (next input) 106 107 int n_input; // no. items in input queue; was njobs 108 int n_output; // no. items in output queue 109 int n_processing; // no. items being processed (executing) 110 111 int shutdown; // true if pool is being destroyed 112 int in_only; // if true, don't queue result up. 113 int wake_dispatch; // unblocks waiting dispatchers 114 115 int ref_count; // used to track safe destruction 116 117 pthread_cond_t output_avail_c; // Signalled on each new output 118 pthread_cond_t input_not_full_c; // Input queue is no longer full 119 pthread_cond_t input_empty_c; // Input queue has become empty 120 pthread_cond_t none_processing_c;// n_processing has hit zero 121 122 struct hts_tpool_process *next, *prev;// to form circular linked list. 123 }; 124 125 /* 126 * The single pool structure itself. 127 * 128 * This knows nothing about the nature of the jobs or where their 129 * output is going, but it maintains a list of queues associated with 130 * this pool from which the jobs are taken. 131 */ 132 struct hts_tpool { 133 int nwaiting; // how many workers waiting for new jobs 134 int njobs; // how many total jobs are waiting in all queues 135 int shutdown; // true if pool is being destroyed 136 137 // I/O queues to check for jobs in and to put results. 138 // Forms a circular linked list. (q_head may be amended 139 // to point to the most recently updated.) 140 hts_tpool_process *q_head; 141 142 // threads 143 int tsize; // maximum number of jobs 144 hts_tpool_worker *t; 145 // array of worker IDs free 146 int *t_stack, t_stack_top; 147 148 // A single mutex used when updating this and any associated structure. 149 pthread_mutex_t pool_m; 150 151 // Tracking of average number of running jobs. 152 // This can be used to dampen any hysteresis caused by bursty 153 // input availability. 154 int n_count, n_running; 155 156 // Debugging to check wait time. 157 // FIXME: should we just delete these and cull the associated code? 158 long long total_time, wait_time; 159 }; 160 161 #ifdef __cplusplus 162 } 163 #endif 164 165 #endif 166