336837e9d812bb0119308a62620a4cb2b1382003
[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         if (o->pool_name == NULL) {
204                 log_err("rbd pool name must be provided.\n");
205                 goto failed_early;
206         }
207         if (!o->rbd_name) {
208                 log_err("rbdname must be provided.\n");
209                 goto failed_early;
210         }
211
212         r = rados_conf_read_file(rbd->cluster, NULL);
213         if (r < 0) {
214                 log_err("rados_conf_read_file failed.\n");
215                 goto failed_early;
216         }
217
218         r = rados_connect(rbd->cluster);
219         if (r < 0) {
220                 log_err("rados_connect failed.\n");
221                 goto failed_shutdown;
222         }
223
224         r = rados_ioctx_create(rbd->cluster, o->pool_name, &rbd->io_ctx);
225         if (r < 0) {
226                 log_err("rados_ioctx_create failed.\n");
227                 goto failed_shutdown;
228         }
229
230         if (td->o.odirect) {
231                 r = rados_conf_set(rbd->cluster, "rbd_cache", "false");
232                 if (r < 0) {
233                         log_info("failed to disable RBD in-memory cache\n");
234                 }
235         }
236
237         r = rbd_open(rbd->io_ctx, o->rbd_name, &rbd->image, NULL /*snap */ );
238         if (r < 0) {
239                 log_err("rbd_open failed.\n");
240                 goto failed_open;
241         }
242
243         if (!_fio_rbd_setup_poll(rbd))
244                 goto failed_poll;
245
246         return 0;
247
248 failed_poll:
249         rbd_close(rbd->image);
250         rbd->image = NULL;
251 failed_open:
252         rados_ioctx_destroy(rbd->io_ctx);
253         rbd->io_ctx = NULL;
254 failed_shutdown:
255         rados_shutdown(rbd->cluster);
256         rbd->cluster = NULL;
257 failed_early:
258         return 1;
259 }
260
261 static void _fio_rbd_disconnect(struct rbd_data *rbd)
262 {
263         if (!rbd)
264                 return;
265
266         /* close eventfd */
267         if (rbd->fd != -1) {
268                 close(rbd->fd);
269                 rbd->fd = -1;
270         }
271
272         /* shutdown everything */
273         if (rbd->image) {
274                 rbd_close(rbd->image);
275                 rbd->image = NULL;
276         }
277
278         if (rbd->io_ctx) {
279                 rados_ioctx_destroy(rbd->io_ctx);
280                 rbd->io_ctx = NULL;
281         }
282
283         if (rbd->cluster) {
284                 rados_shutdown(rbd->cluster);
285                 rbd->cluster = NULL;
286         }
287 }
288
289 static void _fio_rbd_finish_aiocb(rbd_completion_t comp, void *data)
290 {
291         struct fio_rbd_iou *fri = data;
292         struct io_u *io_u = fri->io_u;
293         ssize_t ret;
294
295         /*
296          * Looks like return value is 0 for success, or < 0 for
297          * a specific error. So we have to assume that it can't do
298          * partial completions.
299          */
300         ret = rbd_aio_get_return_value(fri->completion);
301         if (ret < 0) {
302                 io_u->error = -ret;
303                 io_u->resid = io_u->xfer_buflen;
304         } else
305                 io_u->error = 0;
306
307         fri->io_complete = 1;
308 }
309
310 static struct io_u *fio_rbd_event(struct thread_data *td, int event)
311 {
312         struct rbd_data *rbd = td->io_ops_data;
313
314         return rbd->aio_events[event];
315 }
316
317 static inline int fri_check_complete(struct rbd_data *rbd, struct io_u *io_u,
318                                      unsigned int *events)
319 {
320         struct fio_rbd_iou *fri = io_u->engine_data;
321
322         if (fri->io_complete) {
323                 fri->io_seen = 1;
324                 rbd->aio_events[*events] = io_u;
325                 (*events)++;
326
327                 rbd_aio_release(fri->completion);
328                 return 1;
329         }
330
331         return 0;
332 }
333
334 #ifndef CONFIG_RBD_POLL
335 static inline int rbd_io_u_seen(struct io_u *io_u)
336 {
337         struct fio_rbd_iou *fri = io_u->engine_data;
338
339         return fri->io_seen;
340 }
341 #endif
342
343 static void rbd_io_u_wait_complete(struct io_u *io_u)
344 {
345         struct fio_rbd_iou *fri = io_u->engine_data;
346
347         rbd_aio_wait_for_complete(fri->completion);
348 }
349
350 static int rbd_io_u_cmp(const void *p1, const void *p2)
351 {
352         const struct io_u **a = (const struct io_u **) p1;
353         const struct io_u **b = (const struct io_u **) p2;
354         uint64_t at, bt;
355
356         at = utime_since_now(&(*a)->start_time);
357         bt = utime_since_now(&(*b)->start_time);
358
359         if (at < bt)
360                 return -1;
361         else if (at == bt)
362                 return 0;
363         else
364                 return 1;
365 }
366
367 static int rbd_iter_events(struct thread_data *td, unsigned int *events,
368                            unsigned int min_evts, int wait)
369 {
370         struct rbd_data *rbd = td->io_ops_data;
371         unsigned int this_events = 0;
372         struct io_u *io_u;
373         int i, sidx = 0;
374
375 #ifdef CONFIG_RBD_POLL
376         int ret = 0;
377         int event_num = 0;
378         struct fio_rbd_iou *fri = NULL;
379         rbd_completion_t comps[min_evts];
380         uint64_t counter;
381         bool completed;
382
383         struct pollfd pfd;
384         pfd.fd = rbd->fd;
385         pfd.events = POLLIN;
386
387         ret = poll(&pfd, 1, wait ? -1 : 0);
388         if (ret <= 0)
389                 return 0;
390         if (!(pfd.revents & POLLIN))
391                 return 0;
392
393         event_num = rbd_poll_io_events(rbd->image, comps, min_evts);
394
395         for (i = 0; i < event_num; i++) {
396                 fri = rbd_aio_get_arg(comps[i]);
397                 io_u = fri->io_u;
398
399                 /* best effort to decrement the semaphore */
400                 ret = read(rbd->fd, &counter, sizeof(counter));
401                 if (ret <= 0)
402                         log_err("rbd_iter_events failed to decrement semaphore.\n");
403
404                 completed = fri_check_complete(rbd, io_u, events);
405                 assert(completed);
406
407                 this_events++;
408         }
409 #else
410         io_u_qiter(&td->io_u_all, io_u, i) {
411                 if (!(io_u->flags & IO_U_F_FLIGHT))
412                         continue;
413                 if (rbd_io_u_seen(io_u))
414                         continue;
415
416                 if (fri_check_complete(rbd, io_u, events))
417                         this_events++;
418                 else if (wait)
419                         rbd->sort_events[sidx++] = io_u;
420         }
421 #endif
422
423         if (!wait || !sidx)
424                 return this_events;
425
426         /*
427          * Sort events, oldest issue first, then wait on as many as we
428          * need in order of age. If we have enough events, stop waiting,
429          * and just check if any of the older ones are done.
430          */
431         if (sidx > 1)
432                 qsort(rbd->sort_events, sidx, sizeof(struct io_u *), rbd_io_u_cmp);
433
434         for (i = 0; i < sidx; i++) {
435                 io_u = rbd->sort_events[i];
436
437                 if (fri_check_complete(rbd, io_u, events)) {
438                         this_events++;
439                         continue;
440                 }
441
442                 /*
443                  * Stop waiting when we have enough, but continue checking
444                  * all pending IOs if they are complete.
445                  */
446                 if (*events >= min_evts)
447                         continue;
448
449                 rbd_io_u_wait_complete(io_u);
450
451                 if (fri_check_complete(rbd, io_u, events))
452                         this_events++;
453         }
454
455         return this_events;
456 }
457
458 static int fio_rbd_getevents(struct thread_data *td, unsigned int min,
459                              unsigned int max, const struct timespec *t)
460 {
461         unsigned int this_events, events = 0;
462         struct rbd_options *o = td->eo;
463         int wait = 0;
464
465         do {
466                 this_events = rbd_iter_events(td, &events, min, wait);
467
468                 if (events >= min)
469                         break;
470                 if (this_events)
471                         continue;
472
473                 if (!o->busy_poll)
474                         wait = 1;
475                 else
476                         nop;
477         } while (1);
478
479         return events;
480 }
481
482 static enum fio_q_status fio_rbd_queue(struct thread_data *td,
483                                        struct io_u *io_u)
484 {
485         struct rbd_data *rbd = td->io_ops_data;
486         struct fio_rbd_iou *fri = io_u->engine_data;
487         int r = -1;
488
489         fio_ro_check(td, io_u);
490
491         fri->io_seen = 0;
492         fri->io_complete = 0;
493
494         r = rbd_aio_create_completion(fri, _fio_rbd_finish_aiocb,
495                                                 &fri->completion);
496         if (r < 0) {
497                 log_err("rbd_aio_create_completion failed.\n");
498                 goto failed;
499         }
500
501         if (io_u->ddir == DDIR_WRITE) {
502                 r = rbd_aio_write(rbd->image, io_u->offset, io_u->xfer_buflen,
503                                          io_u->xfer_buf, fri->completion);
504                 if (r < 0) {
505                         log_err("rbd_aio_write failed.\n");
506                         goto failed_comp;
507                 }
508
509         } else if (io_u->ddir == DDIR_READ) {
510                 r = rbd_aio_read(rbd->image, io_u->offset, io_u->xfer_buflen,
511                                         io_u->xfer_buf, fri->completion);
512
513                 if (r < 0) {
514                         log_err("rbd_aio_read failed.\n");
515                         goto failed_comp;
516                 }
517         } else if (io_u->ddir == DDIR_TRIM) {
518                 r = rbd_aio_discard(rbd->image, io_u->offset,
519                                         io_u->xfer_buflen, fri->completion);
520                 if (r < 0) {
521                         log_err("rbd_aio_discard failed.\n");
522                         goto failed_comp;
523                 }
524         } else if (io_u->ddir == DDIR_SYNC) {
525                 r = rbd_aio_flush(rbd->image, fri->completion);
526                 if (r < 0) {
527                         log_err("rbd_flush failed.\n");
528                         goto failed_comp;
529                 }
530         } else {
531                 dprint(FD_IO, "%s: Warning: unhandled ddir: %d\n", __func__,
532                        io_u->ddir);
533                 r = -EINVAL;
534                 goto failed_comp;
535         }
536
537         return FIO_Q_QUEUED;
538 failed_comp:
539         rbd_aio_release(fri->completion);
540 failed:
541         io_u->error = -r;
542         td_verror(td, io_u->error, "xfer");
543         return FIO_Q_COMPLETED;
544 }
545
546 static int fio_rbd_init(struct thread_data *td)
547 {
548         int r;
549         struct rbd_data *rbd = td->io_ops_data;
550
551         if (rbd->connected)
552                 return 0;
553
554         r = _fio_rbd_connect(td);
555         if (r) {
556                 log_err("fio_rbd_connect failed, return code: %d .\n", r);
557                 goto failed;
558         }
559
560         return 0;
561
562 failed:
563         return 1;
564 }
565
566 static void fio_rbd_cleanup(struct thread_data *td)
567 {
568         struct rbd_data *rbd = td->io_ops_data;
569
570         if (rbd) {
571                 _fio_rbd_disconnect(rbd);
572                 free(rbd->aio_events);
573                 free(rbd->sort_events);
574                 free(rbd);
575         }
576 }
577
578 static int fio_rbd_setup(struct thread_data *td)
579 {
580         rbd_image_info_t info;
581         struct fio_file *f;
582         struct rbd_data *rbd = NULL;
583         int r;
584
585         /* allocate engine specific structure to deal with librbd. */
586         r = _fio_setup_rbd_data(td, &rbd);
587         if (r) {
588                 log_err("fio_setup_rbd_data failed.\n");
589                 goto cleanup;
590         }
591         td->io_ops_data = rbd;
592
593         /* librbd does not allow us to run first in the main thread and later
594          * in a fork child. It needs to be the same process context all the
595          * time. 
596          */
597         td->o.use_thread = 1;
598
599         /* connect in the main thread to determine to determine
600          * the size of the given RADOS block device. And disconnect
601          * later on.
602          */
603         r = _fio_rbd_connect(td);
604         if (r) {
605                 log_err("fio_rbd_connect failed.\n");
606                 goto cleanup;
607         }
608         rbd->connected = true;
609
610         /* get size of the RADOS block device */
611         r = rbd_stat(rbd->image, &info, sizeof(info));
612         if (r < 0) {
613                 log_err("rbd_status failed.\n");
614                 goto cleanup;
615         } else if (info.size == 0) {
616                 log_err("image size should be larger than zero.\n");
617                 r = -EINVAL;
618                 goto cleanup;
619         }
620
621         dprint(FD_IO, "rbd-engine: image size: %" PRIu64 "\n", info.size);
622
623         /* taken from "net" engine. Pretend we deal with files,
624          * even if we do not have any ideas about files.
625          * The size of the RBD is set instead of a artificial file.
626          */
627         if (!td->files_index) {
628                 add_file(td, td->o.filename ? : "rbd", 0, 0);
629                 td->o.nr_files = td->o.nr_files ? : 1;
630                 td->o.open_files++;
631         }
632         f = td->files[0];
633         f->real_file_size = info.size;
634
635         return 0;
636
637 cleanup:
638         fio_rbd_cleanup(td);
639         return r;
640 }
641
642 static int fio_rbd_open(struct thread_data *td, struct fio_file *f)
643 {
644         return 0;
645 }
646
647 static int fio_rbd_invalidate(struct thread_data *td, struct fio_file *f)
648 {
649 #if defined(CONFIG_RBD_INVAL)
650         struct rbd_data *rbd = td->io_ops_data;
651
652         return rbd_invalidate_cache(rbd->image);
653 #else
654         return 0;
655 #endif
656 }
657
658 static void fio_rbd_io_u_free(struct thread_data *td, struct io_u *io_u)
659 {
660         struct fio_rbd_iou *fri = io_u->engine_data;
661
662         if (fri) {
663                 io_u->engine_data = NULL;
664                 free(fri);
665         }
666 }
667
668 static int fio_rbd_io_u_init(struct thread_data *td, struct io_u *io_u)
669 {
670         struct fio_rbd_iou *fri;
671
672         fri = calloc(1, sizeof(*fri));
673         fri->io_u = io_u;
674         io_u->engine_data = fri;
675         return 0;
676 }
677
678 FIO_STATIC struct ioengine_ops ioengine = {
679         .name                   = "rbd",
680         .version                = FIO_IOOPS_VERSION,
681         .setup                  = fio_rbd_setup,
682         .init                   = fio_rbd_init,
683         .queue                  = fio_rbd_queue,
684         .getevents              = fio_rbd_getevents,
685         .event                  = fio_rbd_event,
686         .cleanup                = fio_rbd_cleanup,
687         .open_file              = fio_rbd_open,
688         .invalidate             = fio_rbd_invalidate,
689         .options                = options,
690         .io_u_init              = fio_rbd_io_u_init,
691         .io_u_free              = fio_rbd_io_u_free,
692         .option_struct_size     = sizeof(struct rbd_options),
693 };
694
695 static void fio_init fio_rbd_register(void)
696 {
697         register_ioengine(&ioengine);
698 }
699
700 static void fio_exit fio_rbd_unregister(void)
701 {
702         unregister_ioengine(&ioengine);
703 }