lib/tp.c

   1 /*
   2  * Basic workqueue like code, that sets up a thread and allows async
   3  * processing of some sort. Could be extended to allow for multiple
   4  * worker threads. But right now fio associates one of this per IO
   5  * thread, so should be enough to have just a single thread doing the
   6  * work.
   7  */
   8 #include <stdio.h>
   9 #include <stdlib.h>
  10 #include <stdarg.h>
  11 #include <unistd.h>
  12 #include <errno.h>
  13 #include <pthread.h>
  14 #include <string.h>
  15
  16 #include "../smalloc.h"
  17 #include "../log.h"
  18 #include "tp.h"
  19
  20 static void tp_flush_work(struct flist_head *list)
  21 {
  22         struct tp_work *work;
  23
  24         while (!flist_empty(list)) {
  25                 int prio;
  26
  27                 work = flist_entry(list->next, struct tp_work, list);
  28                 flist_del(&work->list);
  29
  30                 prio = work->prio;
  31                 if (nice(prio) < 0)
  32                         log_err("fio: nice %s\n", strerror(errno));
  33
  34                 work->fn(work);
  35
  36                 if (nice(prio) < 0)
  37                         log_err("fio: nice %s\n", strerror(errno));
  38         }
  39 }
  40
  41 static void *tp_thread(void *data)
  42 {
  43         struct tp_data *tdat = data;
  44         struct flist_head work_list;
  45
  46         INIT_FLIST_HEAD(&work_list);
  47
  48         while (1) {
  49                 pthread_mutex_lock(&tdat->lock);
  50
  51                 if (!tdat->thread_exit && flist_empty(&tdat->work))
  52                         pthread_cond_wait(&tdat->cv, &tdat->lock);
  53
  54                 if (!flist_empty(&tdat->work))
  55                         flist_splice_tail_init(&tdat->work, &work_list);
  56
  57                 pthread_mutex_unlock(&tdat->lock);
  58
  59                 if (flist_empty(&work_list)) {
  60                         if (tdat->thread_exit)
  61                                 break;
  62                         continue;
  63                 }
  64
  65                 tp_flush_work(&work_list);
  66         }
  67
  68         return NULL;
  69 }
  70
  71 void tp_queue_work(struct tp_data *tdat, struct tp_work *work)
  72 {
  73         work->done = 0;
  74
  75         pthread_mutex_lock(&tdat->lock);
  76         flist_add_tail(&work->list, &tdat->work);
  77         pthread_mutex_unlock(&tdat->lock);
  78
  79         pthread_cond_signal(&tdat->cv);
  80 }
  81
  82 void tp_init(struct tp_data **tdatp)
  83 {
  84         struct tp_data *tdat;
  85         int ret;
  86
  87         if (*tdatp)
  88                 return;
  89
  90         *tdatp = tdat = smalloc(sizeof(*tdat));
  91         pthread_mutex_init(&tdat->lock, NULL);
  92         INIT_FLIST_HEAD(&tdat->work);
  93         pthread_cond_init(&tdat->cv, NULL);
  94         pthread_cond_init(&tdat->sleep_cv, NULL);
  95
  96         ret = pthread_create(&tdat->thread, NULL, tp_thread, tdat);
  97         if (ret)
  98                 log_err("fio: failed to create tp thread\n");
  99 }
 100
 101 void tp_exit(struct tp_data **tdatp)
 102 {
 103         struct tp_data *tdat = *tdatp;
 104         void *ret;
 105
 106         if (!tdat)
 107                 return;
 108
 109         pthread_mutex_lock(&tdat->lock);
 110         tdat->thread_exit = 1;
 111         pthread_mutex_unlock(&tdat->lock);
 112
 113         pthread_cond_signal(&tdat->cv);
 114
 115         pthread_join(tdat->thread, &ret);
 116
 117         sfree(tdat);
 118         *tdatp = NULL;
 119 }