text: /*
   * QEMU Block driver for Archipelago
   *
   * Copyright (C) 2014 Chrysostomos Nanakos <cnanakos@grnet.gr>
   *
   * This work is licensed under the terms of the GNU GPL, version 2 or later.
   * See the COPYING file in the top-level directory.
   *
   */
  
  /*
   * VM Image on Archipelago volume is specified like this:
   *
   * file.driver=archipelago,file.volume=<volumename>
   * [,file.mport=<mapperd_port>[,file.vport=<vlmcd_port>]
   * [,file.segment=<segment_name>]]
   *
   * or
   *
   * file=archipelago:<volumename>[/mport=<mapperd_port>[:vport=<vlmcd_port>][:
   * segment=<segment_name>]]
   *
   * 'archipelago' is the protocol.
   *
   * 'mport' is the port number on which mapperd is listening. This is optional
   * and if not specified, QEMU will make Archipelago to use the default port.
   *
   * 'vport' is the port number on which vlmcd is listening. This is optional
   * and if not specified, QEMU will make Archipelago to use the default port.
   *
   * 'segment' is the name of the shared memory segment Archipelago stack
   * is using. This is optional and if not specified, QEMU will make Archipelago
   * to use the default value, 'archipelago'.
   *
   * Examples:
   *
   * file.driver=archipelago,file.volume=my_vm_volume
   * file.driver=archipelago,file.volume=my_vm_volume,file.mport=123
   * file.driver=archipelago,file.volume=my_vm_volume,file.mport=123,
   *  file.vport=1234
   * file.driver=archipelago,file.volume=my_vm_volume,file.mport=123,
   *  file.vport=1234,file.segment=my_segment
   *
   * or
   *
   * file=archipelago:my_vm_volume
   * file=archipelago:my_vm_volume/mport=123
   * file=archipelago:my_vm_volume/mport=123:vport=1234
   * file=archipelago:my_vm_volume/mport=123:vport=1234:segment=my_segment
   *
   */
  
  #include "qemu-common.h"
  #include "block/block_int.h"
  #include "qemu/error-report.h"
  #include "qemu/thread.h"
  #include "qapi/qmp/qint.h"
  #include "qapi/qmp/qstring.h"
  #include "qapi/qmp/qjson.h"
  #include "qemu/atomic.h"
  
  #include <inttypes.h>
  #include <xseg/xseg.h>
  #include <xseg/protocol.h>
  
  #define MAX_REQUEST_SIZE    524288
  
  #define ARCHIPELAGO_OPT_VOLUME      "volume"
  #define ARCHIPELAGO_OPT_SEGMENT     "segment"
  #define ARCHIPELAGO_OPT_MPORT       "mport"
  #define ARCHIPELAGO_OPT_VPORT       "vport"
  #define ARCHIPELAGO_DFL_MPORT       1001
  #define ARCHIPELAGO_DFL_VPORT       501
  
  #define archipelagolog(fmt, ...) \
      do {                         \
          fprintf(stderr, "archipelago\t%-24s: " fmt, __func__, ##__VA_ARGS__); \
      } while (0)
  
  typedef enum {
      ARCHIP_OP_READ,
      ARCHIP_OP_WRITE,
      ARCHIP_OP_FLUSH,
      ARCHIP_OP_VOLINFO,
      ARCHIP_OP_TRUNCATE,
  } ARCHIPCmd;
  
  typedef struct ArchipelagoAIOCB {
      BlockAIOCB common;
      QEMUBH *bh;
      struct BDRVArchipelagoState *s;
      QEMUIOVector *qiov;
      ARCHIPCmd cmd;
      int status;
      int64_t size;
      int64_t ret;
  } ArchipelagoAIOCB;
  
  typedef struct BDRVArchipelagoState {
      ArchipelagoAIOCB *event_acb;
      char *volname;
      char *segment_name;
      uint64_t size;
      /* Archipelago specific */
      struct xseg *xseg;
      struct xseg_port *port;
      xport srcport;
      xport sport;
      xport mportno;
      xport vportno;
      QemuMutex archip_mutex;
      QemuCond archip_cond;
      bool is_signaled;
      /* Request handler specific */
      QemuThread request_th;
      QemuCond request_cond;
      QemuMutex request_mutex;
      bool th_is_signaled;
      bool stopping;
  } BDRVArchipelagoState;
  
  typedef struct ArchipelagoSegmentedRequest {
      size_t count;
      size_t total;
      int ref;
      int failed;
  } ArchipelagoSegmentedRequest;
  
  typedef struct AIORequestData {
      const char *volname;
      off_t offset;
      size_t size;
      uint64_t bufidx;
      int ret;
      int op;
      ArchipelagoAIOCB *aio_cb;
      ArchipelagoSegmentedRequest *segreq;
  } AIORequestData;
  
  static void qemu_archipelago_complete_aio(void *opaque);
  
  static void init_local_signal(struct xseg *xseg, xport sport, xport srcport)
  {
      if (xseg && (sport != srcport)) {
          xseg_init_local_signal(xseg, srcport);
          sport = srcport;
      }
  }
  
  static void archipelago_finish_aiocb(AIORequestData *reqdata)
  {
      if (reqdata->aio_cb->ret != reqdata->segreq->total) {
          reqdata->aio_cb->ret = -EIO;
      } else if (reqdata->aio_cb->ret == reqdata->segreq->total) {
          reqdata->aio_cb->ret = 0;
      }
      reqdata->aio_cb->bh = aio_bh_new(
                          bdrv_get_aio_context(reqdata->aio_cb->common.bs),
                          qemu_archipelago_complete_aio, reqdata
                          );
      qemu_bh_schedule(reqdata->aio_cb->bh);
  }
  
  static int wait_reply(struct xseg *xseg, xport srcport, struct xseg_port *port,
                        struct xseg_request *expected_req)
  {
      struct xseg_request *req;
      xseg_prepare_wait(xseg, srcport);
      void *psd = xseg_get_signal_desc(xseg, port);
      while (1) {
          req = xseg_receive(xseg, srcport, X_NONBLOCK);
          if (req) {
              if (req != expected_req) {
                  archipelagolog("Unknown received request\n");
                  xseg_put_request(xseg, req, srcport);
              } else if (!(req->state & XS_SERVED)) {
                  return -1;
              } else {
                  break;
              }
          }
          xseg_wait_signal(xseg, psd, 100000UL);
      }
      xseg_cancel_wait(xseg, srcport);
      return 0;
  }
  
  static void xseg_request_handler(void *state)
  {
      BDRVArchipelagoState *s = (BDRVArchipelagoState *) state;
      void *psd = xseg_get_signal_desc(s->xseg, s->port);
      qemu_mutex_lock(&s->request_mutex);
  
      while (!s->stopping) {
          struct xseg_request *req;
          void *data;
          xseg_prepare_wait(s->xseg, s->srcport);
          req = xseg_receive(s->xseg, s->srcport, X_NONBLOCK);
          if (req) {
              AIORequestData *reqdata;
              ArchipelagoSegmentedRequest *segreq;
              xseg_get_req_data(s->xseg, req, (void **)&reqdata);
  
              switch (reqdata->op) {
              case ARCHIP_OP_READ:
                  data = xseg_get_data(s->xseg, req);
                  segreq = reqdata->segreq;
                  segreq->count += req->serviced;
  
                  qemu_iovec_from_buf(reqdata->aio_cb->qiov, reqdata->bufidx,
                                      data,
                                      req->serviced);
  
                  xseg_put_request(s->xseg, req, s->srcport);
  
                  if (atomic_fetch_dec(&segreq->ref) == 1) {
                      if (!segreq->failed) {
                          reqdata->aio_cb->ret = segreq->count;
                          archipelago_finish_aiocb(reqdata);
                          g_free(segreq);
                      } else {
                          g_free(segreq);
                          g_free(reqdata);
                      }
                  } else {
                      g_free(reqdata);
                  }
                  break;
              case ARCHIP_OP_WRITE:
              case ARCHIP_OP_FLUSH:
                  segreq = reqdata->segreq;
                  segreq->count += req->serviced;
                  xseg_put_request(s->xseg, req, s->srcport);
  
                  if (atomic_fetch_dec(&segreq->ref) == 1) {
                      if (!segreq->failed) {
                          reqdata->aio_cb->ret = segreq->count;
                          archipelago_finish_aiocb(reqdata);
                          g_free(segreq);
                      } else {
                          g_free(segreq);
                          g_free(reqdata);
                      }
                  } else {
                      g_free(reqdata);
                  }
                  break;
              case ARCHIP_OP_VOLINFO:
              case ARCHIP_OP_TRUNCATE:
                  s->is_signaled = true;
                  qemu_cond_signal(&s->archip_cond);
                  break;
              }
          } else {
              xseg_wait_signal(s->xseg, psd, 100000UL);
          }
          xseg_cancel_wait(s->xseg, s->srcport);
      }
  
      s->th_is_signaled = true;
      qemu_cond_signal(&s->request_cond);
      qemu_mutex_unlock(&s->request_mutex);
      qemu_thread_exit(NULL);
  }
  
  static int qemu_archipelago_xseg_init(BDRVArchipelagoState *s)
  {
      if (xseg_initialize()) {
          archipelagolog("Cannot initialize XSEG\n");
          goto err_exit;
      }
  
      s->xseg = xseg_join("posix", s->segment_name,
                          "posixfd", NULL);
      if (!s->xseg) {
          archipelagolog("Cannot join XSEG shared memory segment\n");
          goto err_exit;
      }
      s->port = xseg_bind_dynport(s->xseg);
      s->srcport = s->port->portno;
      init_local_signal(s->xseg, s->sport, s->srcport);
      return 0;
  
  err_exit:
      return -1;
  }
  
  static int qemu_archipelago_init(BDRVArchipelagoState *s)
  {
      int ret;
  
      ret = qemu_archipelago_xseg_init(s);
      if (ret < 0) {
          error_report("Cannot initialize XSEG. Aborting...\n");
          goto err_exit;
      }
  
      qemu_cond_init(&s->archip_cond);
      qemu_mutex_init(&s->archip_mutex);
      qemu_cond_init(&s->request_cond);
      qemu_mutex_init(&s->request_mutex);
      s->th_is_signaled = false;
      qemu_thread_create(&s->request_th, "xseg_io_th",
                         (void *) xseg_request_handler,
                         (void *) s, QEMU_THREAD_JOINABLE);
  
  err_exit:
      return ret;
  }
  
  static void qemu_archipelago_complete_aio(void *opaque)
  {
      AIORequestData *reqdata = (AIORequestData *) opaque;
      ArchipelagoAIOCB *aio_cb = (ArchipelagoAIOCB *) reqdata->aio_cb;
  
      qemu_bh_delete(aio_cb->bh);
      aio_cb->common.cb(aio_cb->common.opaque, aio_cb->ret);
      aio_cb->status = 0;
  
      qemu_aio_unref(aio_cb);
      g_free(reqdata);
  }
  
  static void xseg_find_port(char *pstr, const char *needle, xport *aport)
  {
      const char *a;
      char *endptr = NULL;
      unsigned long port;
      if (strstart(pstr, needle, &a)) {
          if (strlen(a) > 0) {
              port = strtoul(a, &endptr, 10);
              if (strlen(endptr)) {
                  *aport = -2;
                  return;
              }
              *aport = (xport) port;
          }
      }
  }
  
  static void xseg_find_segment(char *pstr, const char *needle,
                                char **segment_name)
  {
      const char *a;
      if (strstart(pstr, needle, &a)) {
          if (strlen(a) > 0) {
              *segment_name = g_strdup(a);
          }
      }
  }
  
  static void parse_filename_opts(const char *filename, Error **errp,
                                  char **volume, char **segment_name,
                                  xport *mport, xport *vport)
  {
      const char *start;
      char *tokens[4], *ds;
      int idx;
      xport lmport = NoPort, lvport = NoPort;
  
      strstart(filename, "archipelago:", &start);
  
      ds = g_strdup(start);
      tokens[0] = strtok(ds, "/");
      tokens[1] = strtok(NULL, ":");
      tokens[2] = strtok(NULL, ":");
      tokens[3] = strtok(NULL, "\0");
  
      if (!strlen(tokens[0])) {
          error_setg(errp, "volume name must be specified first");
          g_free(ds);
          return;
      }
  
      for (idx = 1; idx < 4; idx++) {
          if (tokens[idx] != NULL) {
              if (strstart(tokens[idx], "mport=", NULL)) {
                  xseg_find_port(tokens[idx], "mport=", &lmport);
              }
              if (strstart(tokens[idx], "vport=", NULL)) {
                  xseg_find_port(tokens[idx], "vport=", &lvport);
              }
              if (strstart(tokens[idx], "segment=", NULL)) {
                  xseg_find_segment(tokens[idx], "segment=", segment_name);
              }
          }
      }
  
      if ((lmport == -2) || (lvport == -2)) {
          error_setg(errp, "mport and/or vport must be set");
          g_free(ds);
          return;
      }
      *volume = g_strdup(tokens[0]);
      *mport = lmport;
      *vport = lvport;
      g_free(ds);
  }
  
  static void archipelago_parse_filename(const char *filename, QDict *options,
                                         Error **errp)
  {
      const char *start;
      char *volume = NULL, *segment_name = NULL;
      xport mport = NoPort, vport = NoPort;
  
      if (qdict_haskey(options, ARCHIPELAGO_OPT_VOLUME)
              || qdict_haskey(options, ARCHIPELAGO_OPT_SEGMENT)
              || qdict_haskey(options, ARCHIPELAGO_OPT_MPORT)
              || qdict_haskey(options, ARCHIPELAGO_OPT_VPORT)) {
          error_setg(errp, "volume/mport/vport/segment and a file name may not"
                           " be specified at the same time");
          return;
      }
  
      if (!strstart(filename, "archipelago:", &start)) {
          error_setg(errp, "File name must start with 'archipelago:'");
          return;
      }
  
      if (!strlen(start) || strstart(start, "/", NULL)) {
          error_setg(errp, "volume name must be specified");
          return;
      }
  
      parse_filename_opts(filename, errp, &volume, &segment_name, &mport, &vport);
  
      if (volume) {
          qdict_put(options, ARCHIPELAGO_OPT_VOLUME, qstring_from_str(volume));
          g_free(volume);
      }
      if (segment_name) {
          qdict_put(options, ARCHIPELAGO_OPT_SEGMENT,
                    qstring_from_str(segment_name));
          g_free(segment_name);
      }
      if (mport != NoPort) {
          qdict_put(options, ARCHIPELAGO_OPT_MPORT, qint_from_int(mport));
      }
      if (vport != NoPort) {
          qdict_put(options, ARCHIPELAGO_OPT_VPORT, qint_from_int(vport));
      }
  }
  
  static QemuOptsList archipelago_runtime_opts = {
      .name = "archipelago",
      .head = QTAILQ_HEAD_INITIALIZER(archipelago_runtime_opts.head),
      .desc = {
          {
              .name = ARCHIPELAGO_OPT_VOLUME,
              .type = QEMU_OPT_STRING,
              .help = "Name of the volume image",
          },
          {
              .name = ARCHIPELAGO_OPT_SEGMENT,
              .type = QEMU_OPT_STRING,
              .help = "Name of the Archipelago shared memory segment",
          },
          {
              .name = ARCHIPELAGO_OPT_MPORT,
              .type = QEMU_OPT_NUMBER,
              .help = "Archipelago mapperd port number"
          },
          {
              .name = ARCHIPELAGO_OPT_VPORT,
              .type = QEMU_OPT_NUMBER,
              .help = "Archipelago vlmcd port number"
  
          },
          { /* end of list */ }
      },
  };
  
  static int qemu_archipelago_open(BlockDriverState *bs,
                                   QDict *options,
                                   int bdrv_flags,
                                   Error **errp)
  {
      int ret = 0;
      const char *volume, *segment_name;
      QemuOpts *opts;
      Error *local_err = NULL;
      BDRVArchipelagoState *s = bs->opaque;
  
      opts = qemu_opts_create(&archipelago_runtime_opts, NULL, 0, &error_abort);
      qemu_opts_absorb_qdict(opts, options, &local_err);
      if (local_err) {
          error_propagate(errp, local_err);
          ret = -EINVAL;
          goto err_exit;
      }
  
      s->mportno = qemu_opt_get_number(opts, ARCHIPELAGO_OPT_MPORT,
                                       ARCHIPELAGO_DFL_MPORT);
      s->vportno = qemu_opt_get_number(opts, ARCHIPELAGO_OPT_VPORT,
                                       ARCHIPELAGO_DFL_VPORT);
  
      segment_name = qemu_opt_get(opts, ARCHIPELAGO_OPT_SEGMENT);
      if (segment_name == NULL) {
          s->segment_name = g_strdup("archipelago");
      } else {
          s->segment_name = g_strdup(segment_name);
      }
  
      volume = qemu_opt_get(opts, ARCHIPELAGO_OPT_VOLUME);
      if (volume == NULL) {
          error_setg(errp, "archipelago block driver requires the 'volume'"
                     " option");
          ret = -EINVAL;
          goto err_exit;
      }
      s->volname = g_strdup(volume);
  
      /* Initialize XSEG, join shared memory segment */
      ret = qemu_archipelago_init(s);
      if (ret < 0) {
          error_setg(errp, "cannot initialize XSEG and join shared "
                     "memory segment");
          goto err_exit;
      }
  
      qemu_opts_del(opts);
      return 0;
  
  err_exit:
      g_free(s->volname);
      g_free(s->segment_name);
      qemu_opts_del(opts);
      return ret;
  }
  
  static void qemu_archipelago_close(BlockDriverState *bs)
  {
      int r, targetlen;
      char *target;
      struct xseg_request *req;
      BDRVArchipelagoState *s = bs->opaque;
  
      s->stopping = true;
  
      qemu_mutex_lock(&s->request_mutex);
      while (!s->th_is_signaled) {
          qemu_cond_wait(&s->request_cond,
                         &s->request_mutex);
      }
      qemu_mutex_unlock(&s->request_mutex);
      qemu_thread_join(&s->request_th);
      qemu_cond_destroy(&s->request_cond);
      qemu_mutex_destroy(&s->request_mutex);
  
      qemu_cond_destroy(&s->archip_cond);
      qemu_mutex_destroy(&s->archip_mutex);
  
      targetlen = strlen(s->volname);
      req = xseg_get_request(s->xseg, s->srcport, s->vportno, X_ALLOC);
      if (!req) {
          archipelagolog("Cannot get XSEG request\n");
          goto err_exit;
      }
      r = xseg_prep_request(s->xseg, req, targetlen, 0);
      if (r < 0) {
          xseg_put_request(s->xseg, req, s->srcport);
          archipelagolog("Cannot prepare XSEG close request\n");
          goto err_exit;
      }
  
      target = xseg_get_target(s->xseg, req);
      memcpy(target, s->volname, targetlen);
      req->size = req->datalen;
      req->offset = 0;
      req->op = X_CLOSE;
  
      xport p = xseg_submit(s->xseg, req, s->srcport, X_ALLOC);
      if (p == NoPort) {
          xseg_put_request(s->xseg, req, s->srcport);
          archipelagolog("Cannot submit XSEG close request\n");
          goto err_exit;
      }
  
      xseg_signal(s->xseg, p);
      wait_reply(s->xseg, s->srcport, s->port, req);
  
      xseg_put_request(s->xseg, req, s->srcport);
  
  err_exit:
      g_free(s->volname);
      g_free(s->segment_name);
      xseg_quit_local_signal(s->xseg, s->srcport);
      xseg_leave_dynport(s->xseg, s->port);
      xseg_leave(s->xseg);
  }
  
  static int qemu_archipelago_create_volume(Error **errp, const char *volname,
                                            char *segment_name,
                                            uint64_t size, xport mportno,
                                            xport vportno)
  {
      int ret, targetlen;
      struct xseg *xseg = NULL;
      struct xseg_request *req;
      struct xseg_request_clone *xclone;
      struct xseg_port *port;
      xport srcport = NoPort, sport = NoPort;
      char *target;
  
      /* Try default values if none has been set */
      if (mportno == (xport) -1) {
          mportno = ARCHIPELAGO_DFL_MPORT;
      }
  
      if (vportno == (xport) -1) {
          vportno = ARCHIPELAGO_DFL_VPORT;
      }
  
      if (xseg_initialize()) {
          error_setg(errp, "Cannot initialize XSEG");
          return -1;
      }
  
      xseg = xseg_join("posix", segment_name,
                       "posixfd", NULL);
  
      if (!xseg) {
          error_setg(errp, "Cannot join XSEG shared memory segment");
          return -1;
      }
  
      port = xseg_bind_dynport(xseg);
      srcport = port->portno;
      init_local_signal(xseg, sport, srcport);
  
      req = xseg_get_request(xseg, srcport, mportno, X_ALLOC);
      if (!req) {
          error_setg(errp, "Cannot get XSEG request");
          return -1;
      }
  
      targetlen = strlen(volname);
      ret = xseg_prep_request(xseg, req, targetlen,
                              sizeof(struct xseg_request_clone));
      if (ret < 0) {
          error_setg(errp, "Cannot prepare XSEG request");
          goto err_exit;
      }
  
      target = xseg_get_target(xseg, req);
      if (!target) {
          error_setg(errp, "Cannot get XSEG target.\n");
          goto err_exit;
      }
      memcpy(target, volname, targetlen);
      xclone = (struct xseg_request_clone *) xseg_get_data(xseg, req);
      memset(xclone->target, 0 , XSEG_MAX_TARGETLEN);
      xclone->targetlen = 0;
      xclone->size = size;
      req->offset = 0;
      req->size = req->datalen;
      req->op = X_CLONE;
  
      xport p = xseg_submit(xseg, req, srcport, X_ALLOC);
      if (p == NoPort) {
          error_setg(errp, "Could not submit XSEG request");
          goto err_exit;
      }
      xseg_signal(xseg, p);
  
      ret = wait_reply(xseg, srcport, port, req);
      if (ret < 0) {
          error_setg(errp, "wait_reply() error.");
      }
  
      xseg_put_request(xseg, req, srcport);
      xseg_quit_local_signal(xseg, srcport);
      xseg_leave_dynport(xseg, port);
      xseg_leave(xseg);
      return ret;
  
  err_exit:
      xseg_put_request(xseg, req, srcport);
      xseg_quit_local_signal(xseg, srcport);
      xseg_leave_dynport(xseg, port);
      xseg_leave(xseg);
      return -1;
  }
  
  static int qemu_archipelago_create(const char *filename,
                                     QemuOpts *options,
                                     Error **errp)
  {
      int ret = 0;
      uint64_t total_size = 0;
      char *volname = NULL, *segment_name = NULL;
      const char *start;
      xport mport = NoPort, vport = NoPort;
  
      if (!strstart(filename, "archipelago:", &start)) {
          error_setg(errp, "File name must start with 'archipelago:'");
          return -1;
      }
  
      if (!strlen(start) || strstart(start, "/", NULL)) {
          error_setg(errp, "volume name must be specified");
          return -1;
      }
  
      parse_filename_opts(filename, errp, &volname, &segment_name, &mport,
                          &vport);
      total_size = ROUND_UP(qemu_opt_get_size_del(options, BLOCK_OPT_SIZE, 0),
                            BDRV_SECTOR_SIZE);
  
      if (segment_name == NULL) {
          segment_name = g_strdup("archipelago");
      }
  
      /* Create an Archipelago volume */
      ret = qemu_archipelago_create_volume(errp, volname, segment_name,
                                           total_size, mport,
                                           vport);
  
      g_free(volname);
      g_free(segment_name);
      return ret;
  }
  
  static const AIOCBInfo archipelago_aiocb_info = {
      .aiocb_size = sizeof(ArchipelagoAIOCB),
  };
  
  static int archipelago_submit_request(BDRVArchipelagoState *s,
                                          uint64_t bufidx,
                                          size_t count,
                                          off_t offset,
                                          ArchipelagoAIOCB *aio_cb,
                                          ArchipelagoSegmentedRequest *segreq,
                                          int op)
  {
      int ret, targetlen;
      char *target;
      void *data = NULL;
      struct xseg_request *req;
      AIORequestData *reqdata = g_new(AIORequestData, 1);
  
      targetlen = strlen(s->volname);
      req = xseg_get_request(s->xseg, s->srcport, s->vportno, X_ALLOC);
      if (!req) {
          archipelagolog("Cannot get XSEG request\n");
          goto err_exit2;
      }
      ret = xseg_prep_request(s->xseg, req, targetlen, count);
      if (ret < 0) {
          archipelagolog("Cannot prepare XSEG request\n");
          goto err_exit;
      }
      target = xseg_get_target(s->xseg, req);
      if (!target) {
          archipelagolog("Cannot get XSEG target\n");
          goto err_exit;
      }
      memcpy(target, s->volname, targetlen);
      req->size = count;
      req->offset = offset;
  
      switch (op) {
      case ARCHIP_OP_READ:
          req->op = X_READ;
          break;
      case ARCHIP_OP_WRITE:
          req->op = X_WRITE;
          break;
      case ARCHIP_OP_FLUSH:
          req->op = X_FLUSH;
          break;
      }
      reqdata->volname = s->volname;
      reqdata->offset = offset;
      reqdata->size = count;
      reqdata->bufidx = bufidx;
      reqdata->aio_cb = aio_cb;
      reqdata->segreq = segreq;
      reqdata->op = op;
  
      xseg_set_req_data(s->xseg, req, reqdata);
      if (op == ARCHIP_OP_WRITE) {
          data = xseg_get_data(s->xseg, req);
          if (!data) {
              archipelagolog("Cannot get XSEG data\n");
              goto err_exit;
          }
          qemu_iovec_to_buf(aio_cb->qiov, bufidx, data, count);
      }
  
      xport p = xseg_submit(s->xseg, req, s->srcport, X_ALLOC);
      if (p == NoPort) {
          archipelagolog("Could not submit XSEG request\n");
          goto err_exit;
      }
      xseg_signal(s->xseg, p);
      return 0;
  
  err_exit:
      g_free(reqdata);
      xseg_put_request(s->xseg, req, s->srcport);
      return -EIO;
  err_exit2:
      g_free(reqdata);
      return -EIO;
  }
  
  static int archipelago_aio_segmented_rw(BDRVArchipelagoState *s,
                                          size_t count,
                                          off_t offset,
                                          ArchipelagoAIOCB *aio_cb,
                                          int op)
  {
      int ret, segments_nr;
      size_t pos = 0;
      ArchipelagoSegmentedRequest *segreq;
  
      segreq = g_new0(ArchipelagoSegmentedRequest, 1);
  
      if (op == ARCHIP_OP_FLUSH) {
          segments_nr = 1;
      } else {
          segments_nr = (int)(count / MAX_REQUEST_SIZE) + \
                        ((count % MAX_REQUEST_SIZE) ? 1 : 0);
      }
      segreq->total = count;
      atomic_mb_set(&segreq->ref, segments_nr);
  
      while (segments_nr > 1) {
          ret = archipelago_submit_request(s, pos,
                                              MAX_REQUEST_SIZE,
                                              offset + pos,
                                              aio_cb, segreq, op);
  
          if (ret < 0) {
              goto err_exit;
          }
          count -= MAX_REQUEST_SIZE;
          pos += MAX_REQUEST_SIZE;
          segments_nr--;
      }
      ret = archipelago_submit_request(s, pos, count, offset + pos,
                                       aio_cb, segreq, op);
  
      if (ret < 0) {
          goto err_exit;
      }
      return 0;
  
  err_exit:
      segreq->failed = 1;
      if (atomic_fetch_sub(&segreq->ref, segments_nr) == segments_nr) {
          g_free(segreq);
      }
      return ret;
  }
  
  static BlockAIOCB *qemu_archipelago_aio_rw(BlockDriverState *bs,
                                             int64_t sector_num,
                                             QEMUIOVector *qiov,
                                             int nb_sectors,
                                             BlockCompletionFunc *cb,
                                             void *opaque,
                                             int op)
  {
      ArchipelagoAIOCB *aio_cb;
      BDRVArchipelagoState *s = bs->opaque;
      int64_t size, off;
      int ret;
  
      aio_cb = qemu_aio_get(&archipelago_aiocb_info, bs, cb, opaque);
      aio_cb->cmd = op;
      aio_cb->qiov = qiov;
  
      aio_cb->ret = 0;
      aio_cb->s = s;
      aio_cb->status = -EINPROGRESS;
  
      off = sector_num * BDRV_SECTOR_SIZE;
      size = nb_sectors * BDRV_SECTOR_SIZE;
      aio_cb->size = size;
  
      ret = archipelago_aio_segmented_rw(s, size, off,
                                         aio_cb, op);
      if (ret < 0) {
          goto err_exit;
      }
      return &aio_cb->common;
  
  err_exit:
      error_report("qemu_archipelago_aio_rw(): I/O Error\n");
      qemu_aio_unref(aio_cb);
      return NULL;
  }
  
  static BlockAIOCB *qemu_archipelago_aio_readv(BlockDriverState *bs,
          int64_t sector_num, QEMUIOVector *qiov, int nb_sectors,
          BlockCompletionFunc *cb, void *opaque)
  {
      return qemu_archipelago_aio_rw(bs, sector_num, qiov, nb_sectors, cb,
                                     opaque, ARCHIP_OP_READ);
  }
  
  static BlockAIOCB *qemu_archipelago_aio_writev(BlockDriverState *bs,
          int64_t sector_num, QEMUIOVector *qiov, int nb_sectors,
          BlockCompletionFunc *cb, void *opaque)
  {
      return qemu_archipelago_aio_rw(bs, sector_num, qiov, nb_sectors, cb,
                                     opaque, ARCHIP_OP_WRITE);
  }
  
  static int64_t archipelago_volume_info(BDRVArchipelagoState *s)
  {
      uint64_t size;
      int ret, targetlen;
      struct xseg_request *req;
      struct xseg_reply_info *xinfo;
      AIORequestData *reqdata = g_new(AIORequestData, 1);
  
      const char *volname = s->volname;
      targetlen = strlen(volname);
      req = xseg_get_request(s->xseg, s->srcport, s->mportno, X_ALLOC);
      if (!req) {
          archipelagolog("Cannot get XSEG request\n");
          goto err_exit2;
      }
      ret = xseg_prep_request(s->xseg, req, targetlen,
                              sizeof(struct xseg_reply_info));
      if (ret < 0) {
          archipelagolog("Cannot prepare XSEG request\n");
          goto err_exit;
      }
      char *target = xseg_get_target(s->xseg, req);
      if (!target) {
          archipelagolog("Cannot get XSEG target\n");
          goto err_exit;
      }
      memcpy(target, volname, targetlen);
      req->size = req->datalen;
      req->offset = 0;
      req->op = X_INFO;
  
      reqdata->op = ARCHIP_OP_VOLINFO;
      reqdata->volname = volname;
      xseg_set_req_data(s->xseg, req, reqdata);
  
      xport p = xseg_submit(s->xseg, req, s->srcport, X_ALLOC);
      if (p == NoPort) {
          archipelagolog("Cannot submit XSEG request\n");
          goto err_exit;
      }
      xseg_signal(s->xseg, p);
      qemu_mutex_lock(&s->archip_mutex);
      while (!s->is_signaled) {
          qemu_cond_wait(&s->archip_cond, &s->archip_mutex);
      }
      s->is_signaled = false;
      qemu_mutex_unlock(&s->archip_mutex);
  
      xinfo = (struct xseg_reply_info *) xseg_get_data(s->xseg, req);
      size = xinfo->size;
      xseg_put_request(s->xseg, req, s->srcport);
      g_free(reqdata);
      s->size = size;
      return size;
  
  err_exit:
      xseg_put_request(s->xseg, req, s->srcport);
  err_exit2:
      g_free(reqdata);
      return -EIO;
  }
  
  static int64_t qemu_archipelago_getlength(BlockDriverState *bs)
  {
      int64_t ret;
      BDRVArchipelagoState *s = bs->opaque;
  
      ret = archipelago_volume_info(s);
      return ret;
  }
  
  static int qemu_archipelago_truncate(BlockDriverState *bs, int64_t offset)
  {
      int ret, targetlen;
      struct xseg_request *req;
      BDRVArchipelagoState *s = bs->opaque;
      AIORequestData *reqdata = g_new(AIORequestData, 1);
  
      const char *volname = s->volname;
      targetlen = strlen(volname);
      req = xseg_get_request(s->xseg, s->srcport, s->mportno, X_ALLOC);
      if (!req) {
          archipelagolog("Cannot get XSEG request\n");
          goto err_exit2;
      }
  
      ret = xseg_prep_request(s->xseg, req, targetlen, 0);
      if (ret < 0) {
          archipelagolog("Cannot prepare XSEG request\n");
          goto err_exit;
      }
      char *target = xseg_get_target(s->xseg, req);
      if (!target) {
          archipelagolog("Cannot get XSEG target\n");
          goto err_exit;
      }
      memcpy(target, volname, targetlen);
      req->offset = offset;
      req->op = X_TRUNCATE;
  
      reqdata->op = ARCHIP_OP_TRUNCATE;
      reqdata->volname = volname;
  
      xseg_set_req_data(s->xseg, req, reqdata);
  
      xport p = xseg_submit(s->xseg, req, s->srcport, X_ALLOC);
      if (p == NoPort) {
          archipelagolog("Cannot submit XSEG request\n");
          goto err_exit;
      }
  
      xseg_signal(s->xseg, p);
      qemu_mutex_lock(&s->archip_mutex);
      while (!s->is_signaled) {
          qemu_cond_wait(&s->archip_cond, &s->archip_mutex);
      }
      s->is_signaled = false;
      qemu_mutex_unlock(&s->archip_mutex);
      xseg_put_request(s->xseg, req, s->srcport);
      g_free(reqdata);
      return 0;
  
  err_exit:
      xseg_put_request(s->xseg, req, s->srcport);
  err_exit2:
      g_free(reqdata);
      return -EIO;
  }
  
  static QemuOptsList qemu_archipelago_create_opts = {
      .name = "archipelago-create-opts",
      .head = QTAILQ_HEAD_INITIALIZER(qemu_archipelago_create_opts.head),
      .desc = {
          {
              .name = BLOCK_OPT_SIZE,
              .type = QEMU_OPT_SIZE,
              .help = "Virtual disk size"
          },
          { /* end of list */ }
      }
  };
  
  static BlockAIOCB *qemu_archipelago_aio_flush(BlockDriverState *bs,
          BlockCompletionFunc *cb, void *opaque)
  {
      return qemu_archipelago_aio_rw(bs, 0, NULL, 0, cb, opaque,
                                     ARCHIP_OP_FLUSH);
  }
  
  static BlockDriver bdrv_archipelago = {
      .format_name         = "archipelago",
      .protocol_name       = "archipelago",
      .instance_size       = sizeof(BDRVArchipelagoState),
      .bdrv_parse_filename = archipelago_parse_filename,
      .bdrv_file_open      = qemu_archipelago_open,
      .bdrv_close          = qemu_archipelago_close,
      .bdrv_create         = qemu_archipelago_create,
      .bdrv_getlength      = qemu_archipelago_getlength,
      .bdrv_truncate       = qemu_archipelago_truncate,
      .bdrv_aio_readv      = qemu_archipelago_aio_readv,
      .bdrv_aio_writev     = qemu_archipelago_aio_writev,
      .bdrv_aio_flush      = qemu_archipelago_aio_flush,
      .bdrv_has_zero_init  = bdrv_has_zero_init_1,
      .create_opts         = &qemu_archipelago_create_opts,
  };
  
  static void bdrv_archipelago_init(void)
  {
      bdrv_register(&bdrv_archipelago);
  }
  
  block_init(bdrv_archipelago_init);
  