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                         io_u_quiesce(td);
  28                         usleep(td->o.flow_sleep);
  29                 }
  30
  31                 return 1;
  32         }
  33
  34         /* No synchronization needed because it doesn't
  35          * matter if the flow count is slightly inaccurate */
  36         flow->flow_counter += td->o.flow;
  37         return 0;
  38 }
  39
  40 static struct fio_flow *flow_get(unsigned int id)
  41 {
  42         struct fio_flow *flow = NULL;
  43         struct flist_head *n;
  44
  45         if (!flow_lock)
  46                 return NULL;
  47
  48         fio_mutex_down(flow_lock);
  49
  50         flist_for_each(n, flow_list) {
  51                 flow = flist_entry(n, struct fio_flow, list);
  52                 if (flow->id == id)
  53                         break;
  54
  55                 flow = NULL;
  56         }
  57
  58         if (!flow) {
  59                 flow = smalloc(sizeof(*flow));
  60                 if (!flow) {
  61                         log_err("fio: smalloc pool exhausted\n");
  62                         fio_mutex_up(flow_lock);
  63                         return NULL;
  64                 }
  65                 flow->refs = 0;
  66                 INIT_FLIST_HEAD(&flow->list);
  67                 flow->id = id;
  68                 flow->flow_counter = 0;
  69
  70                 flist_add_tail(&flow->list, flow_list);
  71         }
  72
  73         flow->refs++;
  74         fio_mutex_up(flow_lock);
  75         return flow;
  76 }
  77
  78 static void flow_put(struct fio_flow *flow)
  79 {
  80         if (!flow_lock)
  81                 return;
  82
  83         fio_mutex_down(flow_lock);
  84
  85         if (!--flow->refs) {
  86                 flist_del(&flow->list);
  87                 sfree(flow);
  88         }
  89
  90         fio_mutex_up(flow_lock);
  91 }
  92
  93 void flow_init_job(struct thread_data *td)
  94 {
  95         if (td->o.flow)
  96                 td->flow = flow_get(td->o.flow_id);
  97 }
  98
  99 void flow_exit_job(struct thread_data *td)
 100 {
 101         if (td->flow) {
 102                 flow_put(td->flow);
 103                 td->flow = NULL;
 104         }
 105 }
 106
 107 void flow_init(void)
 108 {
 109         flow_list = smalloc(sizeof(*flow_list));
 110         if (!flow_list) {
 111                 log_err("fio: smalloc pool exhausted\n");
 112                 return;
 113         }
 114
 115         flow_lock = fio_mutex_init(FIO_MUTEX_UNLOCKED);
 116         if (!flow_lock) {
 117                 log_err("fio: failed to allocate flow lock\n");
 118                 sfree(flow_list);
 119                 return;
 120         }
 121
 122         INIT_FLIST_HEAD(flow_list);
 123 }
 124
 125 void flow_exit(void)
 126 {
 127         if (flow_lock)
 128                 fio_mutex_remove(flow_lock);
 129         if (flow_list)
 130                 sfree(flow_list);
 131 }