tp.c

   1 #include <stdio.h>
   2 #include <stdlib.h>
   3 #include <stdarg.h>
   4 #include <unistd.h>
   5 #include <errno.h>
   6 #include <pthread.h>
   7
   8 #include "smalloc.h"
   9 #include "log.h"
  10 #include "tp.h"
  11
  12 static void tp_flush_work(struct flist_head *list)
  13 {
  14         struct tp_work *work;
  15
  16         while (!flist_empty(list)) {
  17                 work = flist_entry(list->next, struct tp_work, list);
  18                 flist_del(&work->list);
  19                 work->fn(work);
  20         }
  21 }
  22
  23 static void *tp_thread(void *data)
  24 {
  25         struct tp_data *tdat = data;
  26         struct flist_head work_list;
  27
  28         INIT_FLIST_HEAD(&work_list);
  29
  30         while (1) {
  31                 pthread_mutex_lock(&tdat->lock);
  32
  33                 if (!tdat->thread_exit && flist_empty(&tdat->work))
  34                         pthread_cond_wait(&tdat->cv, &tdat->lock);
  35
  36                 if (!flist_empty(&tdat->work)) {
  37                         flist_splice(&tdat->work, &work_list);
  38                         INIT_FLIST_HEAD(&tdat->work);
  39                 }
  40
  41                 pthread_mutex_unlock(&tdat->lock);
  42
  43                 if (flist_empty(&work_list)) {
  44                         if (tdat->thread_exit)
  45                                 break;
  46                         continue;
  47                 }
  48
  49                 tp_flush_work(&work_list);
  50         }
  51
  52         return NULL;
  53 }
  54
  55 void tp_queue_work(struct tp_data *tdat, struct tp_work *work)
  56 {
  57         work->done = 0;
  58
  59         pthread_mutex_lock(&tdat->lock);
  60         flist_add_tail(&work->list, &tdat->work);
  61         pthread_cond_signal(&tdat->cv);
  62         pthread_mutex_unlock(&tdat->lock);
  63 }
  64
  65 void tp_init(struct tp_data **tdatp)
  66 {
  67         struct tp_data *tdat;
  68         int ret;
  69
  70         if (*tdatp)
  71                 return;
  72
  73         *tdatp = tdat = smalloc(sizeof(*tdat));
  74         pthread_mutex_init(&tdat->lock, NULL);
  75         INIT_FLIST_HEAD(&tdat->work);
  76         pthread_cond_init(&tdat->cv, NULL);
  77         pthread_cond_init(&tdat->sleep_cv, NULL);
  78
  79         ret = pthread_create(&tdat->thread, NULL, tp_thread, tdat);
  80         if (ret)
  81                 log_err("fio: failed to create tp thread\n");
  82 }
  83
  84 void tp_exit(struct tp_data **tdatp)
  85 {
  86         struct tp_data *tdat = *tdatp;
  87         void *ret;
  88
  89         if (!tdat)
  90                 return;
  91
  92         tdat->thread_exit = 1;
  93         pthread_mutex_lock(&tdat->lock);
  94         pthread_cond_signal(&tdat->cv);
  95         pthread_mutex_unlock(&tdat->lock);
  96
  97         pthread_join(tdat->thread, &ret);
  98
  99         sfree(tdat);
 100         *tdatp = NULL;
 101 }