block/archipelago.c

   1 /*
   2  * QEMU Block driver for Archipelago
   3  *
   4  * Copyright (C) 2014 Chrysostomos Nanakos <cnanakos@grnet.gr>
   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
  11 /*
  12  * VM Image on Archipelago volume is specified like this:
  13  *
  14  * file.driver=archipelago,file.volume=<volumename>
  15  * [,file.mport=<mapperd_port>[,file.vport=<vlmcd_port>]
  16  * [,file.segment=<segment_name>]]
  17  *
  18  * 'archipelago' is the protocol.
  19  *
  20  * 'mport' is the port number on which mapperd is listening. This is optional
  21  * and if not specified, QEMU will make Archipelago to use the default port.
  22  *
  23  * 'vport' is the port number on which vlmcd is listening. This is optional
  24  * and if not specified, QEMU will make Archipelago to use the default port.
  25  *
  26  * 'segment' is the name of the shared memory segment Archipelago stack
  27  * is using. This is optional and if not specified, QEMU will make Archipelago
  28  * to use the default value, 'archipelago'.
  29  *
  30  * Examples:
  31  *
  32  * file.driver=archipelago,file.volume=my_vm_volume
  33  * file.driver=archipelago,file.volume=my_vm_volume,file.mport=123
  34  * file.driver=archipelago,file.volume=my_vm_volume,file.mport=123,
  35  * file.vport=1234
  36  * file.driver=archipelago,file.volume=my_vm_volume,file.mport=123,
  37  * file.vport=1234,file.segment=my_segment
  38  */
  39
  40 #include "block/block_int.h"
  41 #include "qemu/error-report.h"
  42 #include "qemu/thread.h"
  43 #include "qapi/qmp/qint.h"
  44 #include "qapi/qmp/qstring.h"
  45 #include "qapi/qmp/qjson.h"
  46
  47 #include <inttypes.h>
  48 #include <xseg/xseg.h>
  49 #include <xseg/protocol.h>
  50
  51 #define ARCHIP_FD_READ      0
  52 #define ARCHIP_FD_WRITE     1
  53 #define MAX_REQUEST_SIZE    524288
  54
  55 #define ARCHIPELAGO_OPT_VOLUME      "volume"
  56 #define ARCHIPELAGO_OPT_SEGMENT     "segment"
  57 #define ARCHIPELAGO_OPT_MPORT       "mport"
  58 #define ARCHIPELAGO_OPT_VPORT       "vport"
  59 #define ARCHIPELAGO_DFL_MPORT       1001
  60 #define ARCHIPELAGO_DFL_VPORT       501
  61
  62 #define archipelagolog(fmt, ...) \
  63     do {                         \
  64         fprintf(stderr, "archipelago\t%-24s: " fmt, __func__, ##__VA_ARGS__); \
  65     } while (0)
  66
  67 typedef enum {
  68     ARCHIP_OP_READ,
  69     ARCHIP_OP_WRITE,
  70     ARCHIP_OP_FLUSH,
  71     ARCHIP_OP_VOLINFO,
  72 } ARCHIPCmd;
  73
  74 typedef struct ArchipelagoAIOCB {
  75     BlockDriverAIOCB common;
  76     QEMUBH *bh;
  77     struct BDRVArchipelagoState *s;
  78     QEMUIOVector *qiov;
  79     ARCHIPCmd cmd;
  80     bool cancelled;
  81     int status;
  82     int64_t size;
  83     int64_t ret;
  84 } ArchipelagoAIOCB;
  85
  86 typedef struct BDRVArchipelagoState {
  87     ArchipelagoAIOCB *event_acb;
  88     char *volname;
  89     char *segment_name;
  90     uint64_t size;
  91     /* Archipelago specific */
  92     struct xseg *xseg;
  93     struct xseg_port *port;
  94     xport srcport;
  95     xport sport;
  96     xport mportno;
  97     xport vportno;
  98     QemuMutex archip_mutex;
  99     QemuCond archip_cond;
 100     bool is_signaled;
 101     /* Request handler specific */
 102     QemuThread request_th;
 103     QemuCond request_cond;
 104     QemuMutex request_mutex;
 105     bool th_is_signaled;
 106     bool stopping;
 107 } BDRVArchipelagoState;
 108
 109 typedef struct ArchipelagoSegmentedRequest {
 110     size_t count;
 111     size_t total;
 112     int ref;
 113     int failed;
 114 } ArchipelagoSegmentedRequest;
 115
 116 typedef struct AIORequestData {
 117     const char *volname;
 118     off_t offset;
 119     size_t size;
 120     uint64_t bufidx;
 121     int ret;
 122     int op;
 123     ArchipelagoAIOCB *aio_cb;
 124     ArchipelagoSegmentedRequest *segreq;
 125 } AIORequestData;
 126
 127 static void qemu_archipelago_complete_aio(void *opaque);
 128
 129 static void init_local_signal(struct xseg *xseg, xport sport, xport srcport)
 130 {
 131     if (xseg && (sport != srcport)) {
 132         xseg_init_local_signal(xseg, srcport);
 133         sport = srcport;
 134     }
 135 }
 136
 137 static void archipelago_finish_aiocb(AIORequestData *reqdata)
 138 {
 139     if (reqdata->aio_cb->ret != reqdata->segreq->total) {
 140         reqdata->aio_cb->ret = -EIO;
 141     } else if (reqdata->aio_cb->ret == reqdata->segreq->total) {
 142         reqdata->aio_cb->ret = 0;
 143     }
 144     reqdata->aio_cb->bh = aio_bh_new(
 145                         bdrv_get_aio_context(reqdata->aio_cb->common.bs),
 146                         qemu_archipelago_complete_aio, reqdata
 147                         );
 148     qemu_bh_schedule(reqdata->aio_cb->bh);
 149 }
 150
 151 static int wait_reply(struct xseg *xseg, xport srcport, struct xseg_port *port,
 152                       struct xseg_request *expected_req)
 153 {
 154     struct xseg_request *req;
 155     xseg_prepare_wait(xseg, srcport);
 156     void *psd = xseg_get_signal_desc(xseg, port);
 157     while (1) {
 158         req = xseg_receive(xseg, srcport, X_NONBLOCK);
 159         if (req) {
 160             if (req != expected_req) {
 161                 archipelagolog("Unknown received request\n");
 162                 xseg_put_request(xseg, req, srcport);
 163             } else if (!(req->state & XS_SERVED)) {
 164                 return -1;
 165             } else {
 166                 break;
 167             }
 168         }
 169         xseg_wait_signal(xseg, psd, 100000UL);
 170     }
 171     xseg_cancel_wait(xseg, srcport);
 172     return 0;
 173 }
 174
 175 static void xseg_request_handler(void *state)
 176 {
 177     BDRVArchipelagoState *s = (BDRVArchipelagoState *) state;
 178     void *psd = xseg_get_signal_desc(s->xseg, s->port);
 179     qemu_mutex_lock(&s->request_mutex);
 180
 181     while (!s->stopping) {
 182         struct xseg_request *req;
 183         void *data;
 184         xseg_prepare_wait(s->xseg, s->srcport);
 185         req = xseg_receive(s->xseg, s->srcport, X_NONBLOCK);
 186         if (req) {
 187             AIORequestData *reqdata;
 188             ArchipelagoSegmentedRequest *segreq;
 189             xseg_get_req_data(s->xseg, req, (void **)&reqdata);
 190
 191             switch (reqdata->op) {
 192             case ARCHIP_OP_READ:
 193                 data = xseg_get_data(s->xseg, req);
 194                 segreq = reqdata->segreq;
 195                 segreq->count += req->serviced;
 196
 197                 qemu_iovec_from_buf(reqdata->aio_cb->qiov, reqdata->bufidx,
 198                                     data,
 199                                     req->serviced);
 200
 201                 xseg_put_request(s->xseg, req, s->srcport);
 202
 203                 if ((__sync_add_and_fetch(&segreq->ref, -1)) == 0) {
 204                     if (!segreq->failed) {
 205                         reqdata->aio_cb->ret = segreq->count;
 206                         archipelago_finish_aiocb(reqdata);
 207                         g_free(segreq);
 208                     } else {
 209                         g_free(segreq);
 210                         g_free(reqdata);
 211                     }
 212                 } else {
 213                     g_free(reqdata);
 214                 }
 215                 break;
 216             case ARCHIP_OP_WRITE:
 217             case ARCHIP_OP_FLUSH:
 218                 segreq = reqdata->segreq;
 219                 segreq->count += req->serviced;
 220                 xseg_put_request(s->xseg, req, s->srcport);
 221
 222                 if ((__sync_add_and_fetch(&segreq->ref, -1)) == 0) {
 223                     if (!segreq->failed) {
 224                         reqdata->aio_cb->ret = segreq->count;
 225                         archipelago_finish_aiocb(reqdata);
 226                         g_free(segreq);
 227                     } else {
 228                         g_free(segreq);
 229                         g_free(reqdata);
 230                     }
 231                 } else {
 232                     g_free(reqdata);
 233                 }
 234                 break;
 235             case ARCHIP_OP_VOLINFO:
 236                 s->is_signaled = true;
 237                 qemu_cond_signal(&s->archip_cond);
 238                 break;
 239             }
 240         } else {
 241             xseg_wait_signal(s->xseg, psd, 100000UL);
 242         }
 243         xseg_cancel_wait(s->xseg, s->srcport);
 244     }
 245
 246     s->th_is_signaled = true;
 247     qemu_cond_signal(&s->request_cond);
 248     qemu_mutex_unlock(&s->request_mutex);
 249     qemu_thread_exit(NULL);
 250 }
 251
 252 static int qemu_archipelago_xseg_init(BDRVArchipelagoState *s)
 253 {
 254     if (xseg_initialize()) {
 255         archipelagolog("Cannot initialize XSEG\n");
 256         goto err_exit;
 257     }
 258
 259     s->xseg = xseg_join("posix", s->segment_name,
 260                         "posixfd", NULL);
 261     if (!s->xseg) {
 262         archipelagolog("Cannot join XSEG shared memory segment\n");
 263         goto err_exit;
 264     }
 265     s->port = xseg_bind_dynport(s->xseg);
 266     s->srcport = s->port->portno;
 267     init_local_signal(s->xseg, s->sport, s->srcport);
 268     return 0;
 269
 270 err_exit:
 271     return -1;
 272 }
 273
 274 static int qemu_archipelago_init(BDRVArchipelagoState *s)
 275 {
 276     int ret;
 277
 278     ret = qemu_archipelago_xseg_init(s);
 279     if (ret < 0) {
 280         error_report("Cannot initialize XSEG. Aborting...\n");
 281         goto err_exit;
 282     }
 283
 284     qemu_cond_init(&s->archip_cond);
 285     qemu_mutex_init(&s->archip_mutex);
 286     qemu_cond_init(&s->request_cond);
 287     qemu_mutex_init(&s->request_mutex);
 288     s->th_is_signaled = false;
 289     qemu_thread_create(&s->request_th, "xseg_io_th",
 290                        (void *) xseg_request_handler,
 291                        (void *) s, QEMU_THREAD_JOINABLE);
 292
 293 err_exit:
 294     return ret;
 295 }
 296
 297 static void qemu_archipelago_complete_aio(void *opaque)
 298 {
 299     AIORequestData *reqdata = (AIORequestData *) opaque;
 300     ArchipelagoAIOCB *aio_cb = (ArchipelagoAIOCB *) reqdata->aio_cb;
 301
 302     qemu_bh_delete(aio_cb->bh);
 303     aio_cb->common.cb(aio_cb->common.opaque, aio_cb->ret);
 304     aio_cb->status = 0;
 305
 306     if (!aio_cb->cancelled) {
 307         qemu_aio_release(aio_cb);
 308     }
 309     g_free(reqdata);
 310 }
 311
 312 static QemuOptsList archipelago_runtime_opts = {
 313     .name = "archipelago",
 314     .head = QTAILQ_HEAD_INITIALIZER(archipelago_runtime_opts.head),
 315     .desc = {
 316         {
 317             .name = ARCHIPELAGO_OPT_VOLUME,
 318             .type = QEMU_OPT_STRING,
 319             .help = "Name of the volume image",
 320         },
 321         {
 322             .name = ARCHIPELAGO_OPT_SEGMENT,
 323             .type = QEMU_OPT_STRING,
 324             .help = "Name of the Archipelago shared memory segment",
 325         },
 326         {
 327             .name = ARCHIPELAGO_OPT_MPORT,
 328             .type = QEMU_OPT_NUMBER,
 329             .help = "Archipelago mapperd port number"
 330         },
 331         {
 332             .name = ARCHIPELAGO_OPT_VPORT,
 333             .type = QEMU_OPT_NUMBER,
 334             .help = "Archipelago vlmcd port number"
 335
 336         },
 337         { /* end of list */ }
 338     },
 339 };
 340
 341 static int qemu_archipelago_open(BlockDriverState *bs,
 342                                  QDict *options,
 343                                  int bdrv_flags,
 344                                  Error **errp)
 345 {
 346     int ret = 0;
 347     const char *volume, *segment_name;
 348     QemuOpts *opts;
 349     Error *local_err = NULL;
 350     BDRVArchipelagoState *s = bs->opaque;
 351
 352     opts = qemu_opts_create(&archipelago_runtime_opts, NULL, 0, &error_abort);
 353     qemu_opts_absorb_qdict(opts, options, &local_err);
 354     if (local_err) {
 355         error_propagate(errp, local_err);
 356         ret = -EINVAL;
 357         goto err_exit;
 358     }
 359
 360     s->mportno = qemu_opt_get_number(opts, ARCHIPELAGO_OPT_MPORT,
 361                                      ARCHIPELAGO_DFL_MPORT);
 362     s->vportno = qemu_opt_get_number(opts, ARCHIPELAGO_OPT_VPORT,
 363                                      ARCHIPELAGO_DFL_VPORT);
 364
 365     segment_name = qemu_opt_get(opts, ARCHIPELAGO_OPT_SEGMENT);
 366     if (segment_name == NULL) {
 367         s->segment_name = g_strdup("archipelago");
 368     } else {
 369         s->segment_name = g_strdup(segment_name);
 370     }
 371
 372     volume = qemu_opt_get(opts, ARCHIPELAGO_OPT_VOLUME);
 373     if (volume == NULL) {
 374         error_setg(errp, "archipelago block driver requires the 'volume'"
 375                    " option");
 376         ret = -EINVAL;
 377         goto err_exit;
 378     }
 379     s->volname = g_strdup(volume);
 380
 381     /* Initialize XSEG, join shared memory segment */
 382     ret = qemu_archipelago_init(s);
 383     if (ret < 0) {
 384         error_setg(errp, "cannot initialize XSEG and join shared "
 385                    "memory segment");
 386         goto err_exit;
 387     }
 388
 389     qemu_opts_del(opts);
 390     return 0;
 391
 392 err_exit:
 393     g_free(s->volname);
 394     g_free(s->segment_name);
 395     qemu_opts_del(opts);
 396     return ret;
 397 }
 398
 399 static void qemu_archipelago_close(BlockDriverState *bs)
 400 {
 401     int r, targetlen;
 402     char *target;
 403     struct xseg_request *req;
 404     BDRVArchipelagoState *s = bs->opaque;
 405
 406     s->stopping = true;
 407
 408     qemu_mutex_lock(&s->request_mutex);
 409     while (!s->th_is_signaled) {
 410         qemu_cond_wait(&s->request_cond,
 411                        &s->request_mutex);
 412     }
 413     qemu_mutex_unlock(&s->request_mutex);
 414     qemu_thread_join(&s->request_th);
 415     qemu_cond_destroy(&s->request_cond);
 416     qemu_mutex_destroy(&s->request_mutex);
 417
 418     qemu_cond_destroy(&s->archip_cond);
 419     qemu_mutex_destroy(&s->archip_mutex);
 420
 421     targetlen = strlen(s->volname);
 422     req = xseg_get_request(s->xseg, s->srcport, s->vportno, X_ALLOC);
 423     if (!req) {
 424         archipelagolog("Cannot get XSEG request\n");
 425         goto err_exit;
 426     }
 427     r = xseg_prep_request(s->xseg, req, targetlen, 0);
 428     if (r < 0) {
 429         xseg_put_request(s->xseg, req, s->srcport);
 430         archipelagolog("Cannot prepare XSEG close request\n");
 431         goto err_exit;
 432     }
 433
 434     target = xseg_get_target(s->xseg, req);
 435     memcpy(target, s->volname, targetlen);
 436     req->size = req->datalen;
 437     req->offset = 0;
 438     req->op = X_CLOSE;
 439
 440     xport p = xseg_submit(s->xseg, req, s->srcport, X_ALLOC);
 441     if (p == NoPort) {
 442         xseg_put_request(s->xseg, req, s->srcport);
 443         archipelagolog("Cannot submit XSEG close request\n");
 444         goto err_exit;
 445     }
 446
 447     xseg_signal(s->xseg, p);
 448     wait_reply(s->xseg, s->srcport, s->port, req);
 449
 450     xseg_put_request(s->xseg, req, s->srcport);
 451
 452 err_exit:
 453     g_free(s->volname);
 454     g_free(s->segment_name);
 455     xseg_quit_local_signal(s->xseg, s->srcport);
 456     xseg_leave_dynport(s->xseg, s->port);
 457     xseg_leave(s->xseg);
 458 }
 459
 460 static void qemu_archipelago_aio_cancel(BlockDriverAIOCB *blockacb)
 461 {
 462     ArchipelagoAIOCB *aio_cb = (ArchipelagoAIOCB *) blockacb;
 463     aio_cb->cancelled = true;
 464     while (aio_cb->status == -EINPROGRESS) {
 465         aio_poll(bdrv_get_aio_context(aio_cb->common.bs), true);
 466     }
 467     qemu_aio_release(aio_cb);
 468 }
 469
 470 static const AIOCBInfo archipelago_aiocb_info = {
 471     .aiocb_size = sizeof(ArchipelagoAIOCB),
 472     .cancel = qemu_archipelago_aio_cancel,
 473 };
 474
 475 static int archipelago_submit_request(BDRVArchipelagoState *s,
 476                                         uint64_t bufidx,
 477                                         size_t count,
 478                                         off_t offset,
 479                                         ArchipelagoAIOCB *aio_cb,
 480                                         ArchipelagoSegmentedRequest *segreq,
 481                                         int op)
 482 {
 483     int ret, targetlen;
 484     char *target;
 485     void *data = NULL;
 486     struct xseg_request *req;
 487     AIORequestData *reqdata = g_malloc(sizeof(AIORequestData));
 488
 489     targetlen = strlen(s->volname);
 490     req = xseg_get_request(s->xseg, s->srcport, s->vportno, X_ALLOC);
 491     if (!req) {
 492         archipelagolog("Cannot get XSEG request\n");
 493         goto err_exit2;
 494     }
 495     ret = xseg_prep_request(s->xseg, req, targetlen, count);
 496     if (ret < 0) {
 497         archipelagolog("Cannot prepare XSEG request\n");
 498         goto err_exit;
 499     }
 500     target = xseg_get_target(s->xseg, req);
 501     if (!target) {
 502         archipelagolog("Cannot get XSEG target\n");
 503         goto err_exit;
 504     }
 505     memcpy(target, s->volname, targetlen);
 506     req->size = count;
 507     req->offset = offset;
 508
 509     switch (op) {
 510     case ARCHIP_OP_READ:
 511         req->op = X_READ;
 512         break;
 513     case ARCHIP_OP_WRITE:
 514         req->op = X_WRITE;
 515         break;
 516     case ARCHIP_OP_FLUSH:
 517         req->op = X_FLUSH;
 518         break;
 519     }
 520     reqdata->volname = s->volname;
 521     reqdata->offset = offset;
 522     reqdata->size = count;
 523     reqdata->bufidx = bufidx;
 524     reqdata->aio_cb = aio_cb;
 525     reqdata->segreq = segreq;
 526     reqdata->op = op;
 527
 528     xseg_set_req_data(s->xseg, req, reqdata);
 529     if (op == ARCHIP_OP_WRITE) {
 530         data = xseg_get_data(s->xseg, req);
 531         if (!data) {
 532             archipelagolog("Cannot get XSEG data\n");
 533             goto err_exit;
 534         }
 535         qemu_iovec_to_buf(aio_cb->qiov, bufidx, data, count);
 536     }
 537
 538     xport p = xseg_submit(s->xseg, req, s->srcport, X_ALLOC);
 539     if (p == NoPort) {
 540         archipelagolog("Could not submit XSEG request\n");
 541         goto err_exit;
 542     }
 543     xseg_signal(s->xseg, p);
 544     return 0;
 545
 546 err_exit:
 547     g_free(reqdata);
 548     xseg_put_request(s->xseg, req, s->srcport);
 549     return -EIO;
 550 err_exit2:
 551     g_free(reqdata);
 552     return -EIO;
 553 }
 554
 555 static int archipelago_aio_segmented_rw(BDRVArchipelagoState *s,
 556                                         size_t count,
 557                                         off_t offset,
 558                                         ArchipelagoAIOCB *aio_cb,
 559                                         int op)
 560 {
 561     int i, ret, segments_nr, last_segment_size;
 562     ArchipelagoSegmentedRequest *segreq;
 563
 564     segreq = g_malloc(sizeof(ArchipelagoSegmentedRequest));
 565
 566     if (op == ARCHIP_OP_FLUSH) {
 567         segments_nr = 1;
 568         segreq->ref = segments_nr;
 569         segreq->total = count;
 570         segreq->count = 0;
 571         segreq->failed = 0;
 572         ret = archipelago_submit_request(s, 0, count, offset, aio_cb,
 573                                            segreq, ARCHIP_OP_FLUSH);
 574         if (ret < 0) {
 575             goto err_exit;
 576         }
 577         return 0;
 578     }
 579
 580     segments_nr = (int)(count / MAX_REQUEST_SIZE) + \
 581                   ((count % MAX_REQUEST_SIZE) ? 1 : 0);
 582     last_segment_size = (int)(count % MAX_REQUEST_SIZE);
 583
 584     segreq->ref = segments_nr;
 585     segreq->total = count;
 586     segreq->count = 0;
 587     segreq->failed = 0;
 588
 589     for (i = 0; i < segments_nr - 1; i++) {
 590         ret = archipelago_submit_request(s, i * MAX_REQUEST_SIZE,
 591                                            MAX_REQUEST_SIZE,
 592                                            offset + i * MAX_REQUEST_SIZE,
 593                                            aio_cb, segreq, op);
 594
 595         if (ret < 0) {
 596             goto err_exit;
 597         }
 598     }
 599
 600     if ((segments_nr > 1) && last_segment_size) {
 601         ret = archipelago_submit_request(s, i * MAX_REQUEST_SIZE,
 602                                            last_segment_size,
 603                                            offset + i * MAX_REQUEST_SIZE,
 604                                            aio_cb, segreq, op);
 605     } else if ((segments_nr > 1) && !last_segment_size) {
 606         ret = archipelago_submit_request(s, i * MAX_REQUEST_SIZE,
 607                                            MAX_REQUEST_SIZE,
 608                                            offset + i * MAX_REQUEST_SIZE,
 609                                            aio_cb, segreq, op);
 610     } else if (segments_nr == 1) {
 611             ret = archipelago_submit_request(s, 0, count, offset, aio_cb,
 612                                                segreq, op);
 613     }
 614
 615     if (ret < 0) {
 616         goto err_exit;
 617     }
 618
 619     return 0;
 620
 621 err_exit:
 622     __sync_add_and_fetch(&segreq->failed, 1);
 623     if (segments_nr == 1) {
 624         if (__sync_add_and_fetch(&segreq->ref, -1) == 0) {
 625             g_free(segreq);
 626         }
 627     } else {
 628         if ((__sync_add_and_fetch(&segreq->ref, -segments_nr + i)) == 0) {
 629             g_free(segreq);
 630         }
 631     }
 632
 633     return ret;
 634 }
 635
 636 static BlockDriverAIOCB *qemu_archipelago_aio_rw(BlockDriverState *bs,
 637                                                  int64_t sector_num,
 638                                                  QEMUIOVector *qiov,
 639                                                  int nb_sectors,
 640                                                  BlockDriverCompletionFunc *cb,
 641                                                  void *opaque,
 642                                                  int op)
 643 {
 644     ArchipelagoAIOCB *aio_cb;
 645     BDRVArchipelagoState *s = bs->opaque;
 646     int64_t size, off;
 647     int ret;
 648
 649     aio_cb = qemu_aio_get(&archipelago_aiocb_info, bs, cb, opaque);
 650     aio_cb->cmd = op;
 651     aio_cb->qiov = qiov;
 652
 653     aio_cb->ret = 0;
 654     aio_cb->s = s;
 655     aio_cb->cancelled = false;
 656     aio_cb->status = -EINPROGRESS;
 657
 658     off = sector_num * BDRV_SECTOR_SIZE;
 659     size = nb_sectors * BDRV_SECTOR_SIZE;
 660     aio_cb->size = size;
 661
 662     ret = archipelago_aio_segmented_rw(s, size, off,
 663                                        aio_cb, op);
 664     if (ret < 0) {
 665         goto err_exit;
 666     }
 667     return &aio_cb->common;
 668
 669 err_exit:
 670     error_report("qemu_archipelago_aio_rw(): I/O Error\n");
 671     qemu_aio_release(aio_cb);
 672     return NULL;
 673 }
 674
 675 static BlockDriverAIOCB *qemu_archipelago_aio_readv(BlockDriverState *bs,
 676         int64_t sector_num, QEMUIOVector *qiov, int nb_sectors,
 677         BlockDriverCompletionFunc *cb, void *opaque)
 678 {
 679     return qemu_archipelago_aio_rw(bs, sector_num, qiov, nb_sectors, cb,
 680                                    opaque, ARCHIP_OP_READ);
 681 }
 682
 683 static BlockDriverAIOCB *qemu_archipelago_aio_writev(BlockDriverState *bs,
 684         int64_t sector_num, QEMUIOVector *qiov, int nb_sectors,
 685         BlockDriverCompletionFunc *cb, void *opaque)
 686 {
 687     return qemu_archipelago_aio_rw(bs, sector_num, qiov, nb_sectors, cb,
 688                                    opaque, ARCHIP_OP_WRITE);
 689 }
 690
 691 static int64_t archipelago_volume_info(BDRVArchipelagoState *s)
 692 {
 693     uint64_t size;
 694     int ret, targetlen;
 695     struct xseg_request *req;
 696     struct xseg_reply_info *xinfo;
 697     AIORequestData *reqdata = g_malloc(sizeof(AIORequestData));
 698
 699     const char *volname = s->volname;
 700     targetlen = strlen(volname);
 701     req = xseg_get_request(s->xseg, s->srcport, s->mportno, X_ALLOC);
 702     if (!req) {
 703         archipelagolog("Cannot get XSEG request\n");
 704         goto err_exit2;
 705     }
 706     ret = xseg_prep_request(s->xseg, req, targetlen,
 707                             sizeof(struct xseg_reply_info));
 708     if (ret < 0) {
 709         archipelagolog("Cannot prepare XSEG request\n");
 710         goto err_exit;
 711     }
 712     char *target = xseg_get_target(s->xseg, req);
 713     if (!target) {
 714         archipelagolog("Cannot get XSEG target\n");
 715         goto err_exit;
 716     }
 717     memcpy(target, volname, targetlen);
 718     req->size = req->datalen;
 719     req->offset = 0;
 720     req->op = X_INFO;
 721
 722     reqdata->op = ARCHIP_OP_VOLINFO;
 723     reqdata->volname = volname;
 724     xseg_set_req_data(s->xseg, req, reqdata);
 725
 726     xport p = xseg_submit(s->xseg, req, s->srcport, X_ALLOC);
 727     if (p == NoPort) {
 728         archipelagolog("Cannot submit XSEG request\n");
 729         goto err_exit;
 730     }
 731     xseg_signal(s->xseg, p);
 732     qemu_mutex_lock(&s->archip_mutex);
 733     while (!s->is_signaled) {
 734         qemu_cond_wait(&s->archip_cond, &s->archip_mutex);
 735     }
 736     s->is_signaled = false;
 737     qemu_mutex_unlock(&s->archip_mutex);
 738
 739     xinfo = (struct xseg_reply_info *) xseg_get_data(s->xseg, req);
 740     size = xinfo->size;
 741     xseg_put_request(s->xseg, req, s->srcport);
 742     g_free(reqdata);
 743     s->size = size;
 744     return size;
 745
 746 err_exit:
 747     xseg_put_request(s->xseg, req, s->srcport);
 748 err_exit2:
 749     g_free(reqdata);
 750     return -EIO;
 751 }
 752
 753 static int64_t qemu_archipelago_getlength(BlockDriverState *bs)
 754 {
 755     int64_t ret;
 756     BDRVArchipelagoState *s = bs->opaque;
 757
 758     ret = archipelago_volume_info(s);
 759     return ret;
 760 }
 761
 762 static BlockDriverAIOCB *qemu_archipelago_aio_flush(BlockDriverState *bs,
 763         BlockDriverCompletionFunc *cb, void *opaque)
 764 {
 765     return qemu_archipelago_aio_rw(bs, 0, NULL, 0, cb, opaque,
 766                                    ARCHIP_OP_FLUSH);
 767 }
 768
 769 static BlockDriver bdrv_archipelago = {
 770     .format_name         = "archipelago",
 771     .protocol_name       = "archipelago",
 772     .instance_size       = sizeof(BDRVArchipelagoState),
 773     .bdrv_file_open      = qemu_archipelago_open,
 774     .bdrv_close          = qemu_archipelago_close,
 775     .bdrv_getlength      = qemu_archipelago_getlength,
 776     .bdrv_aio_readv      = qemu_archipelago_aio_readv,
 777     .bdrv_aio_writev     = qemu_archipelago_aio_writev,
 778     .bdrv_aio_flush      = qemu_archipelago_aio_flush,
 779     .bdrv_has_zero_init  = bdrv_has_zero_init_1,
 780 };
 781
 782 static void bdrv_archipelago_init(void)
 783 {
 784     bdrv_register(&bdrv_archipelago);
 785 }
 786
 787 block_init(bdrv_archipelago_init);