Re: [Qemu-devel] [PATCH 1/1] block: Support Archipelago as a QEMU block backend

[Kevin Wolf]

Am 29.05.2014 um 13:14 hat Chrysostomos Nanakos geschrieben:
> VM Image on Archipelago volume is specified like this:
> 
> file=archipelago:<volumename>[/mport=<mapperd_port>[:vport=<vlmcd_port>]]
> 
> '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.
> 
> Examples:
> 
> file=archipelago:my_vm_volume
> file=archipelago:my_vm_volume/mport=123
> file=archipelago:my_vm_volume/mport=123:vport=1234
> 
> Signed-off-by: Chrysostomos Nanakos <address@hidden>

The command line interface you mention here looks good as a convenience
interface, but the main interface should be separate options, i.e.
something like file.driver=archipelago,file.volume=...,file.mport=...

You can implement .bdrv_parse_filename() to convert the shortcut version
with a single string to the individual options.

>  block/Makefile.objs |    1 +
>  block/archipelago.c | 1129 
> +++++++++++++++++++++++++++++++++++++++++++++++++++
>  configure           |   40 ++
>  3 files changed, 1170 insertions(+)
>  create mode 100644 block/archipelago.c
> 
> diff --git a/block/Makefile.objs b/block/Makefile.objs
> index fd88c03..895c30d 100644
> --- a/block/Makefile.objs
> +++ b/block/Makefile.objs
> @@ -17,6 +17,7 @@ block-obj-$(CONFIG_LIBNFS) += nfs.o
>  block-obj-$(CONFIG_CURL) += curl.o
>  block-obj-$(CONFIG_RBD) += rbd.o
>  block-obj-$(CONFIG_GLUSTERFS) += gluster.o
> +block-obj-$(CONFIG_ARCHIPELAGO) += archipelago.o
>  block-obj-$(CONFIG_LIBSSH2) += ssh.o
>  endif
>  
> diff --git a/block/archipelago.c b/block/archipelago.c
> new file mode 100644
> index 0000000..5318cc8
> --- /dev/null
> +++ b/block/archipelago.c
> @@ -0,0 +1,1129 @@
> +/*
> + * Copyright 2014 GRNET S.A. All rights reserved.
> + *
> + * Redistribution and use in source and binary forms, with or
> + * without modification, are permitted provided that the following
> + * conditions are met:
> + *
> + *   1. Redistributions of source code must retain the above
> + *      copyright notice, this list of conditions and the following
> + *      disclaimer.
> + *   2. Redistributions in binary form must reproduce the above
> + *      copyright notice, this list of conditions and the following
> + *      disclaimer in the documentation and/or other materials
> + *      provided with the distribution.
> + *
> + * THIS SOFTWARE IS PROVIDED BY GRNET S.A. ``AS IS'' AND ANY EXPRESS
> + * OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE IMPLIED
> + * WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR
> + * PURPOSE ARE DISCLAIMED. IN NO EVENT SHALL GRNET S.A OR
> + * CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL,
> + * SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT
> + * LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF
> + * USE, DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED
> + * AND ON ANY THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT
> + * LIABILITY, OR TORT (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN
> + * ANY WAY OUT OF THE USE OF THIS SOFTWARE, EVEN IF ADVISED OF THE
> + * POSSIBILITY OF SUCH DAMAGE.
> + *
> + * The views and conclusions contained in the software and
> + * documentation are those of the authors and should not be
> + * interpreted as representing official policies, either expressed
> + * or implied, of GRNET S.A.
> + */
> +
> +#include "block/block_int.h"
> +#include "qemu/error-report.h"
> +#include "qemu/thread.h"
> +
> +#include <inttypes.h>
> +#include <xseg/xseg.h>
> +#include <xseg/protocol.h>
> +
> +#define ARCHIP_FD_READ      0
> +#define ARCHIP_FD_WRITE     1
> +#define NUM_XSEG_THREADS    1
> +#define MAX_REQUEST_SIZE    524288
> +
> +static struct xseg *xseg;
> +static struct xseg_port *port;
> +xport srcport = NoPort;
> +xport sport = NoPort;
> +xport mportno = NoPort;
> +xport vportno = NoPort;

