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         printf("tp_thread running\n");
  31
  32         while (1) {
  33                 pthread_mutex_lock(&tdat->lock);
  34
  35                 if (!tdat->thread_exit && flist_empty(&tdat->work))
  36                         pthread_cond_wait(&tdat->cv, &tdat->lock);
  37
  38                 if (!flist_empty(&tdat->work)) {
  39                         flist_splice(&tdat->work, &work_list);
  40                         INIT_FLIST_HEAD(&tdat->work);
  41                 }
  42
  43                 pthread_mutex_unlock(&tdat->lock);
  44
  45                 if (flist_empty(&work_list)) {
  46                         if (tdat->thread_exit)
  47                                 break;
  48                         continue;
  49                 }
  50
  51                 tp_flush_work(&work_list);
  52         }
  53
  54         printf("outta here\n");
  55         return NULL;
  56 }
  57
  58 void tp_queue_work(struct tp_data *tdat, struct tp_work *work)
  59 {
  60         work->done = 0;
  61
  62         pthread_mutex_lock(&tdat->lock);
  63         flist_add_tail(&work->list, &tdat->work);
  64         pthread_cond_signal(&tdat->cv);
  65         pthread_mutex_unlock(&tdat->lock);
  66 }
  67
  68 void tp_init(struct tp_data **tdatp)
  69 {
  70         struct tp_data *tdat;
  71         int ret;
  72
  73         if (*tdatp)
  74                 return;
  75
  76         *tdatp = tdat = smalloc(sizeof(*tdat));
  77         pthread_mutex_init(&tdat->lock, NULL);
  78         INIT_FLIST_HEAD(&tdat->work);
  79         pthread_cond_init(&tdat->cv, NULL);
  80         pthread_cond_init(&tdat->sleep_cv, NULL);
  81
  82         ret = pthread_create(&tdat->thread, NULL, tp_thread, tdat);
  83         if (ret)
  84                 log_err("fio: failed to create tp thread\n");
  85 }
  86
  87 void tp_exit(struct tp_data **tdatp)
  88 {
  89         struct tp_data *tdat = *tdatp;
  90         void *ret;
  91
  92         if (!tdat)
  93                 return;
  94
  95         tdat->thread_exit = 1;
  96         pthread_mutex_lock(&tdat->lock);
  97         pthread_cond_signal(&tdat->cv);
  98         pthread_mutex_unlock(&tdat->lock);
  99
 100         pthread_join(tdat->thread, &ret);
 101
 102         sfree(tdat);
 103         *tdatp = NULL;
 104 }