block/gluster.c

   1 /*
   2  * GlusterFS backend for QEMU
   3  *
   4  * Copyright (C) 2012 Bharata B Rao <[email protected]>
   5  *
   6  * This work is licensed under the terms of the GNU GPL, version 2 or later.
   7  * See the COPYING file in the top-level directory.
   8  *
   9  */
  10 #include "qemu/osdep.h"
  11 #include <glusterfs/api/glfs.h>
  12 #include "block/block_int.h"
  13 #include "qapi/error.h"
  14 #include "qapi/qmp/qerror.h"
  15 #include "qapi/util.h"
  16 #include "qemu/uri.h"
  17 #include "qemu/error-report.h"
  18 #include "qemu/cutils.h"
  19
  20 #define GLUSTER_OPT_FILENAME        "filename"
  21 #define GLUSTER_OPT_VOLUME          "volume"
  22 #define GLUSTER_OPT_PATH            "path"
  23 #define GLUSTER_OPT_TYPE            "type"
  24 #define GLUSTER_OPT_SERVER_PATTERN  "server."
  25 #define GLUSTER_OPT_HOST            "host"
  26 #define GLUSTER_OPT_PORT            "port"
  27 #define GLUSTER_OPT_TO              "to"
  28 #define GLUSTER_OPT_IPV4            "ipv4"
  29 #define GLUSTER_OPT_IPV6            "ipv6"
  30 #define GLUSTER_OPT_SOCKET          "socket"
  31 #define GLUSTER_OPT_DEBUG           "debug"
  32 #define GLUSTER_DEFAULT_PORT        24007
  33 #define GLUSTER_DEBUG_DEFAULT       4
  34 #define GLUSTER_DEBUG_MAX           9
  35 #define GLUSTER_OPT_LOGFILE         "logfile"
  36 #define GLUSTER_LOGFILE_DEFAULT     "-" /* handled in libgfapi as /dev/stderr */
  37
  38 #define GERR_INDEX_HINT "hint: check in 'server' array index '%d'\n"
  39
  40 typedef struct GlusterAIOCB {
  41     int64_t size;
  42     int ret;
  43     Coroutine *coroutine;
  44     AioContext *aio_context;
  45 } GlusterAIOCB;
  46
  47 typedef struct BDRVGlusterState {
  48     struct glfs *glfs;
  49     struct glfs_fd *fd;
  50     char *logfile;
  51     bool supports_seek_data;
  52     int debug;
  53 } BDRVGlusterState;
  54
  55 typedef struct BDRVGlusterReopenState {
  56     struct glfs *glfs;
  57     struct glfs_fd *fd;
  58 } BDRVGlusterReopenState;
  59
  60
  61 typedef struct GlfsPreopened {
  62     char *volume;
  63     glfs_t *fs;
  64     int ref;
  65 } GlfsPreopened;
  66
  67 typedef struct ListElement {
  68     QLIST_ENTRY(ListElement) list;
  69     GlfsPreopened saved;
  70 } ListElement;
  71
  72 static QLIST_HEAD(glfs_list, ListElement) glfs_list;
  73
  74 static QemuOptsList qemu_gluster_create_opts = {
  75     .name = "qemu-gluster-create-opts",
  76     .head = QTAILQ_HEAD_INITIALIZER(qemu_gluster_create_opts.head),
  77     .desc = {
  78         {
  79             .name = BLOCK_OPT_SIZE,
  80             .type = QEMU_OPT_SIZE,
  81             .help = "Virtual disk size"
  82         },
  83         {
  84             .name = BLOCK_OPT_PREALLOC,
  85             .type = QEMU_OPT_STRING,
  86             .help = "Preallocation mode (allowed values: off, full)"
  87         },
  88         {
  89             .name = GLUSTER_OPT_DEBUG,
  90             .type = QEMU_OPT_NUMBER,
  91             .help = "Gluster log level, valid range is 0-9",
  92         },
  93         {
  94             .name = GLUSTER_OPT_LOGFILE,
  95             .type = QEMU_OPT_STRING,
  96             .help = "Logfile path of libgfapi",
  97         },
  98         { /* end of list */ }
  99     }
 100 };
 101
 102 static QemuOptsList runtime_opts = {
 103     .name = "gluster",
 104     .head = QTAILQ_HEAD_INITIALIZER(runtime_opts.head),
 105     .desc = {
 106         {
 107             .name = GLUSTER_OPT_FILENAME,
 108             .type = QEMU_OPT_STRING,
 109             .help = "URL to the gluster image",
 110         },
 111         {
 112             .name = GLUSTER_OPT_DEBUG,
 113             .type = QEMU_OPT_NUMBER,
 114             .help = "Gluster log level, valid range is 0-9",
 115         },
 116         {
 117             .name = GLUSTER_OPT_LOGFILE,
 118             .type = QEMU_OPT_STRING,
 119             .help = "Logfile path of libgfapi",
 120         },
 121         { /* end of list */ }
 122     },
 123 };
 124
 125 static QemuOptsList runtime_json_opts = {
 126     .name = "gluster_json",
 127     .head = QTAILQ_HEAD_INITIALIZER(runtime_json_opts.head),
 128     .desc = {
 129         {
 130             .name = GLUSTER_OPT_VOLUME,
 131             .type = QEMU_OPT_STRING,
 132             .help = "name of gluster volume where VM image resides",
 133         },
 134         {
 135             .name = GLUSTER_OPT_PATH,
 136             .type = QEMU_OPT_STRING,
 137             .help = "absolute path to image file in gluster volume",
 138         },
 139         {
 140             .name = GLUSTER_OPT_DEBUG,
 141             .type = QEMU_OPT_NUMBER,
 142             .help = "Gluster log level, valid range is 0-9",
 143         },
 144         { /* end of list */ }
 145     },
 146 };
 147
 148 static QemuOptsList runtime_type_opts = {
 149     .name = "gluster_type",
 150     .head = QTAILQ_HEAD_INITIALIZER(runtime_type_opts.head),
 151     .desc = {
 152         {
 153             .name = GLUSTER_OPT_TYPE,
 154             .type = QEMU_OPT_STRING,
 155             .help = "inet|unix",
 156         },
 157         { /* end of list */ }
 158     },
 159 };
 160
 161 static QemuOptsList runtime_unix_opts = {
 162     .name = "gluster_unix",
 163     .head = QTAILQ_HEAD_INITIALIZER(runtime_unix_opts.head),
 164     .desc = {
 165         {
 166             .name = GLUSTER_OPT_SOCKET,
 167             .type = QEMU_OPT_STRING,
 168             .help = "socket file path)",
 169         },
 170         { /* end of list */ }
 171     },
 172 };
 173
 174 static QemuOptsList runtime_inet_opts = {
 175     .name = "gluster_inet",
 176     .head = QTAILQ_HEAD_INITIALIZER(runtime_inet_opts.head),
 177     .desc = {
 178         {
 179             .name = GLUSTER_OPT_TYPE,
 180             .type = QEMU_OPT_STRING,
 181             .help = "inet|unix",
 182         },
 183         {
 184             .name = GLUSTER_OPT_HOST,
 185             .type = QEMU_OPT_STRING,
 186             .help = "host address (hostname/ipv4/ipv6 addresses)",
 187         },
 188         {
 189             .name = GLUSTER_OPT_PORT,
 190             .type = QEMU_OPT_STRING,
 191             .help = "port number on which glusterd is listening (default 24007)",
 192         },
 193         {
 194             .name = "to",
 195             .type = QEMU_OPT_NUMBER,
 196             .help = "max port number, not supported by gluster",
 197         },
 198         {
 199             .name = "ipv4",
 200             .type = QEMU_OPT_BOOL,
 201             .help = "ipv4 bool value, not supported by gluster",
 202         },
 203         {
 204             .name = "ipv6",
 205             .type = QEMU_OPT_BOOL,
 206             .help = "ipv6 bool value, not supported by gluster",
 207         },
 208         { /* end of list */ }
 209     },
 210 };
 211
 212 static void glfs_set_preopened(const char *volume, glfs_t *fs)
 213 {
 214     ListElement *entry = NULL;
 215
 216     entry = g_new(ListElement, 1);
 217
 218     entry->saved.volume = g_strdup(volume);
 219
 220     entry->saved.fs = fs;
 221     entry->saved.ref = 1;
 222
 223     QLIST_INSERT_HEAD(&glfs_list, entry, list);
 224 }
 225
 226 static glfs_t *glfs_find_preopened(const char *volume)
 227 {
 228     ListElement *entry = NULL;
 229
 230      QLIST_FOREACH(entry, &glfs_list, list) {
 231         if (strcmp(entry->saved.volume, volume) == 0) {
 232             entry->saved.ref++;
 233             return entry->saved.fs;
 234         }
 235      }
 236
 237     return NULL;
 238 }
 239
 240 static void glfs_clear_preopened(glfs_t *fs)
 241 {
 242     ListElement *entry = NULL;
 243     ListElement *next;
 244
 245     if (fs == NULL) {
 246         return;
 247     }
 248
 249     QLIST_FOREACH_SAFE(entry, &glfs_list, list, next) {
 250         if (entry->saved.fs == fs) {
 251             if (--entry->saved.ref) {
 252                 return;
 253             }
 254
 255             QLIST_REMOVE(entry, list);
 256
 257             glfs_fini(entry->saved.fs);
 258             g_free(entry->saved.volume);
 259             g_free(entry);
 260         }
 261     }
 262 }
 263
 264 static int parse_volume_options(BlockdevOptionsGluster *gconf, char *path)
 265 {
 266     char *p, *q;
 267
 268     if (!path) {
 269         return -EINVAL;
 270     }
 271
 272     /* volume */
 273     p = q = path + strspn(path, "/");
 274     p += strcspn(p, "/");
 275     if (*p == '\0') {
 276         return -EINVAL;
 277     }
 278     gconf->volume = g_strndup(q, p - q);
 279
 280     /* path */
 281     p += strspn(p, "/");
 282     if (*p == '\0') {
 283         return -EINVAL;
 284     }
 285     gconf->path = g_strdup(p);
 286     return 0;
 287 }
 288
 289 /*
 290  * file=gluster[+transport]://[host[:port]]/volume/path[?socket=...]
 291  *
 292  * 'gluster' is the protocol.
 293  *
 294  * 'transport' specifies the transport type used to connect to gluster
 295  * management daemon (glusterd). Valid transport types are
 296  * tcp or unix. If a transport type isn't specified, then tcp type is assumed.
 297  *
 298  * 'host' specifies the host where the volume file specification for
 299  * the given volume resides. This can be either hostname or ipv4 address.
 300  * If transport type is 'unix', then 'host' field should not be specified.
 301  * The 'socket' field needs to be populated with the path to unix domain
 302  * socket.
 303  *
 304  * 'port' is the port number on which glusterd is listening. This is optional
 305  * and if not specified, QEMU will send 0 which will make gluster to use the
 306  * default port. If the transport type is unix, then 'port' should not be
 307  * specified.
 308  *
 309  * 'volume' is the name of the gluster volume which contains the VM image.
 310  *
 311  * 'path' is the path to the actual VM image that resides on gluster volume.
 312  *
 313  * Examples:
 314  *
 315  * file=gluster://1.2.3.4/testvol/a.img
 316  * file=gluster+tcp://1.2.3.4/testvol/a.img
 317  * file=gluster+tcp://1.2.3.4:24007/testvol/dir/a.img
 318  * file=gluster+tcp://host.domain.com:24007/testvol/dir/a.img
 319  * file=gluster+unix:///testvol/dir/a.img?socket=/tmp/glusterd.socket
 320  */
 321 static int qemu_gluster_parse_uri(BlockdevOptionsGluster *gconf,
 322                                   const char *filename)
 323 {
 324     SocketAddressFlat *gsconf;
 325     URI *uri;
 326     QueryParams *qp = NULL;
 327     bool is_unix = false;
 328     int ret = 0;
 329
 330     uri = uri_parse(filename);
 331     if (!uri) {
 332         return -EINVAL;
 333     }
 334
 335     gconf->server = g_new0(SocketAddressFlatList, 1);
 336     gconf->server->value = gsconf = g_new0(SocketAddressFlat, 1);
 337
 338     /* transport */
 339     if (!uri->scheme || !strcmp(uri->scheme, "gluster")) {
 340         gsconf->type = SOCKET_ADDRESS_FLAT_TYPE_INET;
 341     } else if (!strcmp(uri->scheme, "gluster+tcp")) {
 342         gsconf->type = SOCKET_ADDRESS_FLAT_TYPE_INET;
 343     } else if (!strcmp(uri->scheme, "gluster+unix")) {
 344         gsconf->type = SOCKET_ADDRESS_FLAT_TYPE_UNIX;
 345         is_unix = true;
 346     } else if (!strcmp(uri->scheme, "gluster+rdma")) {
 347         gsconf->type = SOCKET_ADDRESS_FLAT_TYPE_INET;
 348         error_report("Warning: rdma feature is not supported, falling "
 349                      "back to tcp");
 350     } else {
 351         ret = -EINVAL;
 352         goto out;
 353     }
 354
 355     ret = parse_volume_options(gconf, uri->path);
 356     if (ret < 0) {
 357         goto out;
 358     }
 359
 360     qp = query_params_parse(uri->query);
 361     if (qp->n > 1 || (is_unix && !qp->n) || (!is_unix && qp->n)) {
 362         ret = -EINVAL;
 363         goto out;
 364     }
 365
 366     if (is_unix) {
 367         if (uri->server || uri->port) {
 368             ret = -EINVAL;
 369             goto out;
 370         }
 371         if (strcmp(qp->p[0].name, "socket")) {
 372             ret = -EINVAL;
 373             goto out;
 374         }
 375         gsconf->u.q_unix.path = g_strdup(qp->p[0].value);
 376     } else {
 377         gsconf->u.inet.host = g_strdup(uri->server ? uri->server : "localhost");
 378         if (uri->port) {
 379             gsconf->u.inet.port = g_strdup_printf("%d", uri->port);
 380         } else {
 381             gsconf->u.inet.port = g_strdup_printf("%d", GLUSTER_DEFAULT_PORT);
 382         }
 383     }
 384
 385 out:
 386     if (qp) {
 387         query_params_free(qp);
 388     }
 389     uri_free(uri);
 390     return ret;
 391 }
 392
 393 static struct glfs *qemu_gluster_glfs_init(BlockdevOptionsGluster *gconf,
 394                                            Error **errp)
 395 {
 396     struct glfs *glfs;
 397     int ret;
 398     int old_errno;
 399     SocketAddressFlatList *server;
 400     unsigned long long port;
 401
 402     glfs = glfs_find_preopened(gconf->volume);
 403     if (glfs) {
 404         return glfs;
 405     }
 406
 407     glfs = glfs_new(gconf->volume);
 408     if (!glfs) {
 409         goto out;
 410     }
 411
 412     glfs_set_preopened(gconf->volume, glfs);
 413
 414     for (server = gconf->server; server; server = server->next) {
 415         switch (server->value->type) {
 416         case SOCKET_ADDRESS_FLAT_TYPE_UNIX:
 417             ret = glfs_set_volfile_server(glfs, "unix",
 418                                    server->value->u.q_unix.path, 0);
 419             break;
 420         case SOCKET_ADDRESS_FLAT_TYPE_INET:
 421             if (parse_uint_full(server->value->u.inet.port, &port, 10) < 0 ||
 422                 port > 65535) {
 423                 error_setg(errp, "'%s' is not a valid port number",
 424                            server->value->u.inet.port);
 425                 errno = EINVAL;
 426                 goto out;
 427             }
 428             ret = glfs_set_volfile_server(glfs, "tcp",
 429                                    server->value->u.inet.host,
 430                                    (int)port);
 431             break;
 432         case SOCKET_ADDRESS_FLAT_TYPE_VSOCK:
 433         case SOCKET_ADDRESS_FLAT_TYPE_FD:
 434         default:
 435             abort();
 436         }
 437
 438         if (ret < 0) {
 439             goto out;
 440         }
 441     }
 442
 443     ret = glfs_set_logging(glfs, gconf->logfile, gconf->debug);
 444     if (ret < 0) {
 445         goto out;
 446     }
 447
 448     ret = glfs_init(glfs);
 449     if (ret) {
 450         error_setg(errp, "Gluster connection for volume %s, path %s failed"
 451                          " to connect", gconf->volume, gconf->path);
 452         for (server = gconf->server; server; server = server->next) {
 453             if (server->value->type  == SOCKET_ADDRESS_FLAT_TYPE_UNIX) {
 454                 error_append_hint(errp, "hint: failed on socket %s ",
 455                                   server->value->u.q_unix.path);
 456             } else {
 457                 error_append_hint(errp, "hint: failed on host %s and port %s ",
 458                                   server->value->u.inet.host,
 459                                   server->value->u.inet.port);
 460             }
 461         }
 462
 463         error_append_hint(errp, "Please refer to gluster logs for more info\n");
 464
 465         /* glfs_init sometimes doesn't set errno although docs suggest that */
 466         if (errno == 0) {
 467             errno = EINVAL;
 468         }
 469
 470         goto out;
 471     }
 472     return glfs;
 473
 474 out:
 475     if (glfs) {
 476         old_errno = errno;
 477         glfs_clear_preopened(glfs);
 478         errno = old_errno;
 479     }
 480     return NULL;
 481 }
 482
 483 /*
 484  * Convert the json formatted command line into qapi.
 485 */
 486 static int qemu_gluster_parse_json(BlockdevOptionsGluster *gconf,
 487                                   QDict *options, Error **errp)
 488 {
 489     QemuOpts *opts;
 490     SocketAddressFlat *gsconf = NULL;
 491     SocketAddressFlatList *curr = NULL;
 492     QDict *backing_options = NULL;
 493     Error *local_err = NULL;
 494     char *str = NULL;
 495     const char *ptr;
 496     size_t num_servers;
 497     int i, type;
 498
 499     /* create opts info from runtime_json_opts list */
 500     opts = qemu_opts_create(&runtime_json_opts, NULL, 0, &error_abort);
 501     qemu_opts_absorb_qdict(opts, options, &local_err);
 502     if (local_err) {
 503         goto out;
 504     }
 505
 506     num_servers = qdict_array_entries(options, GLUSTER_OPT_SERVER_PATTERN);
 507     if (num_servers < 1) {
 508         error_setg(&local_err, QERR_MISSING_PARAMETER, "server");
 509         goto out;
 510     }
 511
 512     ptr = qemu_opt_get(opts, GLUSTER_OPT_VOLUME);
 513     if (!ptr) {
 514         error_setg(&local_err, QERR_MISSING_PARAMETER, GLUSTER_OPT_VOLUME);
 515         goto out;
 516     }
 517     gconf->volume = g_strdup(ptr);
 518
 519     ptr = qemu_opt_get(opts, GLUSTER_OPT_PATH);
 520     if (!ptr) {
 521         error_setg(&local_err, QERR_MISSING_PARAMETER, GLUSTER_OPT_PATH);
 522         goto out;
 523     }
 524     gconf->path = g_strdup(ptr);
 525     qemu_opts_del(opts);
 526
 527     for (i = 0; i < num_servers; i++) {
 528         str = g_strdup_printf(GLUSTER_OPT_SERVER_PATTERN"%d.", i);
 529         qdict_extract_subqdict(options, &backing_options, str);
 530
 531         /* create opts info from runtime_type_opts list */
 532         opts = qemu_opts_create(&runtime_type_opts, NULL, 0, &error_abort);
 533         qemu_opts_absorb_qdict(opts, backing_options, &local_err);
 534         if (local_err) {
 535             goto out;
 536         }
 537
 538         ptr = qemu_opt_get(opts, GLUSTER_OPT_TYPE);
 539         if (!ptr) {
 540             error_setg(&local_err, QERR_MISSING_PARAMETER, GLUSTER_OPT_TYPE);
 541             error_append_hint(&local_err, GERR_INDEX_HINT, i);
 542             goto out;
 543
 544         }
 545         gsconf = g_new0(SocketAddressFlat, 1);
 546         if (!strcmp(ptr, "tcp")) {
 547             ptr = "inet";       /* accept legacy "tcp" */
 548         }
 549         type = qapi_enum_parse(SocketAddressFlatType_lookup, ptr,
 550                                SOCKET_ADDRESS_FLAT_TYPE__MAX, -1, NULL);
 551         if (type != SOCKET_ADDRESS_FLAT_TYPE_INET
 552             && type != SOCKET_ADDRESS_FLAT_TYPE_UNIX) {
 553             error_setg(&local_err,
 554                        "Parameter '%s' may be 'inet' or 'unix'",
 555                        GLUSTER_OPT_TYPE);
 556             error_append_hint(&local_err, GERR_INDEX_HINT, i);
 557             goto out;
 558         }
 559         gsconf->type = type;
 560         qemu_opts_del(opts);
 561
 562         if (gsconf->type == SOCKET_ADDRESS_FLAT_TYPE_INET) {
 563             /* create opts info from runtime_inet_opts list */
 564             opts = qemu_opts_create(&runtime_inet_opts, NULL, 0, &error_abort);
 565             qemu_opts_absorb_qdict(opts, backing_options, &local_err);
 566             if (local_err) {
 567                 goto out;
 568             }
 569
 570             ptr = qemu_opt_get(opts, GLUSTER_OPT_HOST);
 571             if (!ptr) {
 572                 error_setg(&local_err, QERR_MISSING_PARAMETER,
 573                            GLUSTER_OPT_HOST);
 574                 error_append_hint(&local_err, GERR_INDEX_HINT, i);
 575                 goto out;
 576             }
 577             gsconf->u.inet.host = g_strdup(ptr);
 578             ptr = qemu_opt_get(opts, GLUSTER_OPT_PORT);
 579             if (!ptr) {
 580                 error_setg(&local_err, QERR_MISSING_PARAMETER,
 581                            GLUSTER_OPT_PORT);
 582                 error_append_hint(&local_err, GERR_INDEX_HINT, i);
 583                 goto out;
 584             }
 585             gsconf->u.inet.port = g_strdup(ptr);
 586
 587             /* defend for unsupported fields in InetSocketAddress,
 588              * i.e. @ipv4, @ipv6  and @to
 589              */
 590             ptr = qemu_opt_get(opts, GLUSTER_OPT_TO);
 591             if (ptr) {
 592                 gsconf->u.inet.has_to = true;
 593             }
 594             ptr = qemu_opt_get(opts, GLUSTER_OPT_IPV4);
 595             if (ptr) {
 596                 gsconf->u.inet.has_ipv4 = true;
 597             }
 598             ptr = qemu_opt_get(opts, GLUSTER_OPT_IPV6);
 599             if (ptr) {
 600                 gsconf->u.inet.has_ipv6 = true;
 601             }
 602             if (gsconf->u.inet.has_to) {
 603                 error_setg(&local_err, "Parameter 'to' not supported");
 604                 goto out;
 605             }
 606             if (gsconf->u.inet.has_ipv4 || gsconf->u.inet.has_ipv6) {
 607                 error_setg(&local_err, "Parameters 'ipv4/ipv6' not supported");
 608                 goto out;
 609             }
 610             qemu_opts_del(opts);
 611         } else {
 612             /* create opts info from runtime_unix_opts list */
 613             opts = qemu_opts_create(&runtime_unix_opts, NULL, 0, &error_abort);
 614             qemu_opts_absorb_qdict(opts, backing_options, &local_err);
 615             if (local_err) {
 616                 goto out;
 617             }
 618
 619             ptr = qemu_opt_get(opts, GLUSTER_OPT_SOCKET);
 620             if (!ptr) {
 621                 error_setg(&local_err, QERR_MISSING_PARAMETER,
 622                            GLUSTER_OPT_SOCKET);
 623                 error_append_hint(&local_err, GERR_INDEX_HINT, i);
 624                 goto out;
 625             }
 626             gsconf->u.q_unix.path = g_strdup(ptr);
 627             qemu_opts_del(opts);
 628         }
 629
 630         if (gconf->server == NULL) {
 631             gconf->server = g_new0(SocketAddressFlatList, 1);
 632             gconf->server->value = gsconf;
 633             curr = gconf->server;
 634         } else {
 635             curr->next = g_new0(SocketAddressFlatList, 1);
 636             curr->next->value = gsconf;
 637             curr = curr->next;
 638         }
 639         gsconf = NULL;
 640
 641         QDECREF(backing_options);
 642         backing_options = NULL;
 643         g_free(str);
 644         str = NULL;
 645     }
 646
 647     return 0;
 648
 649 out:
 650     error_propagate(errp, local_err);
 651     qapi_free_SocketAddressFlat(gsconf);
 652     qemu_opts_del(opts);
 653     g_free(str);
 654     QDECREF(backing_options);
 655     errno = EINVAL;
 656     return -errno;
 657 }
 658
 659 static struct glfs *qemu_gluster_init(BlockdevOptionsGluster *gconf,
 660                                       const char *filename,
 661                                       QDict *options, Error **errp)
 662 {
 663     int ret;
 664     if (filename) {
 665         ret = qemu_gluster_parse_uri(gconf, filename);
 666         if (ret < 0) {
 667             error_setg(errp, "invalid URI");
 668             error_append_hint(errp, "Usage: file=gluster[+transport]://"
 669                                     "[host[:port]]volume/path[?socket=...]"
 670                                     "[,file.debug=N]"
 671                                     "[,file.logfile=/path/filename.log]\n");
 672             errno = -ret;
 673             return NULL;
 674         }
 675     } else {
 676         ret = qemu_gluster_parse_json(gconf, options, errp);
 677         if (ret < 0) {
 678             error_append_hint(errp, "Usage: "
 679                              "-drive driver=qcow2,file.driver=gluster,"
 680                              "file.volume=testvol,file.path=/path/a.qcow2"
 681                              "[,file.debug=9]"
 682                              "[,file.logfile=/path/filename.log],"
 683                              "file.server.0.type=inet,"
 684                              "file.server.0.host=1.2.3.4,"
 685                              "file.server.0.port=24007,"
 686                              "file.server.1.transport=unix,"
 687                              "file.server.1.socket=/var/run/glusterd.socket ..."
 688                              "\n");
 689             errno = -ret;
 690             return NULL;
 691         }
 692
 693     }
 694
 695     return qemu_gluster_glfs_init(gconf, errp);
 696 }
 697
 698 /*
 699  * AIO callback routine called from GlusterFS thread.
 700  */
 701 static void gluster_finish_aiocb(struct glfs_fd *fd, ssize_t ret, void *arg)
 702 {
 703     GlusterAIOCB *acb = (GlusterAIOCB *)arg;
 704
 705     if (!ret || ret == acb->size) {
 706         acb->ret = 0; /* Success */
 707     } else if (ret < 0) {
 708         acb->ret = -errno; /* Read/Write failed */
 709     } else {
 710         acb->ret = -EIO; /* Partial read/write - fail it */
 711     }
 712
 713     aio_co_schedule(acb->aio_context, acb->coroutine);
 714 }
 715
 716 static void qemu_gluster_parse_flags(int bdrv_flags, int *open_flags)
 717 {
 718     assert(open_flags != NULL);
 719
 720     *open_flags |= O_BINARY;
 721
 722     if (bdrv_flags & BDRV_O_RDWR) {
 723         *open_flags |= O_RDWR;
 724     } else {
 725         *open_flags |= O_RDONLY;
 726     }
 727
 728     if ((bdrv_flags & BDRV_O_NOCACHE)) {
 729         *open_flags |= O_DIRECT;
 730     }
 731 }
 732
 733 /*
 734  * Do SEEK_DATA/HOLE to detect if it is functional. Older broken versions of
 735  * gfapi incorrectly return the current offset when SEEK_DATA/HOLE is used.
 736  * - Corrected versions return -1 and set errno to EINVAL.
 737  * - Versions that support SEEK_DATA/HOLE correctly, will return -1 and set
 738  *   errno to ENXIO when SEEK_DATA is called with a position of EOF.
 739  */
 740 static bool qemu_gluster_test_seek(struct glfs_fd *fd)
 741 {
 742     off_t ret = 0;
 743
 744 #if defined SEEK_HOLE && defined SEEK_DATA
 745     off_t eof;
 746
 747     eof = glfs_lseek(fd, 0, SEEK_END);
 748     if (eof < 0) {
 749         /* this should never occur */
 750         return false;
 751     }
 752
 753     /* this should always fail with ENXIO if SEEK_DATA is supported */
 754     ret = glfs_lseek(fd, eof, SEEK_DATA);
 755 #endif
 756
 757     return (ret < 0) && (errno == ENXIO);
 758 }
 759
 760 static int qemu_gluster_open(BlockDriverState *bs,  QDict *options,
 761                              int bdrv_flags, Error **errp)
 762 {
 763     BDRVGlusterState *s = bs->opaque;
 764     int open_flags = 0;
 765     int ret = 0;
 766     BlockdevOptionsGluster *gconf = NULL;
 767     QemuOpts *opts;
 768     Error *local_err = NULL;
 769     const char *filename, *logfile;
 770
 771     opts = qemu_opts_create(&runtime_opts, NULL, 0, &error_abort);
 772     qemu_opts_absorb_qdict(opts, options, &local_err);
 773     if (local_err) {
 774         error_propagate(errp, local_err);
 775         ret = -EINVAL;
 776         goto out;
 777     }
 778
 779     filename = qemu_opt_get(opts, GLUSTER_OPT_FILENAME);
 780
 781     s->debug = qemu_opt_get_number(opts, GLUSTER_OPT_DEBUG,
 782                                    GLUSTER_DEBUG_DEFAULT);
 783     if (s->debug < 0) {
 784         s->debug = 0;
 785     } else if (s->debug > GLUSTER_DEBUG_MAX) {
 786         s->debug = GLUSTER_DEBUG_MAX;
 787     }
 788
 789     gconf = g_new0(BlockdevOptionsGluster, 1);
 790     gconf->debug = s->debug;
 791     gconf->has_debug = true;
 792
 793     logfile = qemu_opt_get(opts, GLUSTER_OPT_LOGFILE);
 794     s->logfile = g_strdup(logfile ? logfile : GLUSTER_LOGFILE_DEFAULT);
 795
 796     gconf->logfile = g_strdup(s->logfile);
 797     gconf->has_logfile = true;
 798
 799     s->glfs = qemu_gluster_init(gconf, filename, options, errp);
 800     if (!s->glfs) {
 801         ret = -errno;
 802         goto out;
 803     }
 804
 805 #ifdef CONFIG_GLUSTERFS_XLATOR_OPT
 806     /* Without this, if fsync fails for a recoverable reason (for instance,
 807      * ENOSPC), gluster will dump its cache, preventing retries.  This means
 808      * almost certain data loss.  Not all gluster versions support the
 809      * 'resync-failed-syncs-after-fsync' key value, but there is no way to
 810      * discover during runtime if it is supported (this api returns success for
 811      * unknown key/value pairs) */
 812     ret = glfs_set_xlator_option(s->glfs, "*-write-behind",
 813                                           "resync-failed-syncs-after-fsync",
 814                                           "on");
 815     if (ret < 0) {
 816         error_setg_errno(errp, errno, "Unable to set xlator key/value pair");
 817         ret = -errno;
 818         goto out;
 819     }
 820 #endif
 821
 822     qemu_gluster_parse_flags(bdrv_flags, &open_flags);
 823
 824     s->fd = glfs_open(s->glfs, gconf->path, open_flags);
 825     if (!s->fd) {
 826         ret = -errno;
 827     }
 828
 829     s->supports_seek_data = qemu_gluster_test_seek(s->fd);
 830
 831 out:
 832     qemu_opts_del(opts);
 833     qapi_free_BlockdevOptionsGluster(gconf);
 834     if (!ret) {
 835         return ret;
 836     }
 837     g_free(s->logfile);
 838     if (s->fd) {
 839         glfs_close(s->fd);
 840     }
 841
 842     glfs_clear_preopened(s->glfs);
 843
 844     return ret;
 845 }
 846
 847 static int qemu_gluster_reopen_prepare(BDRVReopenState *state,
 848                                        BlockReopenQueue *queue, Error **errp)
 849 {
 850     int ret = 0;
 851     BDRVGlusterState *s;
 852     BDRVGlusterReopenState *reop_s;
 853     BlockdevOptionsGluster *gconf;
 854     int open_flags = 0;
 855
 856     assert(state != NULL);
 857     assert(state->bs != NULL);
 858
 859     s = state->bs->opaque;
 860
 861     state->opaque = g_new0(BDRVGlusterReopenState, 1);
 862     reop_s = state->opaque;
 863
 864     qemu_gluster_parse_flags(state->flags, &open_flags);
 865
 866     gconf = g_new0(BlockdevOptionsGluster, 1);
 867     gconf->debug = s->debug;
 868     gconf->has_debug = true;
 869     gconf->logfile = g_strdup(s->logfile);
 870     gconf->has_logfile = true;
 871     reop_s->glfs = qemu_gluster_init(gconf, state->bs->filename, NULL, errp);
 872     if (reop_s->glfs == NULL) {
 873         ret = -errno;
 874         goto exit;
 875     }
 876
 877 #ifdef CONFIG_GLUSTERFS_XLATOR_OPT
 878     ret = glfs_set_xlator_option(reop_s->glfs, "*-write-behind",
 879                                  "resync-failed-syncs-after-fsync", "on");
 880     if (ret < 0) {
 881         error_setg_errno(errp, errno, "Unable to set xlator key/value pair");
 882         ret = -errno;
 883         goto exit;
 884     }
 885 #endif
 886
 887     reop_s->fd = glfs_open(reop_s->glfs, gconf->path, open_flags);
 888     if (reop_s->fd == NULL) {
 889         /* reops->glfs will be cleaned up in _abort */
 890         ret = -errno;
 891         goto exit;
 892     }
 893
 894 exit:
 895     /* state->opaque will be freed in either the _abort or _commit */
 896     qapi_free_BlockdevOptionsGluster(gconf);
 897     return ret;
 898 }
 899
 900 static void qemu_gluster_reopen_commit(BDRVReopenState *state)
 901 {
 902     BDRVGlusterReopenState *reop_s = state->opaque;
 903     BDRVGlusterState *s = state->bs->opaque;
 904
 905
 906     /* close the old */
 907     if (s->fd) {
 908         glfs_close(s->fd);
 909     }
 910
 911     glfs_clear_preopened(s->glfs);
 912
 913     /* use the newly opened image / connection */
 914     s->fd         = reop_s->fd;
 915     s->glfs       = reop_s->glfs;
 916
 917     g_free(state->opaque);
 918     state->opaque = NULL;
 919
 920     return;
 921 }
 922
 923
 924 static void qemu_gluster_reopen_abort(BDRVReopenState *state)
 925 {
 926     BDRVGlusterReopenState *reop_s = state->opaque;
 927
 928     if (reop_s == NULL) {
 929         return;
 930     }
 931
 932     if (reop_s->fd) {
 933         glfs_close(reop_s->fd);
 934     }
 935
 936     glfs_clear_preopened(reop_s->glfs);
 937
 938     g_free(state->opaque);
 939     state->opaque = NULL;
 940
 941     return;
 942 }
 943
 944 #ifdef CONFIG_GLUSTERFS_ZEROFILL
 945 static coroutine_fn int qemu_gluster_co_pwrite_zeroes(BlockDriverState *bs,
 946                                                       int64_t offset,
 947                                                       int size,
 948                                                       BdrvRequestFlags flags)
 949 {
 950     int ret;
 951     GlusterAIOCB acb;
 952     BDRVGlusterState *s = bs->opaque;
 953
 954     acb.size = size;
 955     acb.ret = 0;
 956     acb.coroutine = qemu_coroutine_self();
 957     acb.aio_context = bdrv_get_aio_context(bs);
 958
 959     ret = glfs_zerofill_async(s->fd, offset, size, gluster_finish_aiocb, &acb);
 960     if (ret < 0) {
 961         return -errno;
 962     }
 963
 964     qemu_coroutine_yield();
 965     return acb.ret;
 966 }
 967
 968 static inline bool gluster_supports_zerofill(void)
 969 {
 970     return 1;
 971 }
 972
 973 static inline int qemu_gluster_zerofill(struct glfs_fd *fd, int64_t offset,
 974                                         int64_t size)
 975 {
 976     return glfs_zerofill(fd, offset, size);
 977 }
 978
 979 #else
 980 static inline bool gluster_supports_zerofill(void)
 981 {
 982     return 0;
 983 }
 984
 985 static inline int qemu_gluster_zerofill(struct glfs_fd *fd, int64_t offset,
 986                                         int64_t size)
 987 {
 988     return 0;
 989 }
 990 #endif
 991
 992 static int qemu_gluster_create(const char *filename,
 993                                QemuOpts *opts, Error **errp)
 994 {
 995     BlockdevOptionsGluster *gconf;
 996     struct glfs *glfs;
 997     struct glfs_fd *fd;
 998     int ret = 0;
 999     int prealloc = 0;
1000     int64_t total_size = 0;
1001     char *tmp = NULL;
1002
1003     gconf = g_new0(BlockdevOptionsGluster, 1);
1004     gconf->debug = qemu_opt_get_number_del(opts, GLUSTER_OPT_DEBUG,
1005                                            GLUSTER_DEBUG_DEFAULT);
1006     if (gconf->debug < 0) {
1007         gconf->debug = 0;
1008     } else if (gconf->debug > GLUSTER_DEBUG_MAX) {
1009         gconf->debug = GLUSTER_DEBUG_MAX;
1010     }
1011     gconf->has_debug = true;
1012
1013     gconf->logfile = qemu_opt_get_del(opts, GLUSTER_OPT_LOGFILE);
1014     if (!gconf->logfile) {
1015         gconf->logfile = g_strdup(GLUSTER_LOGFILE_DEFAULT);
1016     }
1017     gconf->has_logfile = true;
1018
1019     glfs = qemu_gluster_init(gconf, filename, NULL, errp);
1020     if (!glfs) {
1021         ret = -errno;
1022         goto out;
1023     }
1024
1025     total_size = ROUND_UP(qemu_opt_get_size_del(opts, BLOCK_OPT_SIZE, 0),
1026                           BDRV_SECTOR_SIZE);
1027
1028     tmp = qemu_opt_get_del(opts, BLOCK_OPT_PREALLOC);
1029     if (!tmp || !strcmp(tmp, "off")) {
1030         prealloc = 0;
1031     } else if (!strcmp(tmp, "full") && gluster_supports_zerofill()) {
1032         prealloc = 1;
1033     } else {
1034         error_setg(errp, "Invalid preallocation mode: '%s'"
1035                          " or GlusterFS doesn't support zerofill API", tmp);
1036         ret = -EINVAL;
1037         goto out;
1038     }
1039
1040     fd = glfs_creat(glfs, gconf->path,
1041                     O_WRONLY | O_CREAT | O_TRUNC | O_BINARY, S_IRUSR | S_IWUSR);
1042     if (!fd) {
1043         ret = -errno;
1044     } else {
1045         if (!glfs_ftruncate(fd, total_size)) {
1046             if (prealloc && qemu_gluster_zerofill(fd, 0, total_size)) {
1047                 ret = -errno;
1048             }
1049         } else {
1050             ret = -errno;
1051         }
1052
1053         if (glfs_close(fd) != 0) {
1054             ret = -errno;
1055         }
1056     }
1057 out:
1058     g_free(tmp);
1059     qapi_free_BlockdevOptionsGluster(gconf);
1060     glfs_clear_preopened(glfs);
1061     return ret;
1062 }
1063
1064 static coroutine_fn int qemu_gluster_co_rw(BlockDriverState *bs,
1065                                            int64_t sector_num, int nb_sectors,
1066                                            QEMUIOVector *qiov, int write)
1067 {
1068     int ret;
1069     GlusterAIOCB acb;
1070     BDRVGlusterState *s = bs->opaque;
1071     size_t size = nb_sectors * BDRV_SECTOR_SIZE;
1072     off_t offset = sector_num * BDRV_SECTOR_SIZE;
1073
1074     acb.size = size;
1075     acb.ret = 0;
1076     acb.coroutine = qemu_coroutine_self();
1077     acb.aio_context = bdrv_get_aio_context(bs);
1078
1079     if (write) {
1080         ret = glfs_pwritev_async(s->fd, qiov->iov, qiov->niov, offset, 0,
1081                                  gluster_finish_aiocb, &acb);
1082     } else {
1083         ret = glfs_preadv_async(s->fd, qiov->iov, qiov->niov, offset, 0,
1084                                 gluster_finish_aiocb, &acb);
1085     }
1086
1087     if (ret < 0) {
1088         return -errno;
1089     }
1090
1091     qemu_coroutine_yield();
1092     return acb.ret;
1093 }
1094
1095 static int qemu_gluster_truncate(BlockDriverState *bs, int64_t offset)
1096 {
1097     int ret;
1098     BDRVGlusterState *s = bs->opaque;
1099
1100     ret = glfs_ftruncate(s->fd, offset);
1101     if (ret < 0) {
1102         return -errno;
1103     }
1104
1105     return 0;
1106 }
1107
1108 static coroutine_fn int qemu_gluster_co_readv(BlockDriverState *bs,
1109                                               int64_t sector_num,
1110                                               int nb_sectors,
1111                                               QEMUIOVector *qiov)
1112 {
1113     return qemu_gluster_co_rw(bs, sector_num, nb_sectors, qiov, 0);
1114 }
1115
1116 static coroutine_fn int qemu_gluster_co_writev(BlockDriverState *bs,
1117                                                int64_t sector_num,
1118                                                int nb_sectors,
1119                                                QEMUIOVector *qiov)
1120 {
1121     return qemu_gluster_co_rw(bs, sector_num, nb_sectors, qiov, 1);
1122 }
1123
1124 static void qemu_gluster_close(BlockDriverState *bs)
1125 {
1126     BDRVGlusterState *s = bs->opaque;
1127
1128     g_free(s->logfile);
1129     if (s->fd) {
1130         glfs_close(s->fd);
1131         s->fd = NULL;
1132     }
1133     glfs_clear_preopened(s->glfs);
1134 }
1135
1136 static coroutine_fn int qemu_gluster_co_flush_to_disk(BlockDriverState *bs)
1137 {
1138     int ret;
1139     GlusterAIOCB acb;
1140     BDRVGlusterState *s = bs->opaque;
1141
1142     acb.size = 0;
1143     acb.ret = 0;
1144     acb.coroutine = qemu_coroutine_self();
1145     acb.aio_context = bdrv_get_aio_context(bs);
1146
1147     ret = glfs_fsync_async(s->fd, gluster_finish_aiocb, &acb);
1148     if (ret < 0) {
1149         ret = -errno;
1150         goto error;
1151     }
1152
1153     qemu_coroutine_yield();
1154     if (acb.ret < 0) {
1155         ret = acb.ret;
1156         goto error;
1157     }
1158
1159     return acb.ret;
1160
1161 error:
1162     /* Some versions of Gluster (3.5.6 -> 3.5.8?) will not retain its cache
1163      * after a fsync failure, so we have no way of allowing the guest to safely
1164      * continue.  Gluster versions prior to 3.5.6 don't retain the cache
1165      * either, but will invalidate the fd on error, so this is again our only
1166      * option.
1167      *
1168      * The 'resync-failed-syncs-after-fsync' xlator option for the
1169      * write-behind cache will cause later gluster versions to retain its
1170      * cache after error, so long as the fd remains open.  However, we
1171      * currently have no way of knowing if this option is supported.
1172      *
1173      * TODO: Once gluster provides a way for us to determine if the option
1174      * is supported, bypass the closure and setting drv to NULL.  */
1175     qemu_gluster_close(bs);
1176     bs->drv = NULL;
1177     return ret;
1178 }
1179
1180 #ifdef CONFIG_GLUSTERFS_DISCARD
1181 static coroutine_fn int qemu_gluster_co_pdiscard(BlockDriverState *bs,
1182                                                  int64_t offset, int size)
1183 {
1184     int ret;
1185     GlusterAIOCB acb;
1186     BDRVGlusterState *s = bs->opaque;
1187
1188     acb.size = 0;
1189     acb.ret = 0;
1190     acb.coroutine = qemu_coroutine_self();
1191     acb.aio_context = bdrv_get_aio_context(bs);
1192
1193     ret = glfs_discard_async(s->fd, offset, size, gluster_finish_aiocb, &acb);
1194     if (ret < 0) {
1195         return -errno;
1196     }
1197
1198     qemu_coroutine_yield();
1199     return acb.ret;
1200 }
1201 #endif
1202
1203 static int64_t qemu_gluster_getlength(BlockDriverState *bs)
1204 {
1205     BDRVGlusterState *s = bs->opaque;
1206     int64_t ret;
1207
1208     ret = glfs_lseek(s->fd, 0, SEEK_END);
1209     if (ret < 0) {
1210         return -errno;
1211     } else {
1212         return ret;
1213     }
1214 }
1215
1216 static int64_t qemu_gluster_allocated_file_size(BlockDriverState *bs)
1217 {
1218     BDRVGlusterState *s = bs->opaque;
1219     struct stat st;
1220     int ret;
1221
1222     ret = glfs_fstat(s->fd, &st);
1223     if (ret < 0) {
1224         return -errno;
1225     } else {
1226         return st.st_blocks * 512;
1227     }
1228 }
1229
1230 static int qemu_gluster_has_zero_init(BlockDriverState *bs)
1231 {
1232     /* GlusterFS volume could be backed by a block device */
1233     return 0;
1234 }
1235
1236 /*
1237  * Find allocation range in @bs around offset @start.
1238  * May change underlying file descriptor's file offset.
1239  * If @start is not in a hole, store @start in @data, and the
1240  * beginning of the next hole in @hole, and return 0.
1241  * If @start is in a non-trailing hole, store @start in @hole and the
1242  * beginning of the next non-hole in @data, and return 0.
1243  * If @start is in a trailing hole or beyond EOF, return -ENXIO.
1244  * If we can't find out, return a negative errno other than -ENXIO.
1245  *
1246  * (Shamefully copied from file-posix.c, only miniscule adaptions.)
1247  */
1248 static int find_allocation(BlockDriverState *bs, off_t start,
1249                            off_t *data, off_t *hole)
1250 {
1251     BDRVGlusterState *s = bs->opaque;
1252
1253     if (!s->supports_seek_data) {
1254         goto exit;
1255     }
1256
1257 #if defined SEEK_HOLE && defined SEEK_DATA
1258     off_t offs;
1259
1260     /*
1261      * SEEK_DATA cases:
1262      * D1. offs == start: start is in data
1263      * D2. offs > start: start is in a hole, next data at offs
1264      * D3. offs < 0, errno = ENXIO: either start is in a trailing hole
1265      *                              or start is beyond EOF
1266      *     If the latter happens, the file has been truncated behind
1267      *     our back since we opened it.  All bets are off then.
1268      *     Treating like a trailing hole is simplest.
1269      * D4. offs < 0, errno != ENXIO: we learned nothing
1270      */
1271     offs = glfs_lseek(s->fd, start, SEEK_DATA);
1272     if (offs < 0) {
1273         return -errno;          /* D3 or D4 */
1274     }
1275     assert(offs >= start);
1276
1277     if (offs > start) {
1278         /* D2: in hole, next data at offs */
1279         *hole = start;
1280         *data = offs;
1281         return 0;
1282     }
1283
1284     /* D1: in data, end not yet known */
1285
1286     /*
1287      * SEEK_HOLE cases:
1288      * H1. offs == start: start is in a hole
1289      *     If this happens here, a hole has been dug behind our back
1290      *     since the previous lseek().
1291      * H2. offs > start: either start is in data, next hole at offs,
1292      *                   or start is in trailing hole, EOF at offs
1293      *     Linux treats trailing holes like any other hole: offs ==
1294      *     start.  Solaris seeks to EOF instead: offs > start (blech).
1295      *     If that happens here, a hole has been dug behind our back
1296      *     since the previous lseek().
1297      * H3. offs < 0, errno = ENXIO: start is beyond EOF
1298      *     If this happens, the file has been truncated behind our
1299      *     back since we opened it.  Treat it like a trailing hole.
1300      * H4. offs < 0, errno != ENXIO: we learned nothing
1301      *     Pretend we know nothing at all, i.e. "forget" about D1.
1302      */
1303     offs = glfs_lseek(s->fd, start, SEEK_HOLE);
1304     if (offs < 0) {
1305         return -errno;          /* D1 and (H3 or H4) */
1306     }
1307     assert(offs >= start);
1308
1309     if (offs > start) {
1310         /*
1311          * D1 and H2: either in data, next hole at offs, or it was in
1312          * data but is now in a trailing hole.  In the latter case,
1313          * all bets are off.  Treating it as if it there was data all
1314          * the way to EOF is safe, so simply do that.
1315          */
1316         *data = start;
1317         *hole = offs;
1318         return 0;
1319     }
1320
1321     /* D1 and H1 */
1322     return -EBUSY;
1323 #endif
1324
1325 exit:
1326     return -ENOTSUP;
1327 }
1328
1329 /*
1330  * Returns the allocation status of the specified sectors.
1331  *
1332  * If 'sector_num' is beyond the end of the disk image the return value is 0
1333  * and 'pnum' is set to 0.
1334  *
1335  * 'pnum' is set to the number of sectors (including and immediately following
1336  * the specified sector) that are known to be in the same
1337  * allocated/unallocated state.
1338  *
1339  * 'nb_sectors' is the max value 'pnum' should be set to.  If nb_sectors goes
1340  * beyond the end of the disk image it will be clamped.
1341  *
1342  * (Based on raw_co_get_block_status() from file-posix.c.)
1343  */
1344 static int64_t coroutine_fn qemu_gluster_co_get_block_status(
1345         BlockDriverState *bs, int64_t sector_num, int nb_sectors, int *pnum,
1346         BlockDriverState **file)
1347 {
1348     BDRVGlusterState *s = bs->opaque;
1349     off_t start, data = 0, hole = 0;
1350     int64_t total_size;
1351     int ret = -EINVAL;
1352
1353     if (!s->fd) {
1354         return ret;
1355     }
1356
1357     start = sector_num * BDRV_SECTOR_SIZE;
1358     total_size = bdrv_getlength(bs);
1359     if (total_size < 0) {
1360         return total_size;
1361     } else if (start >= total_size) {
1362         *pnum = 0;
1363         return 0;
1364     } else if (start + nb_sectors * BDRV_SECTOR_SIZE > total_size) {
1365         nb_sectors = DIV_ROUND_UP(total_size - start, BDRV_SECTOR_SIZE);
1366     }
1367
1368     ret = find_allocation(bs, start, &data, &hole);
1369     if (ret == -ENXIO) {
1370         /* Trailing hole */
1371         *pnum = nb_sectors;
1372         ret = BDRV_BLOCK_ZERO;
1373     } else if (ret < 0) {
1374         /* No info available, so pretend there are no holes */
1375         *pnum = nb_sectors;
1376         ret = BDRV_BLOCK_DATA;
1377     } else if (data == start) {
1378         /* On a data extent, compute sectors to the end of the extent,
1379          * possibly including a partial sector at EOF. */
1380         *pnum = MIN(nb_sectors, DIV_ROUND_UP(hole - start, BDRV_SECTOR_SIZE));
1381         ret = BDRV_BLOCK_DATA;
1382     } else {
1383         /* On a hole, compute sectors to the beginning of the next extent.  */
1384         assert(hole == start);
1385         *pnum = MIN(nb_sectors, (data - start) / BDRV_SECTOR_SIZE);
1386         ret = BDRV_BLOCK_ZERO;
1387     }
1388
1389     *file = bs;
1390
1391     return ret | BDRV_BLOCK_OFFSET_VALID | start;
1392 }
1393
1394
1395 static BlockDriver bdrv_gluster = {
1396     .format_name                  = "gluster",
1397     .protocol_name                = "gluster",
1398     .instance_size                = sizeof(BDRVGlusterState),
1399     .bdrv_needs_filename          = false,
1400     .bdrv_file_open               = qemu_gluster_open,
1401     .bdrv_reopen_prepare          = qemu_gluster_reopen_prepare,
1402     .bdrv_reopen_commit           = qemu_gluster_reopen_commit,
1403     .bdrv_reopen_abort            = qemu_gluster_reopen_abort,
1404     .bdrv_close                   = qemu_gluster_close,
1405     .bdrv_create                  = qemu_gluster_create,
1406     .bdrv_getlength               = qemu_gluster_getlength,
1407     .bdrv_get_allocated_file_size = qemu_gluster_allocated_file_size,
1408     .bdrv_truncate                = qemu_gluster_truncate,
1409     .bdrv_co_readv                = qemu_gluster_co_readv,
1410     .bdrv_co_writev               = qemu_gluster_co_writev,
1411     .bdrv_co_flush_to_disk        = qemu_gluster_co_flush_to_disk,
1412     .bdrv_has_zero_init           = qemu_gluster_has_zero_init,
1413 #ifdef CONFIG_GLUSTERFS_DISCARD
1414     .bdrv_co_pdiscard             = qemu_gluster_co_pdiscard,
1415 #endif
1416 #ifdef CONFIG_GLUSTERFS_ZEROFILL
1417     .bdrv_co_pwrite_zeroes        = qemu_gluster_co_pwrite_zeroes,
1418 #endif
1419     .bdrv_co_get_block_status     = qemu_gluster_co_get_block_status,
1420     .create_opts                  = &qemu_gluster_create_opts,
1421 };
1422
1423 static BlockDriver bdrv_gluster_tcp = {
1424     .format_name                  = "gluster",
1425     .protocol_name                = "gluster+tcp",
1426     .instance_size                = sizeof(BDRVGlusterState),
1427     .bdrv_needs_filename          = false,
1428     .bdrv_file_open               = qemu_gluster_open,
1429     .bdrv_reopen_prepare          = qemu_gluster_reopen_prepare,
1430     .bdrv_reopen_commit           = qemu_gluster_reopen_commit,
1431     .bdrv_reopen_abort            = qemu_gluster_reopen_abort,
1432     .bdrv_close                   = qemu_gluster_close,
1433     .bdrv_create                  = qemu_gluster_create,
1434     .bdrv_getlength               = qemu_gluster_getlength,
1435     .bdrv_get_allocated_file_size = qemu_gluster_allocated_file_size,
1436     .bdrv_truncate                = qemu_gluster_truncate,
1437     .bdrv_co_readv                = qemu_gluster_co_readv,
1438     .bdrv_co_writev               = qemu_gluster_co_writev,
1439     .bdrv_co_flush_to_disk        = qemu_gluster_co_flush_to_disk,
1440     .bdrv_has_zero_init           = qemu_gluster_has_zero_init,
1441 #ifdef CONFIG_GLUSTERFS_DISCARD
1442     .bdrv_co_pdiscard             = qemu_gluster_co_pdiscard,
1443 #endif
1444 #ifdef CONFIG_GLUSTERFS_ZEROFILL
1445     .bdrv_co_pwrite_zeroes        = qemu_gluster_co_pwrite_zeroes,
1446 #endif
1447     .bdrv_co_get_block_status     = qemu_gluster_co_get_block_status,
1448     .create_opts                  = &qemu_gluster_create_opts,
1449 };
1450
1451 static BlockDriver bdrv_gluster_unix = {
1452     .format_name                  = "gluster",
1453     .protocol_name                = "gluster+unix",
1454     .instance_size                = sizeof(BDRVGlusterState),
1455     .bdrv_needs_filename          = true,
1456     .bdrv_file_open               = qemu_gluster_open,
1457     .bdrv_reopen_prepare          = qemu_gluster_reopen_prepare,
1458     .bdrv_reopen_commit           = qemu_gluster_reopen_commit,
1459     .bdrv_reopen_abort            = qemu_gluster_reopen_abort,
1460     .bdrv_close                   = qemu_gluster_close,
1461     .bdrv_create                  = qemu_gluster_create,
1462     .bdrv_getlength               = qemu_gluster_getlength,
1463     .bdrv_get_allocated_file_size = qemu_gluster_allocated_file_size,
1464     .bdrv_truncate                = qemu_gluster_truncate,
1465     .bdrv_co_readv                = qemu_gluster_co_readv,
1466     .bdrv_co_writev               = qemu_gluster_co_writev,
1467     .bdrv_co_flush_to_disk        = qemu_gluster_co_flush_to_disk,
1468     .bdrv_has_zero_init           = qemu_gluster_has_zero_init,
1469 #ifdef CONFIG_GLUSTERFS_DISCARD
1470     .bdrv_co_pdiscard             = qemu_gluster_co_pdiscard,
1471 #endif
1472 #ifdef CONFIG_GLUSTERFS_ZEROFILL
1473     .bdrv_co_pwrite_zeroes        = qemu_gluster_co_pwrite_zeroes,
1474 #endif
1475     .bdrv_co_get_block_status     = qemu_gluster_co_get_block_status,
1476     .create_opts                  = &qemu_gluster_create_opts,
1477 };
1478
1479 /* rdma is deprecated (actually never supported for volfile fetch).
1480  * Let's maintain it for the protocol compatibility, to make sure things
1481  * won't break immediately. For now, gluster+rdma will fall back to gluster+tcp
1482  * protocol with a warning.
1483  * TODO: remove gluster+rdma interface support
1484  */
1485 static BlockDriver bdrv_gluster_rdma = {
1486     .format_name                  = "gluster",
1487     .protocol_name                = "gluster+rdma",
1488     .instance_size                = sizeof(BDRVGlusterState),
1489     .bdrv_needs_filename          = true,
1490     .bdrv_file_open               = qemu_gluster_open,
1491     .bdrv_reopen_prepare          = qemu_gluster_reopen_prepare,
1492     .bdrv_reopen_commit           = qemu_gluster_reopen_commit,
1493     .bdrv_reopen_abort            = qemu_gluster_reopen_abort,
1494     .bdrv_close                   = qemu_gluster_close,
1495     .bdrv_create                  = qemu_gluster_create,
1496     .bdrv_getlength               = qemu_gluster_getlength,
1497     .bdrv_get_allocated_file_size = qemu_gluster_allocated_file_size,
1498     .bdrv_truncate                = qemu_gluster_truncate,
1499     .bdrv_co_readv                = qemu_gluster_co_readv,
1500     .bdrv_co_writev               = qemu_gluster_co_writev,
1501     .bdrv_co_flush_to_disk        = qemu_gluster_co_flush_to_disk,
1502     .bdrv_has_zero_init           = qemu_gluster_has_zero_init,
1503 #ifdef CONFIG_GLUSTERFS_DISCARD
1504     .bdrv_co_pdiscard             = qemu_gluster_co_pdiscard,
1505 #endif
1506 #ifdef CONFIG_GLUSTERFS_ZEROFILL
1507     .bdrv_co_pwrite_zeroes        = qemu_gluster_co_pwrite_zeroes,
1508 #endif
1509     .bdrv_co_get_block_status     = qemu_gluster_co_get_block_status,
1510     .create_opts                  = &qemu_gluster_create_opts,
1511 };
1512
1513 static void bdrv_gluster_init(void)
1514 {
1515     bdrv_register(&bdrv_gluster_rdma);
1516     bdrv_register(&bdrv_gluster_unix);
1517     bdrv_register(&bdrv_gluster_tcp);
1518     bdrv_register(&bdrv_gluster);
1519 }
1520
1521 block_init(bdrv_gluster_init);