engines/syslet-rw.c

   1 /*
   2  * syslet engine
   3  *
   4  * IO engine that does regular pread(2)/pwrite(2) to transfer data, but
   5  * with syslets to make the execution async.
   6  *
   7  */
   8 #include <stdio.h>
   9 #include <stdlib.h>
  10 #include <unistd.h>
  11 #include <errno.h>
  12 #include <assert.h>
  13 #include <asm/unistd.h>
  14
  15 #include "../fio.h"
  16 #include "../os.h"
  17
  18 #ifdef FIO_HAVE_SYSLET
  19
  20 struct syslet_data {
  21         struct io_u **events;
  22         unsigned int nr_events;
  23
  24         struct async_head_user ahu;
  25         struct syslet_uatom **ring;
  26
  27         struct syslet_uatom *head, *tail;
  28 };
  29
  30 static void fio_syslet_complete_atom(struct thread_data *td,
  31                                      struct syslet_uatom *atom)
  32 {
  33         struct syslet_data *sd = td->io_ops->data;
  34         struct syslet_uatom *last;
  35         struct io_u *io_u;
  36
  37         /*
  38          * complete from the beginning of the sequence up to (and
  39          * including) this atom
  40          */
  41         last = atom;
  42         io_u = atom->private;
  43         atom = io_u->req.head;
  44
  45         /*
  46          * now complete in right order
  47          */
  48         do {
  49                 long ret;
  50
  51                 io_u = atom->private;
  52                 ret = *atom->ret_ptr;
  53                 if (ret >= 0)
  54                         io_u->resid = io_u->xfer_buflen - ret;
  55                 else if (ret < 0)
  56                         io_u->error = ret;
  57
  58                 assert(sd->nr_events < td->iodepth);
  59                 sd->events[sd->nr_events++] = io_u;
  60
  61                 if (atom == last)
  62                         break;
  63
  64                 atom = atom->next;
  65         } while (1);
  66
  67         assert(!last->next);
  68 }
  69
  70 /*
  71  * Inspect the ring to see if we have completed events
  72  */
  73 static void fio_syslet_complete(struct thread_data *td)
  74 {
  75         struct syslet_data *sd = td->io_ops->data;
  76
  77         do {
  78                 struct syslet_uatom *atom;
  79
  80                 atom = sd->ring[sd->ahu.user_ring_idx];
  81                 if (!atom)
  82                         break;
  83
  84                 sd->ring[sd->ahu.user_ring_idx] = NULL;
  85                 if (++sd->ahu.user_ring_idx == td->iodepth)
  86                         sd->ahu.user_ring_idx = 0;
  87
  88                 fio_syslet_complete_atom(td, atom);
  89         } while (1);
  90 }
  91
  92 static int fio_syslet_getevents(struct thread_data *td, int min,
  93                                 int fio_unused max,
  94                                 struct timespec fio_unused *t)
  95 {
  96         struct syslet_data *sd = td->io_ops->data;
  97         long ret;
  98
  99         do {
 100                 fio_syslet_complete(td);
 101
 102                 /*
 103                  * do we have enough immediate completions?
 104                  */
 105                 if (sd->nr_events >= (unsigned int) min)
 106                         break;
 107
 108                 /*
 109                  * OK, we need to wait for some events...
 110                  */
 111                 ret = async_wait(1, sd->ahu.user_ring_idx, &sd->ahu);
 112                 if (ret < 0)
 113                         return -errno;
 114         } while (1);
 115
 116         ret = sd->nr_events;
 117         sd->nr_events = 0;
 118         return ret;
 119 }
 120
 121 static struct io_u *fio_syslet_event(struct thread_data *td, int event)
 122 {
 123         struct syslet_data *sd = td->io_ops->data;
 124
 125         return sd->events[event];
 126 }
 127
 128 static void init_atom(struct syslet_uatom *atom, int nr, void *arg0,
 129                       void *arg1, void *arg2, void *arg3, void *ret_ptr,
 130                       unsigned long flags, void *priv)
 131 {
 132         atom->flags = flags;
 133         atom->nr = nr;
 134         atom->ret_ptr = ret_ptr;
 135         atom->next = NULL;
 136         atom->arg_ptr[0] = arg0;
 137         atom->arg_ptr[1] = arg1;
 138         atom->arg_ptr[2] = arg2;
 139         atom->arg_ptr[3] = arg3;
 140         atom->arg_ptr[4] = atom->arg_ptr[5] = NULL;
 141         atom->private = priv;
 142 }
 143
 144 /*
 145  * Use seek atom for sync
 146  */
 147 static void fio_syslet_prep_sync(struct io_u *io_u, struct fio_file *f)
 148 {
 149         init_atom(&io_u->req.atom, __NR_fsync, &f->fd, NULL, NULL, NULL,
 150                   &io_u->req.ret, 0, io_u);
 151 }
 152
 153 static void fio_syslet_prep_rw(struct io_u *io_u, struct fio_file *f)
 154 {
 155         int nr;
 156
 157         /*
 158          * prepare rw
 159          */
 160         if (io_u->ddir == DDIR_READ)
 161                 nr = __NR_pread64;
 162         else
 163                 nr = __NR_pwrite64;
 164
 165         init_atom(&io_u->req.atom, nr, &f->fd, &io_u->xfer_buf,
 166                   &io_u->xfer_buflen, &io_u->offset, &io_u->req.ret, 0, io_u);
 167 }
 168
 169 static int fio_syslet_prep(struct thread_data fio_unused *td, struct io_u *io_u)
 170 {
 171         struct fio_file *f = io_u->file;
 172
 173         if (io_u->ddir == DDIR_SYNC)
 174                 fio_syslet_prep_sync(io_u, f);
 175         else
 176                 fio_syslet_prep_rw(io_u, f);
 177
 178         return 0;
 179 }
 180
 181 static void cachemiss_thread_start(void)
 182 {
 183         while (1)
 184                 async_thread(NULL, NULL);
 185 }
 186
 187 #define THREAD_STACK_SIZE (16384)
 188
 189 static unsigned long thread_stack_alloc()
 190 {
 191         return (unsigned long) malloc(THREAD_STACK_SIZE) + THREAD_STACK_SIZE;
 192 }
 193
 194 static void fio_syslet_queued(struct thread_data *td, struct syslet_data *sd)
 195 {
 196         struct syslet_uatom *atom;
 197         struct timeval now;
 198
 199         fio_gettime(&now, NULL);
 200
 201         atom = sd->head;
 202         while (atom) {
 203                 struct io_u *io_u = atom->private;
 204
 205                 memcpy(&io_u->issue_time, &now, sizeof(now));
 206                 io_u_queued(td, io_u);
 207                 atom = atom->next;
 208         }
 209 }
 210
 211 static int fio_syslet_commit(struct thread_data *td)
 212 {
 213         struct syslet_data *sd = td->io_ops->data;
 214         struct syslet_uatom *done;
 215
 216         if (!sd->head)
 217                 return 0;
 218
 219         assert(!sd->tail->next);
 220
 221         if (!sd->ahu.new_thread_stack)
 222                 sd->ahu.new_thread_stack = thread_stack_alloc();
 223
 224         fio_syslet_queued(td, sd);
 225
 226         /*
 227          * On sync completion, the atom is returned. So on NULL return
 228          * it's queued asynchronously.
 229          */
 230         done = async_exec(sd->head, &sd->ahu);
 231
 232         sd->head = sd->tail = NULL;
 233
 234         if (done)
 235                 fio_syslet_complete_atom(td, done);
 236
 237         return 0;
 238 }
 239
 240 static int fio_syslet_queue(struct thread_data *td, struct io_u *io_u)
 241 {
 242         struct syslet_data *sd = td->io_ops->data;
 243
 244         if (sd->tail) {
 245                 sd->tail->next = &io_u->req.atom;
 246                 sd->tail = &io_u->req.atom;
 247         } else
 248                 sd->head = sd->tail = &io_u->req.atom;
 249
 250         io_u->req.head = sd->head;
 251         return FIO_Q_QUEUED;
 252 }
 253
 254 static int async_head_init(struct syslet_data *sd, unsigned int depth)
 255 {
 256         unsigned long ring_size;
 257
 258         memset(&sd->ahu, 0, sizeof(struct async_head_user));
 259
 260         ring_size = sizeof(struct syslet_uatom *) * depth;
 261         sd->ring = malloc(ring_size);
 262         memset(sd->ring, 0, ring_size);
 263
 264         sd->ahu.user_ring_idx = 0;
 265         sd->ahu.completion_ring = sd->ring;
 266         sd->ahu.ring_size_bytes = ring_size;
 267         sd->ahu.head_stack = thread_stack_alloc();
 268         sd->ahu.head_eip = (unsigned long) cachemiss_thread_start;
 269         sd->ahu.new_thread_eip = (unsigned long) cachemiss_thread_start;
 270
 271         return 0;
 272 }
 273
 274 static void async_head_exit(struct syslet_data *sd)
 275 {
 276         free(sd->ring);
 277 }
 278
 279 static void fio_syslet_cleanup(struct thread_data *td)
 280 {
 281         struct syslet_data *sd = td->io_ops->data;
 282
 283         if (sd) {
 284                 async_head_exit(sd);
 285                 free(sd->events);
 286                 free(sd);
 287                 td->io_ops->data = NULL;
 288         }
 289 }
 290
 291 static int fio_syslet_init(struct thread_data *td)
 292 {
 293         struct syslet_data *sd;
 294
 295
 296         sd = malloc(sizeof(*sd));
 297         memset(sd, 0, sizeof(*sd));
 298         sd->events = malloc(sizeof(struct io_u *) * td->iodepth);
 299         memset(sd->events, 0, sizeof(struct io_u *) * td->iodepth);
 300
 301         /*
 302          * This will handily fail for kernels where syslet isn't available
 303          */
 304         if (async_head_init(sd, td->iodepth)) {
 305                 free(sd->events);
 306                 free(sd);
 307                 return 1;
 308         }
 309
 310         td->io_ops->data = sd;
 311         return 0;
 312 }
 313
 314 static struct ioengine_ops ioengine = {
 315         .name           = "syslet-rw",
 316         .version        = FIO_IOOPS_VERSION,
 317         .init           = fio_syslet_init,
 318         .prep           = fio_syslet_prep,
 319         .queue          = fio_syslet_queue,
 320         .commit         = fio_syslet_commit,
 321         .getevents      = fio_syslet_getevents,
 322         .event          = fio_syslet_event,
 323         .cleanup        = fio_syslet_cleanup,
 324         .open_file      = generic_open_file,
 325         .close_file     = generic_close_file,
 326 };
 327
 328 #else /* FIO_HAVE_SYSLET */
 329
 330 /*
 331  * When we have a proper configure system in place, we simply wont build
 332  * and install this io engine. For now install a crippled version that
 333  * just complains and fails to load.
 334  */
 335 static int fio_syslet_init(struct thread_data fio_unused *td)
 336 {
 337         fprintf(stderr, "fio: syslet not available\n");
 338         return 1;
 339 }
 340
 341 static struct ioengine_ops ioengine = {
 342         .name           = "syslet-rw",
 343         .version        = FIO_IOOPS_VERSION,
 344         .init           = fio_syslet_init,
 345 };
 346
 347 #endif /* FIO_HAVE_SYSLET */
 348
 349 static void fio_init fio_syslet_register(void)
 350 {
 351         register_ioengine(&ioengine);
 352 }
 353
 354 static void fio_exit fio_syslet_unregister(void)
 355 {
 356         unregister_ioengine(&ioengine);
 357 }