xref: /openbmc/qemu/block/rbd.c (revision cf83f140059f21d4629ae4b61d468c3baef2bb4c)
1  /*
2   * QEMU Block driver for RADOS (Ceph)
3   *
4   * Copyright (C) 2010-2011 Christian Brunner <chb@muc.de>,
5   *                         Josh Durgin <josh.durgin@dreamhost.com>
6   *
7   * This work is licensed under the terms of the GNU GPL, version 2.  See
8   * the COPYING file in the top-level directory.
9   *
10   * Contributions after 2012-01-13 are licensed under the terms of the
11   * GNU GPL, version 2 or (at your option) any later version.
12   */
13  
14  #include "qemu/osdep.h"
15  
16  #include <rbd/librbd.h>
17  #include "qapi/error.h"
18  #include "qemu/error-report.h"
19  #include "block/block_int.h"
20  #include "crypto/secret.h"
21  #include "qemu/cutils.h"
22  #include "qapi/qmp/qstring.h"
23  #include "qapi/qmp/qjson.h"
24  
25  /*
26   * When specifying the image filename use:
27   *
28   * rbd:poolname/devicename[@snapshotname][:option1=value1[:option2=value2...]]
29   *
30   * poolname must be the name of an existing rados pool.
31   *
32   * devicename is the name of the rbd image.
33   *
34   * Each option given is used to configure rados, and may be any valid
35   * Ceph option, "id", or "conf".
36   *
37   * The "id" option indicates what user we should authenticate as to
38   * the Ceph cluster.  If it is excluded we will use the Ceph default
39   * (normally 'admin').
40   *
41   * The "conf" option specifies a Ceph configuration file to read.  If
42   * it is not specified, we will read from the default Ceph locations
43   * (e.g., /etc/ceph/ceph.conf).  To avoid reading _any_ configuration
44   * file, specify conf=/dev/null.
45   *
46   * Configuration values containing :, @, or = can be escaped with a
47   * leading "\".
48   */
49  
50  /* rbd_aio_discard added in 0.1.2 */
51  #if LIBRBD_VERSION_CODE >= LIBRBD_VERSION(0, 1, 2)
52  #define LIBRBD_SUPPORTS_DISCARD
53  #else
54  #undef LIBRBD_SUPPORTS_DISCARD
55  #endif
56  
57  #define OBJ_MAX_SIZE (1UL << OBJ_DEFAULT_OBJ_ORDER)
58  
59  #define RBD_MAX_SNAPS 100
60  
61  /* The LIBRBD_SUPPORTS_IOVEC is defined in librbd.h */
62  #ifdef LIBRBD_SUPPORTS_IOVEC
63  #define LIBRBD_USE_IOVEC 1
64  #else
65  #define LIBRBD_USE_IOVEC 0
66  #endif
67  
68  typedef enum {
69      RBD_AIO_READ,
70      RBD_AIO_WRITE,
71      RBD_AIO_DISCARD,
72      RBD_AIO_FLUSH
73  } RBDAIOCmd;
74  
75  typedef struct RBDAIOCB {
76      BlockAIOCB common;
77      int64_t ret;
78      QEMUIOVector *qiov;
79      char *bounce;
80      RBDAIOCmd cmd;
81      int error;
82      struct BDRVRBDState *s;
83  } RBDAIOCB;
84  
85  typedef struct RADOSCB {
86      RBDAIOCB *acb;
87      struct BDRVRBDState *s;
88      int64_t size;
89      char *buf;
90      int64_t ret;
91  } RADOSCB;
92  
93  typedef struct BDRVRBDState {
94      rados_t cluster;
95      rados_ioctx_t io_ctx;
96      rbd_image_t image;
97      char *image_name;
98      char *snap;
99  } BDRVRBDState;
100  
101  static char *qemu_rbd_next_tok(char *src, char delim, char **p)
102  {
103      char *end;
104  
105      *p = NULL;
106  
107      for (end = src; *end; ++end) {
108          if (*end == delim) {
109              break;
110          }
111          if (*end == '\\' && end[1] != '\0') {
112              end++;
113          }
114      }
115      if (*end == delim) {
116          *p = end + 1;
117          *end = '\0';
118      }
119      return src;
120  }
121  
122  static void qemu_rbd_unescape(char *src)
123  {
124      char *p;
125  
126      for (p = src; *src; ++src, ++p) {
127          if (*src == '\\' && src[1] != '\0') {
128              src++;
129          }
130          *p = *src;
131      }
132      *p = '\0';
133  }
134  
135  static void qemu_rbd_parse_filename(const char *filename, QDict *options,
136                                      Error **errp)
137  {
138      const char *start;
139      char *p, *buf;
140      QList *keypairs = NULL;
141      char *found_str;
142  
143      if (!strstart(filename, "rbd:", &start)) {
144          error_setg(errp, "File name must start with 'rbd:'");
145          return;
146      }
147  
148      buf = g_strdup(start);
149      p = buf;
150  
151      found_str = qemu_rbd_next_tok(p, '/', &p);
152      if (!p) {
153          error_setg(errp, "Pool name is required");
154          goto done;
155      }
156      qemu_rbd_unescape(found_str);
157      qdict_put_str(options, "pool", found_str);
158  
159      if (strchr(p, '@')) {
160          found_str = qemu_rbd_next_tok(p, '@', &p);
161          qemu_rbd_unescape(found_str);
162          qdict_put_str(options, "image", found_str);
163  
164          found_str = qemu_rbd_next_tok(p, ':', &p);
165          qemu_rbd_unescape(found_str);
166          qdict_put_str(options, "snapshot", found_str);
167      } else {
168          found_str = qemu_rbd_next_tok(p, ':', &p);
169          qemu_rbd_unescape(found_str);
170          qdict_put_str(options, "image", found_str);
171      }
172      if (!p) {
173          goto done;
174      }
175  
176      /* The following are essentially all key/value pairs, and we treat
177       * 'id' and 'conf' a bit special.  Key/value pairs may be in any order. */
178      while (p) {
179          char *name, *value;
180          name = qemu_rbd_next_tok(p, '=', &p);
181          if (!p) {
182              error_setg(errp, "conf option %s has no value", name);
183              break;
184          }
185  
186          qemu_rbd_unescape(name);
187  
188          value = qemu_rbd_next_tok(p, ':', &p);
189          qemu_rbd_unescape(value);
190  
191          if (!strcmp(name, "conf")) {
192              qdict_put_str(options, "conf", value);
193          } else if (!strcmp(name, "id")) {
194              qdict_put_str(options, "user", value);
195          } else {
196              /*
197               * We pass these internally to qemu_rbd_set_keypairs(), so
198               * we can get away with the simpler list of [ "key1",
199               * "value1", "key2", "value2" ] rather than a raw dict
200               * { "key1": "value1", "key2": "value2" } where we can't
201               * guarantee order, or even a more correct but complex
202               * [ { "key1": "value1" }, { "key2": "value2" } ]
203               */
204              if (!keypairs) {
205                  keypairs = qlist_new();
206              }
207              qlist_append_str(keypairs, name);
208              qlist_append_str(keypairs, value);
209          }
210      }
211  
212      if (keypairs) {
213          qdict_put(options, "=keyvalue-pairs",
214                    qobject_to_json(QOBJECT(keypairs)));
215      }
216  
217  done:
218      g_free(buf);
219      QDECREF(keypairs);
220      return;
221  }
222  
223  
224  static int qemu_rbd_set_auth(rados_t cluster, const char *secretid,
225                               Error **errp)
226  {
227      if (secretid == 0) {
228          return 0;
229      }
230  
231      gchar *secret = qcrypto_secret_lookup_as_base64(secretid,
232                                                      errp);
233      if (!secret) {
234          return -1;
235      }
236  
237      rados_conf_set(cluster, "key", secret);
238      g_free(secret);
239  
240      return 0;
241  }
242  
243  static int qemu_rbd_set_keypairs(rados_t cluster, const char *keypairs_json,
244                                   Error **errp)
245  {
246      QList *keypairs;
247      QString *name;
248      QString *value;
249      const char *key;
250      size_t remaining;
251      int ret = 0;
252  
253      if (!keypairs_json) {
254          return ret;
255      }
256      keypairs = qobject_to_qlist(qobject_from_json(keypairs_json,
257                                                    &error_abort));
258      remaining = qlist_size(keypairs) / 2;
259      assert(remaining);
260  
261      while (remaining--) {
262          name = qobject_to_qstring(qlist_pop(keypairs));
263          value = qobject_to_qstring(qlist_pop(keypairs));
264          assert(name && value);
265          key = qstring_get_str(name);
266  
267          ret = rados_conf_set(cluster, key, qstring_get_str(value));
268          QDECREF(name);
269          QDECREF(value);
270          if (ret < 0) {
271              error_setg_errno(errp, -ret, "invalid conf option %s", key);
272              ret = -EINVAL;
273              break;
274          }
275      }
276  
277      QDECREF(keypairs);
278      return ret;
279  }
280  
281  static void qemu_rbd_memset(RADOSCB *rcb, int64_t offs)
282  {
283      if (LIBRBD_USE_IOVEC) {
284          RBDAIOCB *acb = rcb->acb;
285          iov_memset(acb->qiov->iov, acb->qiov->niov, offs, 0,
286                     acb->qiov->size - offs);
287      } else {
288          memset(rcb->buf + offs, 0, rcb->size - offs);
289      }
290  }
291  
292  static QemuOptsList runtime_opts = {
293      .name = "rbd",
294      .head = QTAILQ_HEAD_INITIALIZER(runtime_opts.head),
295      .desc = {
296          {
297              .name = "pool",
298              .type = QEMU_OPT_STRING,
299              .help = "Rados pool name",
300          },
301          {
302              .name = "image",
303              .type = QEMU_OPT_STRING,
304              .help = "Image name in the pool",
305          },
306          {
307              .name = "conf",
308              .type = QEMU_OPT_STRING,
309              .help = "Rados config file location",
310          },
311          {
312              .name = "snapshot",
313              .type = QEMU_OPT_STRING,
314              .help = "Ceph snapshot name",
315          },
316          {
317              /* maps to 'id' in rados_create() */
318              .name = "user",
319              .type = QEMU_OPT_STRING,
320              .help = "Rados id name",
321          },
322          /*
323           * server.* extracted manually, see qemu_rbd_mon_host()
324           */
325          {
326              .name = "password-secret",
327              .type = QEMU_OPT_STRING,
328              .help = "ID of secret providing the password",
329          },
330  
331          /*
332           * Keys for qemu_rbd_parse_filename(), not in the QAPI schema
333           */
334          {
335              /*
336               * HACK: name starts with '=' so that qemu_opts_parse()
337               * can't set it
338               */
339              .name = "=keyvalue-pairs",
340              .type = QEMU_OPT_STRING,
341              .help = "Legacy rados key/value option parameters",
342          },
343          { /* end of list */ }
344      },
345  };
346  
347  static int qemu_rbd_create(const char *filename, QemuOpts *opts, Error **errp)
348  {
349      Error *local_err = NULL;
350      int64_t bytes = 0;
351      int64_t objsize;
352      int obj_order = 0;
353      const char *pool, *image_name, *conf, *user, *keypairs;
354      const char *secretid;
355      rados_t cluster;
356      rados_ioctx_t io_ctx;
357      QDict *options = NULL;
358      int ret = 0;
359  
360      secretid = qemu_opt_get(opts, "password-secret");
361  
362      /* Read out options */
363      bytes = ROUND_UP(qemu_opt_get_size_del(opts, BLOCK_OPT_SIZE, 0),
364                       BDRV_SECTOR_SIZE);
365      objsize = qemu_opt_get_size_del(opts, BLOCK_OPT_CLUSTER_SIZE, 0);
366      if (objsize) {
367          if ((objsize - 1) & objsize) {    /* not a power of 2? */
368              error_setg(errp, "obj size needs to be power of 2");
369              ret = -EINVAL;
370              goto exit;
371          }
372          if (objsize < 4096) {
373              error_setg(errp, "obj size too small");
374              ret = -EINVAL;
375              goto exit;
376          }
377          obj_order = ctz32(objsize);
378      }
379  
380      options = qdict_new();
381      qemu_rbd_parse_filename(filename, options, &local_err);
382      if (local_err) {
383          ret = -EINVAL;
384          error_propagate(errp, local_err);
385          goto exit;
386      }
387  
388      /*
389       * Caution: while qdict_get_try_str() is fine, getting non-string
390       * types would require more care.  When @options come from -blockdev
391       * or blockdev_add, its members are typed according to the QAPI
392       * schema, but when they come from -drive, they're all QString.
393       */
394      pool       = qdict_get_try_str(options, "pool");
395      conf       = qdict_get_try_str(options, "conf");
396      user       = qdict_get_try_str(options, "user");
397      image_name = qdict_get_try_str(options, "image");
398      keypairs   = qdict_get_try_str(options, "=keyvalue-pairs");
399  
400      ret = rados_create(&cluster, user);
401      if (ret < 0) {
402          error_setg_errno(errp, -ret, "error initializing");
403          goto exit;
404      }
405  
406      /* try default location when conf=NULL, but ignore failure */
407      ret = rados_conf_read_file(cluster, conf);
408      if (conf && ret < 0) {
409          error_setg_errno(errp, -ret, "error reading conf file %s", conf);
410          ret = -EIO;
411          goto shutdown;
412      }
413  
414      ret = qemu_rbd_set_keypairs(cluster, keypairs, errp);
415      if (ret < 0) {
416          ret = -EIO;
417          goto shutdown;
418      }
419  
420      if (qemu_rbd_set_auth(cluster, secretid, errp) < 0) {
421          ret = -EIO;
422          goto shutdown;
423      }
424  
425      ret = rados_connect(cluster);
426      if (ret < 0) {
427          error_setg_errno(errp, -ret, "error connecting");
428          goto shutdown;
429      }
430  
431      ret = rados_ioctx_create(cluster, pool, &io_ctx);
432      if (ret < 0) {
433          error_setg_errno(errp, -ret, "error opening pool %s", pool);
434          goto shutdown;
435      }
436  
437      ret = rbd_create(io_ctx, image_name, bytes, &obj_order);
438      if (ret < 0) {
439          error_setg_errno(errp, -ret, "error rbd create");
440      }
441  
442      rados_ioctx_destroy(io_ctx);
443  
444  shutdown:
445      rados_shutdown(cluster);
446  
447  exit:
448      QDECREF(options);
449      return ret;
450  }
451  
452  /*
453   * This aio completion is being called from rbd_finish_bh() and runs in qemu
454   * BH context.
455   */
456  static void qemu_rbd_complete_aio(RADOSCB *rcb)
457  {
458      RBDAIOCB *acb = rcb->acb;
459      int64_t r;
460  
461      r = rcb->ret;
462  
463      if (acb->cmd != RBD_AIO_READ) {
464          if (r < 0) {
465              acb->ret = r;
466              acb->error = 1;
467          } else if (!acb->error) {
468              acb->ret = rcb->size;
469          }
470      } else {
471          if (r < 0) {
472              qemu_rbd_memset(rcb, 0);
473              acb->ret = r;
474              acb->error = 1;
475          } else if (r < rcb->size) {
476              qemu_rbd_memset(rcb, r);
477              if (!acb->error) {
478                  acb->ret = rcb->size;
479              }
480          } else if (!acb->error) {
481              acb->ret = r;
482          }
483      }
484  
485      g_free(rcb);
486  
487      if (!LIBRBD_USE_IOVEC) {
488          if (acb->cmd == RBD_AIO_READ) {
489              qemu_iovec_from_buf(acb->qiov, 0, acb->bounce, acb->qiov->size);
490          }
491          qemu_vfree(acb->bounce);
492      }
493  
494      acb->common.cb(acb->common.opaque, (acb->ret > 0 ? 0 : acb->ret));
495  
496      qemu_aio_unref(acb);
497  }
498  
499  static char *qemu_rbd_mon_host(QDict *options, Error **errp)
500  {
501      const char **vals = g_new(const char *, qdict_size(options) + 1);
502      char keybuf[32];
503      const char *host, *port;
504      char *rados_str;
505      int i;
506  
507      for (i = 0;; i++) {
508          sprintf(keybuf, "server.%d.host", i);
509          host = qdict_get_try_str(options, keybuf);
510          qdict_del(options, keybuf);
511          sprintf(keybuf, "server.%d.port", i);
512          port = qdict_get_try_str(options, keybuf);
513          qdict_del(options, keybuf);
514          if (!host && !port) {
515              break;
516          }
517          if (!host) {
518              error_setg(errp, "Parameter server.%d.host is missing", i);
519              rados_str = NULL;
520              goto out;
521          }
522  
523          if (strchr(host, ':')) {
524              vals[i] = port ? g_strdup_printf("[%s]:%s", host, port)
525                  : g_strdup_printf("[%s]", host);
526          } else {
527              vals[i] = port ? g_strdup_printf("%s:%s", host, port)
528                  : g_strdup(host);
529          }
530      }
531      vals[i] = NULL;
532  
533      rados_str = i ? g_strjoinv(";", (char **)vals) : NULL;
534  out:
535      g_strfreev((char **)vals);
536      return rados_str;
537  }
538  
539  static int qemu_rbd_open(BlockDriverState *bs, QDict *options, int flags,
540                           Error **errp)
541  {
542      BDRVRBDState *s = bs->opaque;
543      const char *pool, *snap, *conf, *user, *image_name, *keypairs;
544      const char *secretid;
545      QemuOpts *opts;
546      Error *local_err = NULL;
547      char *mon_host = NULL;
548      int r;
549  
550      opts = qemu_opts_create(&runtime_opts, NULL, 0, &error_abort);
551      qemu_opts_absorb_qdict(opts, options, &local_err);
552      if (local_err) {
553          error_propagate(errp, local_err);
554          r = -EINVAL;
555          goto failed_opts;
556      }
557  
558      mon_host = qemu_rbd_mon_host(options, &local_err);
559      if (local_err) {
560          error_propagate(errp, local_err);
561          r = -EINVAL;
562          goto failed_opts;
563      }
564  
565      secretid = qemu_opt_get(opts, "password-secret");
566  
567      pool           = qemu_opt_get(opts, "pool");
568      conf           = qemu_opt_get(opts, "conf");
569      snap           = qemu_opt_get(opts, "snapshot");
570      user           = qemu_opt_get(opts, "user");
571      image_name     = qemu_opt_get(opts, "image");
572      keypairs       = qemu_opt_get(opts, "=keyvalue-pairs");
573  
574      if (!pool || !image_name) {
575          error_setg(errp, "Parameters 'pool' and 'image' are required");
576          r = -EINVAL;
577          goto failed_opts;
578      }
579  
580      r = rados_create(&s->cluster, user);
581      if (r < 0) {
582          error_setg_errno(errp, -r, "error initializing");
583          goto failed_opts;
584      }
585  
586      s->snap = g_strdup(snap);
587      s->image_name = g_strdup(image_name);
588  
589      /* try default location when conf=NULL, but ignore failure */
590      r = rados_conf_read_file(s->cluster, conf);
591      if (conf && r < 0) {
592          error_setg_errno(errp, -r, "error reading conf file %s", conf);
593          goto failed_shutdown;
594      }
595  
596      r = qemu_rbd_set_keypairs(s->cluster, keypairs, errp);
597      if (r < 0) {
598          goto failed_shutdown;
599      }
600  
601      if (mon_host) {
602          r = rados_conf_set(s->cluster, "mon_host", mon_host);
603          if (r < 0) {
604              goto failed_shutdown;
605          }
606      }
607  
608      if (qemu_rbd_set_auth(s->cluster, secretid, errp) < 0) {
609          r = -EIO;
610          goto failed_shutdown;
611      }
612  
613      /*
614       * Fallback to more conservative semantics if setting cache
615       * options fails. Ignore errors from setting rbd_cache because the
616       * only possible error is that the option does not exist, and
617       * librbd defaults to no caching. If write through caching cannot
618       * be set up, fall back to no caching.
619       */
620      if (flags & BDRV_O_NOCACHE) {
621          rados_conf_set(s->cluster, "rbd_cache", "false");
622      } else {
623          rados_conf_set(s->cluster, "rbd_cache", "true");
624      }
625  
626      r = rados_connect(s->cluster);
627      if (r < 0) {
628          error_setg_errno(errp, -r, "error connecting");
629          goto failed_shutdown;
630      }
631  
632      r = rados_ioctx_create(s->cluster, pool, &s->io_ctx);
633      if (r < 0) {
634          error_setg_errno(errp, -r, "error opening pool %s", pool);
635          goto failed_shutdown;
636      }
637  
638      /* rbd_open is always r/w */
639      r = rbd_open(s->io_ctx, s->image_name, &s->image, s->snap);
640      if (r < 0) {
641          error_setg_errno(errp, -r, "error reading header from %s",
642                           s->image_name);
643          goto failed_open;
644      }
645  
646      /* If we are using an rbd snapshot, we must be r/o, otherwise
647       * leave as-is */
648      if (s->snap != NULL) {
649          r = bdrv_set_read_only(bs, true, &local_err);
650          if (r < 0) {
651              error_propagate(errp, local_err);
652              goto failed_open;
653          }
654      }
655  
656      qemu_opts_del(opts);
657      return 0;
658  
659  failed_open:
660      rados_ioctx_destroy(s->io_ctx);
661  failed_shutdown:
662      rados_shutdown(s->cluster);
663      g_free(s->snap);
664      g_free(s->image_name);
665  failed_opts:
666      qemu_opts_del(opts);
667      g_free(mon_host);
668      return r;
669  }
670  
671  
672  /* Since RBD is currently always opened R/W via the API,
673   * we just need to check if we are using a snapshot or not, in
674   * order to determine if we will allow it to be R/W */
675  static int qemu_rbd_reopen_prepare(BDRVReopenState *state,
676                                     BlockReopenQueue *queue, Error **errp)
677  {
678      BDRVRBDState *s = state->bs->opaque;
679      int ret = 0;
680  
681      if (s->snap && state->flags & BDRV_O_RDWR) {
682          error_setg(errp,
683                     "Cannot change node '%s' to r/w when using RBD snapshot",
684                     bdrv_get_device_or_node_name(state->bs));
685          ret = -EINVAL;
686      }
687  
688      return ret;
689  }
690  
691  static void qemu_rbd_close(BlockDriverState *bs)
692  {
693      BDRVRBDState *s = bs->opaque;
694  
695      rbd_close(s->image);
696      rados_ioctx_destroy(s->io_ctx);
697      g_free(s->snap);
698      g_free(s->image_name);
699      rados_shutdown(s->cluster);
700  }
701  
702  static const AIOCBInfo rbd_aiocb_info = {
703      .aiocb_size = sizeof(RBDAIOCB),
704  };
705  
706  static void rbd_finish_bh(void *opaque)
707  {
708      RADOSCB *rcb = opaque;
709      qemu_rbd_complete_aio(rcb);
710  }
711  
712  /*
713   * This is the callback function for rbd_aio_read and _write
714   *
715   * Note: this function is being called from a non qemu thread so
716   * we need to be careful about what we do here. Generally we only
717   * schedule a BH, and do the rest of the io completion handling
718   * from rbd_finish_bh() which runs in a qemu context.
719   */
720  static void rbd_finish_aiocb(rbd_completion_t c, RADOSCB *rcb)
721  {
722      RBDAIOCB *acb = rcb->acb;
723  
724      rcb->ret = rbd_aio_get_return_value(c);
725      rbd_aio_release(c);
726  
727      aio_bh_schedule_oneshot(bdrv_get_aio_context(acb->common.bs),
728                              rbd_finish_bh, rcb);
729  }
730  
731  static int rbd_aio_discard_wrapper(rbd_image_t image,
732                                     uint64_t off,
733                                     uint64_t len,
734                                     rbd_completion_t comp)
735  {
736  #ifdef LIBRBD_SUPPORTS_DISCARD
737      return rbd_aio_discard(image, off, len, comp);
738  #else
739      return -ENOTSUP;
740  #endif
741  }
742  
743  static int rbd_aio_flush_wrapper(rbd_image_t image,
744                                   rbd_completion_t comp)
745  {
746  #ifdef LIBRBD_SUPPORTS_AIO_FLUSH
747      return rbd_aio_flush(image, comp);
748  #else
749      return -ENOTSUP;
750  #endif
751  }
752  
753  static BlockAIOCB *rbd_start_aio(BlockDriverState *bs,
754                                   int64_t off,
755                                   QEMUIOVector *qiov,
756                                   int64_t size,
757                                   BlockCompletionFunc *cb,
758                                   void *opaque,
759                                   RBDAIOCmd cmd)
760  {
761      RBDAIOCB *acb;
762      RADOSCB *rcb = NULL;
763      rbd_completion_t c;
764      int r;
765  
766      BDRVRBDState *s = bs->opaque;
767  
768      acb = qemu_aio_get(&rbd_aiocb_info, bs, cb, opaque);
769      acb->cmd = cmd;
770      acb->qiov = qiov;
771      assert(!qiov || qiov->size == size);
772  
773      rcb = g_new(RADOSCB, 1);
774  
775      if (!LIBRBD_USE_IOVEC) {
776          if (cmd == RBD_AIO_DISCARD || cmd == RBD_AIO_FLUSH) {
777              acb->bounce = NULL;
778          } else {
779              acb->bounce = qemu_try_blockalign(bs, qiov->size);
780              if (acb->bounce == NULL) {
781                  goto failed;
782              }
783          }
784          if (cmd == RBD_AIO_WRITE) {
785              qemu_iovec_to_buf(acb->qiov, 0, acb->bounce, qiov->size);
786          }
787          rcb->buf = acb->bounce;
788      }
789  
790      acb->ret = 0;
791      acb->error = 0;
792      acb->s = s;
793  
794      rcb->acb = acb;
795      rcb->s = acb->s;
796      rcb->size = size;
797      r = rbd_aio_create_completion(rcb, (rbd_callback_t) rbd_finish_aiocb, &c);
798      if (r < 0) {
799          goto failed;
800      }
801  
802      switch (cmd) {
803      case RBD_AIO_WRITE:
804  #ifdef LIBRBD_SUPPORTS_IOVEC
805              r = rbd_aio_writev(s->image, qiov->iov, qiov->niov, off, c);
806  #else
807              r = rbd_aio_write(s->image, off, size, rcb->buf, c);
808  #endif
809          break;
810      case RBD_AIO_READ:
811  #ifdef LIBRBD_SUPPORTS_IOVEC
812              r = rbd_aio_readv(s->image, qiov->iov, qiov->niov, off, c);
813  #else
814              r = rbd_aio_read(s->image, off, size, rcb->buf, c);
815  #endif
816          break;
817      case RBD_AIO_DISCARD:
818          r = rbd_aio_discard_wrapper(s->image, off, size, c);
819          break;
820      case RBD_AIO_FLUSH:
821          r = rbd_aio_flush_wrapper(s->image, c);
822          break;
823      default:
824          r = -EINVAL;
825      }
826  
827      if (r < 0) {
828          goto failed_completion;
829      }
830      return &acb->common;
831  
832  failed_completion:
833      rbd_aio_release(c);
834  failed:
835      g_free(rcb);
836      if (!LIBRBD_USE_IOVEC) {
837          qemu_vfree(acb->bounce);
838      }
839  
840      qemu_aio_unref(acb);
841      return NULL;
842  }
843  
844  static BlockAIOCB *qemu_rbd_aio_readv(BlockDriverState *bs,
845                                        int64_t sector_num,
846                                        QEMUIOVector *qiov,
847                                        int nb_sectors,
848                                        BlockCompletionFunc *cb,
849                                        void *opaque)
850  {
851      return rbd_start_aio(bs, sector_num << BDRV_SECTOR_BITS, qiov,
852                           (int64_t) nb_sectors << BDRV_SECTOR_BITS, cb, opaque,
853                           RBD_AIO_READ);
854  }
855  
856  static BlockAIOCB *qemu_rbd_aio_writev(BlockDriverState *bs,
857                                         int64_t sector_num,
858                                         QEMUIOVector *qiov,
859                                         int nb_sectors,
860                                         BlockCompletionFunc *cb,
861                                         void *opaque)
862  {
863      return rbd_start_aio(bs, sector_num << BDRV_SECTOR_BITS, qiov,
864                           (int64_t) nb_sectors << BDRV_SECTOR_BITS, cb, opaque,
865                           RBD_AIO_WRITE);
866  }
867  
868  #ifdef LIBRBD_SUPPORTS_AIO_FLUSH
869  static BlockAIOCB *qemu_rbd_aio_flush(BlockDriverState *bs,
870                                        BlockCompletionFunc *cb,
871                                        void *opaque)
872  {
873      return rbd_start_aio(bs, 0, NULL, 0, cb, opaque, RBD_AIO_FLUSH);
874  }
875  
876  #else
877  
878  static int qemu_rbd_co_flush(BlockDriverState *bs)
879  {
880  #if LIBRBD_VERSION_CODE >= LIBRBD_VERSION(0, 1, 1)
881      /* rbd_flush added in 0.1.1 */
882      BDRVRBDState *s = bs->opaque;
883      return rbd_flush(s->image);
884  #else
885      return 0;
886  #endif
887  }
888  #endif
889  
890  static int qemu_rbd_getinfo(BlockDriverState *bs, BlockDriverInfo *bdi)
891  {
892      BDRVRBDState *s = bs->opaque;
893      rbd_image_info_t info;
894      int r;
895  
896      r = rbd_stat(s->image, &info, sizeof(info));
897      if (r < 0) {
898          return r;
899      }
900  
901      bdi->cluster_size = info.obj_size;
902      return 0;
903  }
904  
905  static int64_t qemu_rbd_getlength(BlockDriverState *bs)
906  {
907      BDRVRBDState *s = bs->opaque;
908      rbd_image_info_t info;
909      int r;
910  
911      r = rbd_stat(s->image, &info, sizeof(info));
912      if (r < 0) {
913          return r;
914      }
915  
916      return info.size;
917  }
918  
919  static int qemu_rbd_truncate(BlockDriverState *bs, int64_t offset, Error **errp)
920  {
921      BDRVRBDState *s = bs->opaque;
922      int r;
923  
924      r = rbd_resize(s->image, offset);
925      if (r < 0) {
926          error_setg_errno(errp, -r, "Failed to resize file");
927          return r;
928      }
929  
930      return 0;
931  }
932  
933  static int qemu_rbd_snap_create(BlockDriverState *bs,
934                                  QEMUSnapshotInfo *sn_info)
935  {
936      BDRVRBDState *s = bs->opaque;
937      int r;
938  
939      if (sn_info->name[0] == '\0') {
940          return -EINVAL; /* we need a name for rbd snapshots */
941      }
942  
943      /*
944       * rbd snapshots are using the name as the user controlled unique identifier
945       * we can't use the rbd snapid for that purpose, as it can't be set
946       */
947      if (sn_info->id_str[0] != '\0' &&
948          strcmp(sn_info->id_str, sn_info->name) != 0) {
949          return -EINVAL;
950      }
951  
952      if (strlen(sn_info->name) >= sizeof(sn_info->id_str)) {
953          return -ERANGE;
954      }
955  
956      r = rbd_snap_create(s->image, sn_info->name);
957      if (r < 0) {
958          error_report("failed to create snap: %s", strerror(-r));
959          return r;
960      }
961  
962      return 0;
963  }
964  
965  static int qemu_rbd_snap_remove(BlockDriverState *bs,
966                                  const char *snapshot_id,
967                                  const char *snapshot_name,
968                                  Error **errp)
969  {
970      BDRVRBDState *s = bs->opaque;
971      int r;
972  
973      if (!snapshot_name) {
974          error_setg(errp, "rbd need a valid snapshot name");
975          return -EINVAL;
976      }
977  
978      /* If snapshot_id is specified, it must be equal to name, see
979         qemu_rbd_snap_list() */
980      if (snapshot_id && strcmp(snapshot_id, snapshot_name)) {
981          error_setg(errp,
982                     "rbd do not support snapshot id, it should be NULL or "
983                     "equal to snapshot name");
984          return -EINVAL;
985      }
986  
987      r = rbd_snap_remove(s->image, snapshot_name);
988      if (r < 0) {
989          error_setg_errno(errp, -r, "Failed to remove the snapshot");
990      }
991      return r;
992  }
993  
994  static int qemu_rbd_snap_rollback(BlockDriverState *bs,
995                                    const char *snapshot_name)
996  {
997      BDRVRBDState *s = bs->opaque;
998  
999      return rbd_snap_rollback(s->image, snapshot_name);
1000  }
1001  
1002  static int qemu_rbd_snap_list(BlockDriverState *bs,
1003                                QEMUSnapshotInfo **psn_tab)
1004  {
1005      BDRVRBDState *s = bs->opaque;
1006      QEMUSnapshotInfo *sn_info, *sn_tab = NULL;
1007      int i, snap_count;
1008      rbd_snap_info_t *snaps;
1009      int max_snaps = RBD_MAX_SNAPS;
1010  
1011      do {
1012          snaps = g_new(rbd_snap_info_t, max_snaps);
1013          snap_count = rbd_snap_list(s->image, snaps, &max_snaps);
1014          if (snap_count <= 0) {
1015              g_free(snaps);
1016          }
1017      } while (snap_count == -ERANGE);
1018  
1019      if (snap_count <= 0) {
1020          goto done;
1021      }
1022  
1023      sn_tab = g_new0(QEMUSnapshotInfo, snap_count);
1024  
1025      for (i = 0; i < snap_count; i++) {
1026          const char *snap_name = snaps[i].name;
1027  
1028          sn_info = sn_tab + i;
1029          pstrcpy(sn_info->id_str, sizeof(sn_info->id_str), snap_name);
1030          pstrcpy(sn_info->name, sizeof(sn_info->name), snap_name);
1031  
1032          sn_info->vm_state_size = snaps[i].size;
1033          sn_info->date_sec = 0;
1034          sn_info->date_nsec = 0;
1035          sn_info->vm_clock_nsec = 0;
1036      }
1037      rbd_snap_list_end(snaps);
1038      g_free(snaps);
1039  
1040   done:
1041      *psn_tab = sn_tab;
1042      return snap_count;
1043  }
1044  
1045  #ifdef LIBRBD_SUPPORTS_DISCARD
1046  static BlockAIOCB *qemu_rbd_aio_pdiscard(BlockDriverState *bs,
1047                                           int64_t offset,
1048                                           int count,
1049                                           BlockCompletionFunc *cb,
1050                                           void *opaque)
1051  {
1052      return rbd_start_aio(bs, offset, NULL, count, cb, opaque,
1053                           RBD_AIO_DISCARD);
1054  }
1055  #endif
1056  
1057  #ifdef LIBRBD_SUPPORTS_INVALIDATE
1058  static void qemu_rbd_invalidate_cache(BlockDriverState *bs,
1059                                        Error **errp)
1060  {
1061      BDRVRBDState *s = bs->opaque;
1062      int r = rbd_invalidate_cache(s->image);
1063      if (r < 0) {
1064          error_setg_errno(errp, -r, "Failed to invalidate the cache");
1065      }
1066  }
1067  #endif
1068  
1069  static QemuOptsList qemu_rbd_create_opts = {
1070      .name = "rbd-create-opts",
1071      .head = QTAILQ_HEAD_INITIALIZER(qemu_rbd_create_opts.head),
1072      .desc = {
1073          {
1074              .name = BLOCK_OPT_SIZE,
1075              .type = QEMU_OPT_SIZE,
1076              .help = "Virtual disk size"
1077          },
1078          {
1079              .name = BLOCK_OPT_CLUSTER_SIZE,
1080              .type = QEMU_OPT_SIZE,
1081              .help = "RBD object size"
1082          },
1083          {
1084              .name = "password-secret",
1085              .type = QEMU_OPT_STRING,
1086              .help = "ID of secret providing the password",
1087          },
1088          { /* end of list */ }
1089      }
1090  };
1091  
1092  static BlockDriver bdrv_rbd = {
1093      .format_name            = "rbd",
1094      .instance_size          = sizeof(BDRVRBDState),
1095      .bdrv_parse_filename    = qemu_rbd_parse_filename,
1096      .bdrv_file_open         = qemu_rbd_open,
1097      .bdrv_close             = qemu_rbd_close,
1098      .bdrv_reopen_prepare    = qemu_rbd_reopen_prepare,
1099      .bdrv_create            = qemu_rbd_create,
1100      .bdrv_has_zero_init     = bdrv_has_zero_init_1,
1101      .bdrv_get_info          = qemu_rbd_getinfo,
1102      .create_opts            = &qemu_rbd_create_opts,
1103      .bdrv_getlength         = qemu_rbd_getlength,
1104      .bdrv_truncate          = qemu_rbd_truncate,
1105      .protocol_name          = "rbd",
1106  
1107      .bdrv_aio_readv         = qemu_rbd_aio_readv,
1108      .bdrv_aio_writev        = qemu_rbd_aio_writev,
1109  
1110  #ifdef LIBRBD_SUPPORTS_AIO_FLUSH
1111      .bdrv_aio_flush         = qemu_rbd_aio_flush,
1112  #else
1113      .bdrv_co_flush_to_disk  = qemu_rbd_co_flush,
1114  #endif
1115  
1116  #ifdef LIBRBD_SUPPORTS_DISCARD
1117      .bdrv_aio_pdiscard      = qemu_rbd_aio_pdiscard,
1118  #endif
1119  
1120      .bdrv_snapshot_create   = qemu_rbd_snap_create,
1121      .bdrv_snapshot_delete   = qemu_rbd_snap_remove,
1122      .bdrv_snapshot_list     = qemu_rbd_snap_list,
1123      .bdrv_snapshot_goto     = qemu_rbd_snap_rollback,
1124  #ifdef LIBRBD_SUPPORTS_INVALIDATE
1125      .bdrv_invalidate_cache  = qemu_rbd_invalidate_cache,
1126  #endif
1127  };
1128  
1129  static void bdrv_rbd_init(void)
1130  {
1131      bdrv_register(&bdrv_rbd);
1132  }
1133  
1134  block_init(bdrv_rbd_init);
1135