flow.c

   1 #include "fio.h"
   2 #include "mutex.h"
   3 #include "smalloc.h"
   4 #include "flist.h"
   5
   6 struct fio_flow {
   7         unsigned int refs;
   8         struct flist_head list;
   9         unsigned int id;
  10         long long int flow_counter;
  11 };
  12
  13 static struct flist_head *flow_list;
  14 static struct fio_mutex *flow_lock;
  15
  16 int flow_threshold_exceeded(struct thread_data *td)
  17 {
  18         struct fio_flow *flow = td->flow;
  19         int sign;
  20
  21         if (!flow)
  22                 return 0;
  23
  24         sign = td->o.flow > 0 ? 1 : -1;
  25         if (sign * flow->flow_counter > td->o.flow_watermark) {
  26                 if (td->o.flow_sleep)
  27                         usleep(td->o.flow_sleep);
  28                 return 1;
  29         }
  30
  31         /* No synchronization needed because it doesn't
  32          * matter if the flow count is slightly inaccurate */
  33         flow->flow_counter += td->o.flow;
  34         return 0;
  35 }
  36
  37 static struct fio_flow *flow_get(unsigned int id)
  38 {
  39         struct fio_flow *flow = NULL;
  40         struct flist_head *n;
  41
  42         fio_mutex_down(flow_lock);
  43
  44         flist_for_each(n, flow_list) {
  45                 flow = flist_entry(n, struct fio_flow, list);
  46                 if (flow->id == id)
  47                         break;
  48
  49                 flow = NULL;
  50         }
  51
  52         if (!flow) {
  53                 flow = smalloc(sizeof(*flow));
  54                 flow->refs = 0;
  55                 INIT_FLIST_HEAD(&flow->list);
  56                 flow->id = id;
  57                 flow->flow_counter = 0;
  58
  59                 flist_add_tail(&flow->list, flow_list);
  60         }
  61
  62         flow->refs++;
  63         fio_mutex_up(flow_lock);
  64         return flow;
  65 }
  66
  67 static void flow_put(struct fio_flow *flow)
  68 {
  69         fio_mutex_down(flow_lock);
  70
  71         if (!--flow->refs) {
  72                 flist_del(&flow->list);
  73                 sfree(flow);
  74         }
  75
  76         fio_mutex_up(flow_lock);
  77 }
  78
  79 void flow_init_job(struct thread_data *td)
  80 {
  81         if (td->o.flow)
  82                 td->flow = flow_get(td->o.flow_id);
  83 }
  84
  85 void flow_exit_job(struct thread_data *td)
  86 {
  87         if (td->flow) {
  88                 flow_put(td->flow);
  89                 td->flow = NULL;
  90         }
  91 }
  92
  93 void flow_init(void)
  94 {
  95         flow_lock = fio_mutex_init(FIO_MUTEX_UNLOCKED);
  96         flow_list = smalloc(sizeof(*flow_list));
  97         INIT_FLIST_HEAD(flow_list);
  98 }
  99
 100 void flow_exit(void)
 101 {
 102         fio_mutex_remove(flow_lock);
 103         sfree(flow_list);
 104 }