You can't use global variables for per-device state. When you try to use
a second Archipelago block device, things would break this way.

> +#define archipelagolog(fmt, ...) \
> +    fprintf(stderr, "archipelago\t%-24s: " fmt, __func__, ##__VA_ARGS__)

Wrap this in do { ... } while(0)

> +
> +typedef enum {
> +    ARCHIP_OP_READ,
> +    ARCHIP_OP_WRITE,
> +    ARCHIP_OP_FLUSH,
> +    ARCHIP_OP_VOLINFO,
> +} ARCHIPCmd;
> +
> +typedef struct ArchipelagoConf {
> +    char *volname;
> +    int64_t size;
> +} ArchipelagoConf;

I can see two users of this struct: The first is BDRVArchipelagoState,
which is used in the bdrv_open() path, and the second one is for image
creation. Only image creation makes use of the size field.

Is it really worth keeping this struct instead of moving the fields
directly into BDRVArchipelagoState and letting .bdrv_create() pass the
two fields as individual arguments?

Also, why is size signed?

> +typedef struct ArchipelagoAIOCB {
> +    BlockDriverAIOCB common;
> +    QEMUBH *bh;
> +    int64_t ret;
> +    QEMUIOVector *qiov;
> +    char *buffer;

Why char* instead of void* when you never dereference it?

> +    ARCHIPCmd cmd;
> +    int64_t sector_num;
> +    int error;
> +    struct BDRVArchipelagoState *s;
> +    int cancelled;

bool?

> +    int status;

This is only ever -EINPROGRESS or 0. Would 'bool completed' be clearer?

> +} ArchipelagoAIOCB;
> +
> +typedef struct ArchipelagoCB {
> +    ArchipelagoAIOCB *acb;
> +    struct BDRVArchipelagoState *s;
> +    int done;

Looks unused (initialised to 0, but never read)

> +    int64_t size;
> +    char *buf;
> +    int64_t ret;
> +} ArchipelagoCB;

What is the difference between ArchipelagoAIOCB and ArchipelagoCB that
explains why they are separate structs?

Also, why do you keep a reference to the acb here, but duplicate fields
like size and buf in multiple structs and pass them as additional
parameters to functions, too?

> +typedef struct BDRVArchipelagoState {
> +    int fds[2];
> +    ArchipelagoConf *gconf;
> +    int qemu_aio_count;
> +    int event_reader_pos;
> +    ArchipelagoCB *event_acb;
> +} BDRVArchipelagoState;
> +
> +typedef struct ArchipelagoSegmentedRequest {
> +    size_t count;
> +    size_t total;
> +    int ref;
> +    int failed;
> +} ArchipelagoSegmentedRequest;
> +
> +typedef struct AIORequestData {
> +    char *volname;
> +    off_t offset;
> +    size_t size;
> +    char *buf;
> +    int ret;
> +    int op;
> +    ArchipelagoCB *aio_cb;
> +    ArchipelagoSegmentedRequest *segreq;
> +} AIORequestData;
> +
> +
> +typedef struct ArchipelagoThread {
> +    QemuThread request_th;
> +    QemuCond request_cond;
> +    QemuMutex request_mutex;
> +    int is_signaled;
> +    int is_running;

bool

> +} ArchipelagoThread;
> +
> +static void archipelago_aio_bh_cb(void *opaque);
> +
> +static int qemu_archipelago_signal_pipe(BDRVArchipelagoState *s,
> +                                        ArchipelagoCB *aio_cb);
> +
> +QemuMutex archip_mutex;
> +QemuCond archip_cond;
> +ArchipelagoThread archipelago_th[NUM_XSEG_THREADS];
> +static int is_signaled;

Shouldn't these be per-device rather than global?

> +static void init_local_signal(void)
> +{
> +    if (xseg && (sport != srcport)) {
> +        xseg_init_local_signal(xseg, srcport);
> +        sport = srcport;
> +    }
> +}
> +
> +static void archipelago_finish_aiocb(ArchipelagoCB *aio_cb,
> +                                     ssize_t c,
> +                                     AIORequestData *reqdata)
> +{
> +    int ret;
> +    aio_cb->ret = c;
> +    ret = qemu_archipelago_signal_pipe(aio_cb->s, aio_cb);
> +    if (ret < 0) {
> +        error_report("archipelago_finish_aiocb(): failed writing 
> acb->s->fds");
> +        g_free(aio_cb);
> +        /* Lock disk and exit ??*/
> +    }
> +    g_free(reqdata);
> +}
> +
> +static int wait_reply(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, 0);
> +        if (req) {
> +            if (req != expected_req) {
> +                archipelagolog("Unknown received request.\n");
> +                xseg_put_request(xseg, req, srcport);
> +            } else if (!(req->state & XS_SERVED)) {
> +                archipelagolog("Failed req.\n");
> +                return -1;
> +            } else {
> +                break;
> +            }
> +        }
> +        xseg_wait_signal(xseg, psd, 10000000UL);
> +    }
> +    xseg_cancel_wait(xseg, srcport);
> +    return 0;
> +}
> +
> +static void xseg_request_handler(void *arthd)
> +{
> +
> +    void *psd = xseg_get_signal_desc(xseg, port);
> +    ArchipelagoThread *th = (ArchipelagoThread *) arthd;
> +    while (th->is_running) {
> +        struct xseg_request *req;
> +        xseg_prepare_wait(xseg, srcport);
> +        req = xseg_receive(xseg, srcport, 0);
> +        if (req) {
> +            AIORequestData *reqdata;
> +            ArchipelagoSegmentedRequest *segreq;
> +            xseg_get_req_data(xseg, req, (void **)&reqdata);
> +
> +            if (!(req->state & XS_SERVED)) {
> +                    segreq = reqdata->segreq;
> +                    __sync_bool_compare_and_swap(&segreq->failed, 0, 1);
> +            }
> +
> +            if (reqdata->op == ARCHIP_OP_READ) {

Could be a switch statement.

> +                char *data = xseg_get_data(xseg, req);
> +                segreq = reqdata->segreq;
> +                segreq->count += req->serviced;
> +
> +                memcpy(reqdata->buf, data, req->serviced);

So we're copying data around twice? Why don't you have the qiov here so
that you can directly copy into the real destination memory?

> +                xseg_put_request(xseg, req, srcport);
> +
> +                __sync_add_and_fetch(&segreq->ref, -1);
> +
> +                if (!segreq->ref && !segreq->failed) {
> +                    reqdata->ret = segreq->count;
> +                    g_free(reqdata->segreq);
> +                    archipelago_finish_aiocb(reqdata->aio_cb, reqdata->ret,
> +                                             reqdata);
> +                } else if (segreq->ref && segreq->failed) {
> +                    g_free(reqdata);
> +                } else if (!segreq->ref && segreq->failed) {
> +                    g_free(reqdata->segreq);
> +                    g_free(reqdata);
> +                }

You're trying to handle multiple things at once here. I think this looks
a bit more readable:

if (seqreq->ref == 0) {
    if (!failed) {
        reqdata->ret = segreq->count;
        archipelago_finish_aiocb(reqdata->aio_cb, reqdata->ret,
                                 reqdata);
    }
    g_free(segreq);
}
if (seqreq->failed) {
    g_free(reqdata);
}

> +            } else if (reqdata->op == ARCHIP_OP_WRITE) {
> +                segreq = reqdata->segreq;
> +                reqdata->ret = req->serviced;
> +                segreq->count += req->serviced;
> +                xseg_put_request(xseg, req, srcport);
> +
> +                __sync_add_and_fetch(&segreq->ref, -1);
> +
> +                if (!segreq->ref && !segreq->failed) {
> +                    reqdata->ret = segreq->count;
> +                    g_free(reqdata->segreq);
> +                    archipelago_finish_aiocb(reqdata->aio_cb, reqdata->ret,
> +                                             reqdata);
> +                } else if (segreq->ref && segreq->failed) {
> +                    g_free(reqdata);
> +                } else if (!segreq->ref && segreq->failed) {
> +                    g_free(reqdata->segreq);
> +                    g_free(reqdata);
> +                }

Same as above.

> +            } else if (reqdata->op == ARCHIP_OP_VOLINFO) {
> +                is_signaled = 1;
> +                qemu_cond_signal(&archip_cond);
> +            }
> +        } else {
> +            xseg_wait_signal(xseg, psd, 10000000UL);
> +        }
> +        xseg_cancel_wait(xseg, srcport);
> +    }
> +    th->is_signaled = 1;
> +    qemu_cond_signal(&th->request_cond);
> +    qemu_thread_exit(NULL);
> +}
> +
> +static void qemu_archipelago_gconf_free(ArchipelagoConf *gconf)
> +{
> +    g_free(gconf->volname);
> +    g_free(gconf);
> +}
> +
> +static void xseg_find_port(char *pstr, const char *needle, xport *port)
> +{
> +    char *a;
> +    char *dpstr = strdup(pstr);
> +    a = strtok(dpstr, needle);
> +    *port = (xport) atoi(a);
> +    free(dpstr);
> +}

