c_cpp/lib/htslib/thread_pool_internal.h

   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