Merge branch 'dev' of https://github.com/smartxworks/fio
[fio.git] / engines / rbd.c
1 /*
2  * rbd engine
3  *
4  * IO engine using Ceph's librbd to test RADOS Block Devices.
5  *
6  */
7
8 #include <rbd/librbd.h>
9
10 #include "../fio.h"
11 #include "../optgroup.h"
12
13 #ifdef CONFIG_RBD_POLL
14 /* add for poll */
15 #include <poll.h>
16 #include <sys/eventfd.h>
17 #endif
18
19 struct fio_rbd_iou {
20         struct io_u *io_u;
21         rbd_completion_t completion;
22         int io_seen;
23         int io_complete;
24 };
25
26 struct rbd_data {
27         rados_t cluster;
28         rados_ioctx_t io_ctx;
29         rbd_image_t image;
30         struct io_u **aio_events;
31         struct io_u **sort_events;
32         int fd; /* add for poll */
33         bool connected;
34 };
35
36 struct rbd_options {
37         void *pad;
38         char *cluster_name;
39         char *rbd_name;
40         char *pool_name;
41         char *client_name;
42         int busy_poll;
43 };
44
45 static struct fio_option options[] = {
46         {
47                 .name           = "clustername",
48                 .lname          = "ceph cluster name",
49                 .type           = FIO_OPT_STR_STORE,
50                 .help           = "Cluster name for ceph",
51                 .off1           = offsetof(struct rbd_options, cluster_name),
52                 .category       = FIO_OPT_C_ENGINE,
53                 .group          = FIO_OPT_G_RBD,
54         },
55         {
56                 .name           = "rbdname",
57                 .lname          = "rbd engine rbdname",
58                 .type           = FIO_OPT_STR_STORE,
59                 .help           = "RBD name for RBD engine",
60                 .off1           = offsetof(struct rbd_options, rbd_name),
61                 .category       = FIO_OPT_C_ENGINE,
62                 .group          = FIO_OPT_G_RBD,
63         },
64         {
65                 .name           = "pool",
66                 .lname          = "rbd engine pool",
67                 .type           = FIO_OPT_STR_STORE,
68                 .help           = "Name of the pool hosting the RBD for the RBD engine",
69                 .off1           = offsetof(struct rbd_options, pool_name),
70                 .category       = FIO_OPT_C_ENGINE,
71                 .group          = FIO_OPT_G_RBD,
72         },
73         {
74                 .name           = "clientname",
75                 .lname          = "rbd engine clientname",
76                 .type           = FIO_OPT_STR_STORE,
77                 .help           = "Name of the ceph client to access the RBD for the RBD engine",
78                 .off1           = offsetof(struct rbd_options, client_name),
79                 .category       = FIO_OPT_C_ENGINE,
80                 .group          = FIO_OPT_G_RBD,
81         },
82         {
83                 .name           = "busy_poll",
84                 .lname          = "Busy poll",
85                 .type           = FIO_OPT_BOOL,
86                 .help           = "Busy poll for completions instead of sleeping",
87                 .off1           = offsetof(struct rbd_options, busy_poll),
88                 .def            = "0",
89                 .category       = FIO_OPT_C_ENGINE,
90                 .group          = FIO_OPT_G_RBD,
91         },
92         {
93                 .name = NULL,
94         },
95 };
96
97 static int _fio_setup_rbd_data(struct thread_data *td,
98                                struct rbd_data **rbd_data_ptr)
99 {
100         struct rbd_data *rbd;
101
102         if (td->io_ops_data)
103                 return 0;
104
105         rbd = calloc(1, sizeof(struct rbd_data));
106         if (!rbd)
107                 goto failed;
108
109         rbd->connected = false;
110
111         /* add for poll, init fd: -1 */
112         rbd->fd = -1;
113
114         rbd->aio_events = calloc(td->o.iodepth, sizeof(struct io_u *));
115         if (!rbd->aio_events)
116                 goto failed;
117
118         rbd->sort_events = calloc(td->o.iodepth, sizeof(struct io_u *));
119         if (!rbd->sort_events)
120                 goto failed;
121
122         *rbd_data_ptr = rbd;
123         return 0;
124
125 failed:
126         if (rbd) {
127                 if (rbd->aio_events) 
128                         free(rbd->aio_events);
129                 if (rbd->sort_events)
130                         free(rbd->sort_events);
131                 free(rbd);
132         }
133         return 1;
134
135 }
136
137 #ifdef CONFIG_RBD_POLL
138 static bool _fio_rbd_setup_poll(struct rbd_data *rbd)
139 {
140         int r;
141
142         /* add for rbd poll */
143         rbd->fd = eventfd(0, EFD_SEMAPHORE);
144         if (rbd->fd < 0) {
145                 log_err("eventfd failed.\n");
146                 return false;
147         }
148
149         r = rbd_set_image_notification(rbd->image, rbd->fd, EVENT_TYPE_EVENTFD);
150         if (r < 0) {
151                 log_err("rbd_set_image_notification failed.\n");
152                 close(rbd->fd);
153                 rbd->fd = -1;
154                 return false;
155         }
156
157         return true;
158 }
159 #else
160 static bool _fio_rbd_setup_poll(struct rbd_data *rbd)
161 {
162         return true;
163 }
164 #endif
165
166 static int _fio_rbd_connect(struct thread_data *td)
167 {
168         struct rbd_data *rbd = td->io_ops_data;
169         struct rbd_options *o = td->eo;
170         int r;
171
172         if (o->cluster_name) {
173                 char *client_name = NULL; 
174
175                 /*
176                  * If we specify cluser name, the rados_create2
177                  * will not assume 'client.'. name is considered
178                  * as a full type.id namestr
179                  */
180                 if (o->client_name) {
181                         if (!index(o->client_name, '.')) {
182                                 client_name = calloc(1, strlen("client.") +
183                                                     strlen(o->client_name) + 1);
184                                 strcat(client_name, "client.");
185                                 strcat(client_name, o->client_name);
186                         } else {
187                                 client_name = o->client_name;
188                         }
189                 }
190
191                 r = rados_create2(&rbd->cluster, o->cluster_name,
192                                  client_name, 0);
193
194                 if (client_name && !index(o->client_name, '.'))
195                         free(client_name);
196         } else
197                 r = rados_create(&rbd->cluster, o->client_name);
198         
199         if (r < 0) {
200                 log_err("rados_create failed.\n");
201                 goto failed_early;
202         }
203
204         r = rados_conf_read_file(rbd->cluster, NULL);
205         if (r < 0) {
206                 log_err("rados_conf_read_file failed.\n");
207                 goto failed_early;
208         }
209
210         r = rados_connect(rbd->cluster);
211         if (r < 0) {
212                 log_err("rados_connect failed.\n");
213                 goto failed_shutdown;
214         }
215
216         r = rados_ioctx_create(rbd->cluster, o->pool_name, &rbd->io_ctx);
217         if (r < 0) {
218                 log_err("rados_ioctx_create failed.\n");
219                 goto failed_shutdown;
220         }
221
222         r = rbd_open(rbd->io_ctx, o->rbd_name, &rbd->image, NULL /*snap */ );
223         if (r < 0) {
224                 log_err("rbd_open failed.\n");
225                 goto failed_open;
226         }
227
228         if (!_fio_rbd_setup_poll(rbd))
229                 goto failed_poll;
230
231         return 0;
232
233 failed_poll:
234         rbd_close(rbd->image);
235         rbd->image = NULL;
236 failed_open:
237         rados_ioctx_destroy(rbd->io_ctx);
238         rbd->io_ctx = NULL;
239 failed_shutdown:
240         rados_shutdown(rbd->cluster);
241         rbd->cluster = NULL;
242 failed_early:
243         return 1;
244 }
245
246 static void _fio_rbd_disconnect(struct rbd_data *rbd)
247 {
248         if (!rbd)
249                 return;
250
251         /* close eventfd */
252         if (rbd->fd != -1) {
253                 close(rbd->fd);
254                 rbd->fd = -1;
255         }
256
257         /* shutdown everything */
258         if (rbd->image) {
259                 rbd_close(rbd->image);
260                 rbd->image = NULL;
261         }
262
263         if (rbd->io_ctx) {
264                 rados_ioctx_destroy(rbd->io_ctx);
265                 rbd->io_ctx = NULL;
266         }
267
268         if (rbd->cluster) {
269                 rados_shutdown(rbd->cluster);
270                 rbd->cluster = NULL;
271         }
272 }
273
274 static void _fio_rbd_finish_aiocb(rbd_completion_t comp, void *data)
275 {
276         struct fio_rbd_iou *fri = data;
277         struct io_u *io_u = fri->io_u;
278         ssize_t ret;
279
280         /*
281          * Looks like return value is 0 for success, or < 0 for
282          * a specific error. So we have to assume that it can't do
283          * partial completions.
284          */
285         ret = rbd_aio_get_return_value(fri->completion);
286         if (ret < 0) {
287                 io_u->error = -ret;
288                 io_u->resid = io_u->xfer_buflen;
289         } else
290                 io_u->error = 0;
291
292         fri->io_complete = 1;
293 }
294
295 static struct io_u *fio_rbd_event(struct thread_data *td, int event)
296 {
297         struct rbd_data *rbd = td->io_ops_data;
298
299         return rbd->aio_events[event];
300 }
301
302 static inline int fri_check_complete(struct rbd_data *rbd, struct io_u *io_u,
303                                      unsigned int *events)
304 {
305         struct fio_rbd_iou *fri = io_u->engine_data;
306
307         if (fri->io_complete) {
308                 fri->io_seen = 1;
309                 rbd->aio_events[*events] = io_u;
310                 (*events)++;
311
312                 rbd_aio_release(fri->completion);
313                 return 1;
314         }
315
316         return 0;
317 }
318
319 #ifndef CONFIG_RBD_POLL
320 static inline int rbd_io_u_seen(struct io_u *io_u)
321 {
322         struct fio_rbd_iou *fri = io_u->engine_data;
323
324         return fri->io_seen;
325 }
326 #endif
327
328 static void rbd_io_u_wait_complete(struct io_u *io_u)
329 {
330         struct fio_rbd_iou *fri = io_u->engine_data;
331
332         rbd_aio_wait_for_complete(fri->completion);
333 }
334
335 static int rbd_io_u_cmp(const void *p1, const void *p2)
336 {
337         const struct io_u **a = (const struct io_u **) p1;
338         const struct io_u **b = (const struct io_u **) p2;
339         uint64_t at, bt;
340
341         at = utime_since_now(&(*a)->start_time);
342         bt = utime_since_now(&(*b)->start_time);
343
344         if (at < bt)
345                 return -1;
346         else if (at == bt)
347                 return 0;
348         else
349                 return 1;
350 }
351
352 static int rbd_iter_events(struct thread_data *td, unsigned int *events,
353                            unsigned int min_evts, int wait)
354 {
355         struct rbd_data *rbd = td->io_ops_data;
356         unsigned int this_events = 0;
357         struct io_u *io_u;
358         int i, sidx = 0;
359
360 #ifdef CONFIG_RBD_POLL
361         int ret = 0;
362         int event_num = 0;
363         struct fio_rbd_iou *fri = NULL;
364         rbd_completion_t comps[min_evts];
365         uint64_t counter;
366         bool completed;
367
368         struct pollfd pfd;
369         pfd.fd = rbd->fd;
370         pfd.events = POLLIN;
371
372         ret = poll(&pfd, 1, wait ? -1 : 0);
373         if (ret <= 0)
374                 return 0;
375         if (!(pfd.revents & POLLIN))
376                 return 0;
377
378         event_num = rbd_poll_io_events(rbd->image, comps, min_evts);
379
380         for (i = 0; i < event_num; i++) {
381                 fri = rbd_aio_get_arg(comps[i]);
382                 io_u = fri->io_u;
383
384                 /* best effort to decrement the semaphore */
385                 ret = read(rbd->fd, &counter, sizeof(counter));
386                 if (ret <= 0)
387                         log_err("rbd_iter_events failed to decrement semaphore.\n");
388
389                 completed = fri_check_complete(rbd, io_u, events);
390                 assert(completed);
391
392                 this_events++;
393         }
394 #else
395         io_u_qiter(&td->io_u_all, io_u, i) {
396                 if (!(io_u->flags & IO_U_F_FLIGHT))
397                         continue;
398                 if (rbd_io_u_seen(io_u))
399                         continue;
400
401                 if (fri_check_complete(rbd, io_u, events))
402                         this_events++;
403                 else if (wait)
404                         rbd->sort_events[sidx++] = io_u;
405         }
406 #endif
407
408         if (!wait || !sidx)
409                 return this_events;
410
411         /*
412          * Sort events, oldest issue first, then wait on as many as we
413          * need in order of age. If we have enough events, stop waiting,
414          * and just check if any of the older ones are done.
415          */
416         if (sidx > 1)
417                 qsort(rbd->sort_events, sidx, sizeof(struct io_u *), rbd_io_u_cmp);
418
419         for (i = 0; i < sidx; i++) {
420                 io_u = rbd->sort_events[i];
421
422                 if (fri_check_complete(rbd, io_u, events)) {
423                         this_events++;
424                         continue;
425                 }
426
427                 /*
428                  * Stop waiting when we have enough, but continue checking
429                  * all pending IOs if they are complete.
430                  */
431                 if (*events >= min_evts)
432                         continue;
433
434                 rbd_io_u_wait_complete(io_u);
435
436                 if (fri_check_complete(rbd, io_u, events))
437                         this_events++;
438         }
439
440         return this_events;
441 }
442
443 static int fio_rbd_getevents(struct thread_data *td, unsigned int min,
444                              unsigned int max, const struct timespec *t)
445 {
446         unsigned int this_events, events = 0;
447         struct rbd_options *o = td->eo;
448         int wait = 0;
449
450         do {
451                 this_events = rbd_iter_events(td, &events, min, wait);
452
453                 if (events >= min)
454                         break;
455                 if (this_events)
456                         continue;
457
458                 if (!o->busy_poll)
459                         wait = 1;
460                 else
461                         nop;
462         } while (1);
463
464         return events;
465 }
466
467 static enum fio_q_status fio_rbd_queue(struct thread_data *td,
468                                        struct io_u *io_u)
469 {
470         struct rbd_data *rbd = td->io_ops_data;
471         struct fio_rbd_iou *fri = io_u->engine_data;
472         int r = -1;
473
474         fio_ro_check(td, io_u);
475
476         fri->io_seen = 0;
477         fri->io_complete = 0;
478
479         r = rbd_aio_create_completion(fri, _fio_rbd_finish_aiocb,
480                                                 &fri->completion);
481         if (r < 0) {
482                 log_err("rbd_aio_create_completion failed.\n");
483                 goto failed;
484         }
485
486         if (io_u->ddir == DDIR_WRITE) {
487                 r = rbd_aio_write(rbd->image, io_u->offset, io_u->xfer_buflen,
488                                          io_u->xfer_buf, fri->completion);
489                 if (r < 0) {
490                         log_err("rbd_aio_write failed.\n");
491                         goto failed_comp;
492                 }
493
494         } else if (io_u->ddir == DDIR_READ) {
495                 r = rbd_aio_read(rbd->image, io_u->offset, io_u->xfer_buflen,
496                                         io_u->xfer_buf, fri->completion);
497
498                 if (r < 0) {
499                         log_err("rbd_aio_read failed.\n");
500                         goto failed_comp;
501                 }
502         } else if (io_u->ddir == DDIR_TRIM) {
503                 r = rbd_aio_discard(rbd->image, io_u->offset,
504                                         io_u->xfer_buflen, fri->completion);
505                 if (r < 0) {
506                         log_err("rbd_aio_discard failed.\n");
507                         goto failed_comp;
508                 }
509         } else if (io_u->ddir == DDIR_SYNC) {
510                 r = rbd_aio_flush(rbd->image, fri->completion);
511                 if (r < 0) {
512                         log_err("rbd_flush failed.\n");
513                         goto failed_comp;
514                 }
515         } else {
516                 dprint(FD_IO, "%s: Warning: unhandled ddir: %d\n", __func__,
517                        io_u->ddir);
518                 r = -EINVAL;
519                 goto failed_comp;
520         }
521
522         return FIO_Q_QUEUED;
523 failed_comp:
524         rbd_aio_release(fri->completion);
525 failed:
526         io_u->error = -r;
527         td_verror(td, io_u->error, "xfer");
528         return FIO_Q_COMPLETED;
529 }
530
531 static int fio_rbd_init(struct thread_data *td)
532 {
533         int r;
534         struct rbd_data *rbd = td->io_ops_data;
535
536         if (rbd->connected)
537                 return 0;
538
539         r = _fio_rbd_connect(td);
540         if (r) {
541                 log_err("fio_rbd_connect failed, return code: %d .\n", r);
542                 goto failed;
543         }
544
545         return 0;
546
547 failed:
548         return 1;
549 }
550
551 static void fio_rbd_cleanup(struct thread_data *td)
552 {
553         struct rbd_data *rbd = td->io_ops_data;
554
555         if (rbd) {
556                 _fio_rbd_disconnect(rbd);
557                 free(rbd->aio_events);
558                 free(rbd->sort_events);
559                 free(rbd);
560         }
561 }
562
563 static int fio_rbd_setup(struct thread_data *td)
564 {
565         rbd_image_info_t info;
566         struct fio_file *f;
567         struct rbd_data *rbd = NULL;
568         int r;
569
570         /* allocate engine specific structure to deal with librbd. */
571         r = _fio_setup_rbd_data(td, &rbd);
572         if (r) {
573                 log_err("fio_setup_rbd_data failed.\n");
574                 goto cleanup;
575         }
576         td->io_ops_data = rbd;
577
578         /* librbd does not allow us to run first in the main thread and later
579          * in a fork child. It needs to be the same process context all the
580          * time. 
581          */
582         td->o.use_thread = 1;
583
584         /* connect in the main thread to determine to determine
585          * the size of the given RADOS block device. And disconnect
586          * later on.
587          */
588         r = _fio_rbd_connect(td);
589         if (r) {
590                 log_err("fio_rbd_connect failed.\n");
591                 goto cleanup;
592         }
593         rbd->connected = true;
594
595         /* get size of the RADOS block device */
596         r = rbd_stat(rbd->image, &info, sizeof(info));
597         if (r < 0) {
598                 log_err("rbd_status failed.\n");
599                 goto cleanup;
600         } else if (info.size == 0) {
601                 log_err("image size should be larger than zero.\n");
602                 r = -EINVAL;
603                 goto cleanup;
604         }
605
606         dprint(FD_IO, "rbd-engine: image size: %" PRIu64 "\n", info.size);
607
608         /* taken from "net" engine. Pretend we deal with files,
609          * even if we do not have any ideas about files.
610          * The size of the RBD is set instead of a artificial file.
611          */
612         if (!td->files_index) {
613                 add_file(td, td->o.filename ? : "rbd", 0, 0);
614                 td->o.nr_files = td->o.nr_files ? : 1;
615                 td->o.open_files++;
616         }
617         f = td->files[0];
618         f->real_file_size = info.size;
619
620         return 0;
621
622 cleanup:
623         fio_rbd_cleanup(td);
624         return r;
625 }
626
627 static int fio_rbd_open(struct thread_data *td, struct fio_file *f)
628 {
629         return 0;
630 }
631
632 static int fio_rbd_invalidate(struct thread_data *td, struct fio_file *f)
633 {
634 #if defined(CONFIG_RBD_INVAL)
635         struct rbd_data *rbd = td->io_ops_data;
636
637         return rbd_invalidate_cache(rbd->image);
638 #else
639         return 0;
640 #endif
641 }
642
643 static void fio_rbd_io_u_free(struct thread_data *td, struct io_u *io_u)
644 {
645         struct fio_rbd_iou *fri = io_u->engine_data;
646
647         if (fri) {
648                 io_u->engine_data = NULL;
649                 free(fri);
650         }
651 }
652
653 static int fio_rbd_io_u_init(struct thread_data *td, struct io_u *io_u)
654 {
655         struct fio_rbd_iou *fri;
656
657         fri = calloc(1, sizeof(*fri));
658         fri->io_u = io_u;
659         io_u->engine_data = fri;
660         return 0;
661 }
662
663 static struct ioengine_ops ioengine = {
664         .name                   = "rbd",
665         .version                = FIO_IOOPS_VERSION,
666         .setup                  = fio_rbd_setup,
667         .init                   = fio_rbd_init,
668         .queue                  = fio_rbd_queue,
669         .getevents              = fio_rbd_getevents,
670         .event                  = fio_rbd_event,
671         .cleanup                = fio_rbd_cleanup,
672         .open_file              = fio_rbd_open,
673         .invalidate             = fio_rbd_invalidate,
674         .options                = options,
675         .io_u_init              = fio_rbd_io_u_init,
676         .io_u_free              = fio_rbd_io_u_free,
677         .option_struct_size     = sizeof(struct rbd_options),
678 };
679
680 static void fio_init fio_rbd_register(void)
681 {
682         register_ioengine(&ioengine);
683 }
684
685 static void fio_exit fio_rbd_unregister(void)
686 {
687         unregister_ioengine(&ioengine);
688 }