g_strdup/g_free

But it looks like duplicating the string weren't even necessary if you
just searched in the existing string without modifying it.

> +
> +static int parse_volume_options(ArchipelagoConf *gconf, char *path)

const char *path

> +{
> +    char *tokens[4];
> +    int i;
> +    if (!path) {
> +        return -EINVAL;
> +    }
> +    /* Find Volume Name, mapperd and vlmcd ports */
> +    char *ds = g_strndup(path, strlen(path));

If you need the full length, use g_strdup()

> +    tokens[0] = strtok(ds, ":");
> +    tokens[1] = strtok(NULL, "/");
> +    tokens[2] = strtok(NULL, ":");
> +    tokens[3] = strtok(NULL, ":");
> +    if (strcmp(tokens[0], "archipelago") != 0) {
> +        /* Should not be here. Protocol is already not supported */
> +        return -EINVAL;
> +    }
> +
> +    gconf->volname = g_strndup(tokens[1], strlen(tokens[1]));

Isn't tokens[1] == NULL when there is no colon in path?

> +    for (i = 0; i < 4; i++) {
> +        if (tokens[i] != NULL) {
> +            if (strstr(tokens[i], "mport=")) {
> +                xseg_find_port(tokens[i], "mport=", &mportno);
> +            }
> +            if (strstr(tokens[i], "vport=")) {
> +                xseg_find_port(tokens[i], "vport=", &vportno);
> +            }

Use the return value of strstr() + strlen("mport=") and do a strtoul()
from there.

> +        }
> +    }
> +
> +    return 0;
> +}
> +
> +static int archipelago_parse_uri(ArchipelagoConf *gconf, const char 
> *filename)
> +{
> +    return parse_volume_options(gconf, (char *)filename);
> +}

Not a very useful function. And casting away const is bad, too.

> +static int qemu_archipelago_xseg_init(void)
> +{
> +    if (xseg_initialize()) {
> +        archipelagolog("Cannot initialize xseg.\n");
> +        goto err_exit;
> +    }
> +    xseg = xseg_join((char *)"posix", (char *)"archipelago",
> +                     (char *)"posixfd", NULL);
> +    if (!xseg) {
> +        archipelagolog("Cannot join segment.\n");
> +        goto err_exit;
> +    }
> +    port = xseg_bind_dynport(xseg);
> +    srcport = port->portno;
> +    init_local_signal();
> +    return 0;
> +err_exit:
> +    return -1;

This is used as -errno later, so you should return something better than
-1 here (it ends up as -EPERM on Linux).

> +}
> +
> +static int qemu_archipelago_init(ArchipelagoConf *gconf, const char 
> *filename)
> +{
> +    int ret, i;
> +    /* Set default values */
> +    vportno = 501;
> +    mportno = 1001;
> +
> +    ret = archipelago_parse_uri(gconf, filename);
> +    if (ret < 0) {
> +        const char *err_msg =
> +            "<volumename>[/mport=<mapperd_port>[:vport=<vlmcd_port>]]";
> +        error_report("Usage: file=archipelago:%s", err_msg);

Use the Error* API. Also this local variable is pretty useless. You can
split strings into multiple lines in C:

error_report("Usage: file=archipelago:"
             "<volumename>[/mport=<mapperd_port>"
             "[:vport=<vlmcd_port>]]");


> +        errno = -ret;
> +        goto err_exit;
> +    }
> +
> +    ret = qemu_archipelago_xseg_init();
> +    if (ret < 0) {
> +        error_report("Cannot initialize xseg. Aborting...\n");
> +        errno = -ret;
> +        goto err_exit;
> +    }
> +
> +    qemu_cond_init(&archip_cond);
> +    qemu_mutex_init(&archip_mutex);
> +    for (i = 0; i < NUM_XSEG_THREADS; i++) {
> +        qemu_cond_init(&archipelago_th[i].request_cond);
> +        qemu_mutex_init(&archipelago_th[i].request_mutex);
> +        archipelago_th[i].is_signaled = 0;
> +        archipelago_th[i].is_running = 1;
> +        qemu_thread_create(&archipelago_th[i].request_th, "xseg_io_th",
> +                           (void *) xseg_request_handler,
> +                           (void *) &archipelago_th[i], 
> QEMU_THREAD_DETACHED);
> +    }
> +
> +err_exit:
> +    return ret;
> +}
> +
> +static void qemu_archipelago_complete_aio(ArchipelagoCB *aio_cb)
> +{
> +    int64_t r;
> +    ArchipelagoAIOCB *acb = aio_cb->acb;
> +
> +    r = aio_cb->ret;
> +
> +    if (acb->cmd != ARCHIP_OP_READ) {
> +        if (r < 0) {
> +            acb->ret = r;
> +            acb->error = 1;
> +        } else if (!acb->error) {
> +            acb->ret = aio_cb->size;

In the read case you set acb->ret = r. What is the difference here?

> +        }
> +    } else {
> +        if (r < 0) {
> +            memset(aio_cb->buf, 0, aio_cb->size);

Why that? archipelago_aio_bh_cb() should simply check for acb->error
before it calls qemu_iovec_from_buf().

> +            acb->ret = r;
> +            acb->error = 1;
> +        } else if (r < aio_cb->size) {
> +            memset(aio_cb->buf + r, 0, aio_cb->size - r);
> +            if (!acb->error) {
> +                acb->ret = aio_cb->size;
> +            }

How would acb->error become true here? It is only set by the other if
branches in this function.

More importantly, why is this a success case? The caller would expect
the full request to be completed when archipelago_aio_bh_cb() calls its
callback with ret = 0, but you're really delivering a buffer that is
partly filled with zeros instead of the real data.

> +        } else if (!acb->error) {
> +            acb->ret = r;
> +        }
> +    }
> +
> +    acb->bh = qemu_bh_new(archipelago_aio_bh_cb, acb);
> +    qemu_bh_schedule(acb->bh);
> +    g_free(aio_cb);
> +}
> +
> +static void qemu_archipelago_aio_event_reader(void *opaque)
> +{
> +    BDRVArchipelagoState *s = opaque;
> +    ssize_t ret;
> +
> +    do {
> +        char *p = (char *)&s->event_acb;
> +
> +        ret = read(s->fds[ARCHIP_FD_READ], p + s->event_reader_pos,
> +                   sizeof(s->event_acb) - s->event_reader_pos);
> +        if (ret > 0) {
> +            s->event_reader_pos += ret;
> +            if (s->event_reader_pos == sizeof(s->event_acb)) {
> +                s->event_reader_pos = 0;
> +                qemu_archipelago_complete_aio(s->event_acb);
> +                s->qemu_aio_count--;
> +            }
> +        }
> +    } while (ret < 0 && errno == EINTR);
> +}
> +
> +static QemuOptsList runtime_opts = {
> +    .name = "archipelago",
> +    .head = QTAILQ_HEAD_INITIALIZER(runtime_opts.head),
> +    .desc = {
> +        {
> +            .name = "filename",
> +            .type = QEMU_OPT_STRING,
> +            .help = "Specification of the volume image",
> +        },
> +        { /* end of list */ }
> +    },
> +};

This is where you need to split the options into volume, mport, vport
etc.

> +static int qemu_archipelago_open(BlockDriverState *bs,
> +                                 QDict *options,
> +                                 int bdrv_flags,
> +                                 Error **errp)
> +{
> +    int ret = 0;
> +    QemuOpts *opts;
> +    Error *local_err = NULL;
> +    const char *filename;
> +    BDRVArchipelagoState *s = bs->opaque;
> +
> +    opts = qemu_opts_create(&runtime_opts, NULL, 0, &error_abort);
> +    qemu_opts_absorb_qdict(opts, options, &local_err);
> +    if (local_err) {
> +        qerror_report_err(local_err);
> +        error_free(local_err);

No, you need to error_propagate(errp, local_err); to pass the error up
to the caller instead of reporting and freeing it here.

> +        ret = -EINVAL;
> +        goto err_exit;
> +    }
> +
> +    filename = qemu_opt_get(opts, "filename");
> +
> +    /* Initialize XSEG, join segment and set s->gconf->volname */
> +    s->gconf = g_malloc0(sizeof(ArchipelagoConf));
> +    ret = qemu_archipelago_init(s->gconf, filename);
> +    if (ret < 0) {
> +        ret = -errno;

Don't mess with errno here, ret is already correct.
qemu_archipelago_init() doesn't need to set errno either.

Also, try to make sure to set errp whenever you return failure from this
function.

> +        goto err_exit;
> +    }
> +
> +    s->event_reader_pos = 0;
> +    ret = qemu_pipe(s->fds);
> +    if (ret < 0) {
> +        ret = -errno;
> +        goto err_exit;
> +    }
> +
> +    fcntl(s->fds[ARCHIP_FD_READ], F_SETFL, O_NONBLOCK);
> +    fcntl(s->fds[ARCHIP_FD_WRITE], F_SETFL, O_NONBLOCK);
> +    qemu_aio_set_fd_handler(s->fds[ARCHIP_FD_READ],
> +                            qemu_archipelago_aio_event_reader, NULL,
> +                            s);
> +
> +    qemu_opts_del(opts);
> +    return 0;
> +
> +err_exit:
> +    qemu_opts_del(opts);
> +    qemu_archipelago_gconf_free(s->gconf);
> +    return ret;
> +}
> +
[...]
> +
> +static int qemu_archipelago_truncate(BlockDriverState *bs, int64_t offset)
> +{
> +    /* Not implemented yet */
> +    return 0;
> +}

Then remove the function or at the very least return an error. Returning
success while doing nothing is wrong.

[...]
> +static QEMUOptionParameter qemu_archipelago_create_options[] = {
> +    {
> +        .name = BLOCK_OPT_SIZE,
> +        .type = OPT_SIZE,
> +        .help = "Virtual disk size"
> +    },
> +    {NULL}
> +};
> +
> +static BlockDriverAIOCB *qemu_archipelago_aio_flush(BlockDriverState *bs,
> +        BlockDriverCompletionFunc *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_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_options = qemu_archipelago_create_options,
> +};

Please align the = on the same column.

Sorry, I've run out of steam and didn't actually review the main
read/write logic. This will be an important part for the next review,
especially with respect to the threading. But I think this gives you
already a few things to work on.

Kevin