Merge remote-tracking branch 'remotes/kevin/tags/for-upstream' into staging

Block layer patches:

- Add block export infrastructure
- iotests improvements
- Document the throttle block filter
- Misc code cleanups

# gpg: Signature made Fri 02 Oct 2020 15:36:53 BST
# gpg:                using RSA key DC3DEB159A9AF95D3D7456FE7F09B272C88F2FD6
# gpg:                issuer "kwolf@redhat.com"
# gpg: Good signature from "Kevin Wolf <kwolf@redhat.com>" [full]
# Primary key fingerprint: DC3D EB15 9A9A F95D 3D74  56FE 7F09 B272 C88F 2FD6

* remotes/kevin/tags/for-upstream: (37 commits)
  qcow2: Use L1E_SIZE in qcow2_write_l1_entry()
  qemu-storage-daemon: Fix help line for --export
  iotests: Test block-export-* QMP interface
  iotests: Allow supported and unsupported formats at the same time
  iotests: Introduce qemu_nbd_list_log()
  iotests: Factor out qemu_tool_pipe_and_status()
  nbd: Deprecate nbd-server-add/remove
  nbd: Merge nbd_export_new() and nbd_export_create()
  block/export: Move writable to BlockExportOptions
  block/export: Add query-block-exports
  block/export: Create BlockBackend in blk_exp_add()
  block/export: Move blk to BlockExport
  block/export: Add BLOCK_EXPORT_DELETED event
  block/export: Add block-export-del
  block/export: Move strong user reference to block_exports
  block/export: Add 'id' option to block-export-add
  block/export: Add blk_exp_close_all(_type)
  block/export: Allocate BlockExport in blk_exp_add()
  block/export: Add node-name to BlockExportOptions
  block/export: Move AioContext from NBDExport to BlockExport
  ...

Signed-off-by: Peter Maydell <peter.maydell@linaro.org>

30 files changed, 1428 insertions, 545 deletions

diff --git a/block.c b/block.c
index f72a2e26e8..f4b6dd5d3d 100644
--- a/block.c
+++ b/block.c
@@ -4462,7 +4462,7 @@ static void bdrv_close(BlockDriverState *bs)
 void bdrv_close_all(void)
 {
     assert(job_next(NULL) == NULL);
-    nbd_export_close_all();
+    blk_exp_close_all();
 
     /* Drop references from requests still in flight, such as canceled block
      * jobs whose AIO context has not been polled yet */
diff --git a/block/export/export.c b/block/export/export.c
new file mode 100644
index 0000000000..f2c00d13bf
--- /dev/null
+++ b/block/export/export.c
@@ -0,0 +1,325 @@
+/*
+ * Common block export infrastructure
+ *
+ * Copyright (c) 2012, 2020 Red Hat, Inc.
+ *
+ * Authors:
+ * Paolo Bonzini <pbonzini@redhat.com>
+ * Kevin Wolf <kwolf@redhat.com>
+ *
+ * This work is licensed under the terms of the GNU GPL, version 2 or
+ * later.  See the COPYING file in the top-level directory.
+ */
+
+#include "qemu/osdep.h"
+
+#include "block/block.h"
+#include "sysemu/block-backend.h"
+#include "block/export.h"
+#include "block/nbd.h"
+#include "qapi/error.h"
+#include "qapi/qapi-commands-block-export.h"
+#include "qapi/qapi-events-block-export.h"
+#include "qemu/id.h"
+
+static const BlockExportDriver *blk_exp_drivers[] = {
+    &blk_exp_nbd,
+};
+
+/* Only accessed from the main thread */
+static QLIST_HEAD(, BlockExport) block_exports =
+    QLIST_HEAD_INITIALIZER(block_exports);
+
+BlockExport *blk_exp_find(const char *id)
+{
+    BlockExport *exp;
+
+    QLIST_FOREACH(exp, &block_exports, next) {
+        if (strcmp(id, exp->id) == 0) {
+            return exp;
+        }
+    }
+
+    return NULL;
+}
+
+static const BlockExportDriver *blk_exp_find_driver(BlockExportType type)
+{
+    int i;
+
+    for (i = 0; i < ARRAY_SIZE(blk_exp_drivers); i++) {
+        if (blk_exp_drivers[i]->type == type) {
+            return blk_exp_drivers[i];
+        }
+    }
+    return NULL;
+}
+
+BlockExport *blk_exp_add(BlockExportOptions *export, Error **errp)
+{
+    const BlockExportDriver *drv;
+    BlockExport *exp = NULL;
+    BlockDriverState *bs;
+    BlockBackend *blk;
+    AioContext *ctx;
+    uint64_t perm;
+    int ret;
+
+    if (!id_wellformed(export->id)) {
+        error_setg(errp, "Invalid block export id");
+        return NULL;
+    }
+    if (blk_exp_find(export->id)) {
+        error_setg(errp, "Block export id '%s' is already in use", export->id);
+        return NULL;
+    }
+
+    drv = blk_exp_find_driver(export->type);
+    if (!drv) {
+        error_setg(errp, "No driver found for the requested export type");
+        return NULL;
+    }
+
+    bs = bdrv_lookup_bs(NULL, export->node_name, errp);
+    if (!bs) {
+        return NULL;
+    }
+
+    if (!export->has_writable) {
+        export->writable = false;
+    }
+    if (bdrv_is_read_only(bs) && export->writable) {
+        error_setg(errp, "Cannot export read-only node as writable");
+        return NULL;
+    }
+
+    ctx = bdrv_get_aio_context(bs);
+    aio_context_acquire(ctx);
+
+    /*
+     * Block exports are used for non-shared storage migration. Make sure
+     * that BDRV_O_INACTIVE is cleared and the image is ready for write
+     * access since the export could be available before migration handover.
+     * ctx was acquired in the caller.
+     */
+    bdrv_invalidate_cache(bs, NULL);
+
+    perm = BLK_PERM_CONSISTENT_READ;
+    if (export->writable) {
+        perm |= BLK_PERM_WRITE;
+    }
+
+    blk = blk_new(ctx, perm, BLK_PERM_ALL);
+    ret = blk_insert_bs(blk, bs, errp);
+    if (ret < 0) {
+        goto fail;
+    }
+
+    if (!export->has_writethrough) {
+        export->writethrough = false;
+    }
+    blk_set_enable_write_cache(blk, !export->writethrough);
+
+    assert(drv->instance_size >= sizeof(BlockExport));
+    exp = g_malloc0(drv->instance_size);
+    *exp = (BlockExport) {
+        .drv        = drv,
+        .refcount   = 1,
+        .user_owned = true,
+        .id         = g_strdup(export->id),
+        .ctx        = ctx,
+        .blk        = blk,
+    };
+
+    ret = drv->create(exp, export, errp);
+    if (ret < 0) {
+        goto fail;
+    }
+
+    assert(exp->blk != NULL);
+
+    QLIST_INSERT_HEAD(&block_exports, exp, next);
+
+    aio_context_release(ctx);
+    return exp;
+
+fail:
+    blk_unref(blk);
+    aio_context_release(ctx);
+    if (exp) {
+        g_free(exp->id);
+        g_free(exp);
+    }
+    return NULL;
+}
+
+/* Callers must hold exp->ctx lock */
+void blk_exp_ref(BlockExport *exp)
+{
+    assert(exp->refcount > 0);
+    exp->refcount++;
+}
+
+/* Runs in the main thread */
+static void blk_exp_delete_bh(void *opaque)
+{
+    BlockExport *exp = opaque;
+    AioContext *aio_context = exp->ctx;
+
+    aio_context_acquire(aio_context);
+
+    assert(exp->refcount == 0);
+    QLIST_REMOVE(exp, next);
+    exp->drv->delete(exp);
+    blk_unref(exp->blk);
+    qapi_event_send_block_export_deleted(exp->id);
+    g_free(exp->id);
+    g_free(exp);
+
+    aio_context_release(aio_context);
+}
+
+/* Callers must hold exp->ctx lock */
+void blk_exp_unref(BlockExport *exp)
+{
+    assert(exp->refcount > 0);
+    if (--exp->refcount == 0) {
+        /* Touch the block_exports list only in the main thread */
+        aio_bh_schedule_oneshot(qemu_get_aio_context(), blk_exp_delete_bh,
+                                exp);
+    }
+}
+
+/*
+ * Drops the user reference to the export and requests that all client
+ * connections and other internally held references start to shut down. When
+ * the function returns, there may still be active references while the export
+ * is in the process of shutting down.
+ *
+ * Acquires exp->ctx internally. Callers must *not* hold the lock.
+ */
+void blk_exp_request_shutdown(BlockExport *exp)
+{
+    AioContext *aio_context = exp->ctx;
+
+    aio_context_acquire(aio_context);
+
+    /*
+     * If the user doesn't own the export any more, it is already shutting
+     * down. We must not call .request_shutdown and decrease the refcount a
+     * second time.
+     */
+    if (!exp->user_owned) {
+        goto out;
+    }
+
+    exp->drv->request_shutdown(exp);
+
+    assert(exp->user_owned);
+    exp->user_owned = false;
+    blk_exp_unref(exp);
+
+out:
+    aio_context_release(aio_context);
+}
+
+/*
+ * Returns whether a block export of the given type exists.
+ * type == BLOCK_EXPORT_TYPE__MAX checks for an export of any type.
+ */
+static bool blk_exp_has_type(BlockExportType type)
+{
+    BlockExport *exp;
+
+    if (type == BLOCK_EXPORT_TYPE__MAX) {
+        return !QLIST_EMPTY(&block_exports);
+    }
+
+    QLIST_FOREACH(exp, &block_exports, next) {
+        if (exp->drv->type == type) {
+            return true;
+        }
+    }
+
+    return false;
+}
+
+/* type == BLOCK_EXPORT_TYPE__MAX for all types */
+void blk_exp_close_all_type(BlockExportType type)
+{
+    BlockExport *exp, *next;
+
+    assert(in_aio_context_home_thread(qemu_get_aio_context()));
+
+    QLIST_FOREACH_SAFE(exp, &block_exports, next, next) {
+        if (type != BLOCK_EXPORT_TYPE__MAX && exp->drv->type != type) {
+            continue;
+        }
+        blk_exp_request_shutdown(exp);
+    }
+
+    AIO_WAIT_WHILE(NULL, blk_exp_has_type(type));
+}
+
+void blk_exp_close_all(void)
+{
+    blk_exp_close_all_type(BLOCK_EXPORT_TYPE__MAX);
+}
+
+void qmp_block_export_add(BlockExportOptions *export, Error **errp)
+{
+    blk_exp_add(export, errp);
+}
+
+void qmp_block_export_del(const char *id,
+                          bool has_mode, BlockExportRemoveMode mode,
+                          Error **errp)
+{
+    ERRP_GUARD();
+    BlockExport *exp;
+
+    exp = blk_exp_find(id);
+    if (exp == NULL) {
+        error_setg(errp, "Export '%s' is not found", id);
+        return;
+    }
+    if (!exp->user_owned) {
+        error_setg(errp, "Export '%s' is already shutting down", id);
+        return;
+    }
+
+    if (!has_mode) {
+        mode = BLOCK_EXPORT_REMOVE_MODE_SAFE;
+    }
+    if (mode == BLOCK_EXPORT_REMOVE_MODE_SAFE && exp->refcount > 1) {
+        error_setg(errp, "export '%s' still in use", exp->id);
+        error_append_hint(errp, "Use mode='hard' to force client "
+                          "disconnect\n");
+        return;
+    }
+
+    blk_exp_request_shutdown(exp);
+}
+
+BlockExportInfoList *qmp_query_block_exports(Error **errp)
+{
+    BlockExportInfoList *head = NULL, **p_next = &head;
+    BlockExport *exp;
+
+    QLIST_FOREACH(exp, &block_exports, next) {
+        BlockExportInfoList *entry = g_new0(BlockExportInfoList, 1);
+        BlockExportInfo *info = g_new(BlockExportInfo, 1);
+        *info = (BlockExportInfo) {
+            .id             = g_strdup(exp->id),
+            .type           = exp->drv->type,
+            .node_name      = g_strdup(bdrv_get_node_name(blk_bs(exp->blk))),
+            .shutting_down  = !exp->user_owned,
+        };
+
+        entry->value = info;
+        *p_next = entry;
+        p_next = &entry->next;
+    }
+
+    return head;
+}
diff --git a/block/export/meson.build b/block/export/meson.build
new file mode 100644
index 0000000000..558ef35d38
--- /dev/null
+++ b/block/export/meson.build
@@ -0,0 +1 @@
+block_ss.add(files('export.c'))
diff --git a/block/meson.build b/block/meson.build
index a3e56b7cd1..0b38dc36f7 100644
--- a/block/meson.build
+++ b/block/meson.build
@@ -110,6 +110,8 @@ block_ss.add(module_block_h)
 block_ss.add(files('stream.c'))
 
 softmmu_ss.add(files('qapi-sysemu.c'))
+
+subdir('export')
 subdir('monitor')
 
 modules += {'block': block_modules}
diff --git a/block/monitor/block-hmp-cmds.c b/block/monitor/block-hmp-cmds.c
index 4d3db5ed3c..d15a2be827 100644
--- a/block/monitor/block-hmp-cmds.c
+++ b/block/monitor/block-hmp-cmds.c
@@ -40,6 +40,7 @@
 #include "sysemu/block-backend.h"
 #include "sysemu/blockdev.h"
 #include "qapi/qapi-commands-block.h"
+#include "qapi/qapi-commands-block-export.h"
 #include "qapi/qmp/qdict.h"
 #include "qapi/error.h"
 #include "qapi/qmp/qerror.h"
@@ -397,7 +398,7 @@ void hmp_nbd_server_start(Monitor *mon, const QDict *qdict)
     Error *local_err = NULL;
     BlockInfoList *block_list, *info;
     SocketAddress *addr;
-    BlockExportNbd export;
+    NbdServerAddOptions export;
 
     if (writable && !all) {
         error_setg(&local_err, "-w only valid together with -a");
@@ -410,7 +411,7 @@ void hmp_nbd_server_start(Monitor *mon, const QDict *qdict)
         goto exit;
     }
 
-    nbd_server_start(addr, NULL, NULL, &local_err);
+    nbd_server_start(addr, NULL, NULL, 0, &local_err);
     qapi_free_SocketAddress(addr);
     if (local_err != NULL) {
         goto exit;
@@ -430,7 +431,7 @@ void hmp_nbd_server_start(Monitor *mon, const QDict *qdict)
             continue;
         }
 
-        export = (BlockExportNbd) {
+        export = (NbdServerAddOptions) {
             .device         = info->value->device,
             .has_writable   = true,
             .writable       = writable,
@@ -457,7 +458,7 @@ void hmp_nbd_server_add(Monitor *mon, const QDict *qdict)
     bool writable = qdict_get_try_bool(qdict, "writable", false);
     Error *local_err = NULL;
 
-    BlockExportNbd export = {
+    NbdServerAddOptions export = {
         .device         = (char *) device,
         .has_name       = !!name,
         .name           = (char *) name,
@@ -475,8 +476,8 @@ void hmp_nbd_server_remove(Monitor *mon, const QDict *qdict)
     bool force = qdict_get_try_bool(qdict, "force", false);
     Error *err = NULL;
 
-    /* Rely on NBD_SERVER_REMOVE_MODE_SAFE being the default */
-    qmp_nbd_server_remove(name, force, NBD_SERVER_REMOVE_MODE_HARD, &err);
+    /* Rely on BLOCK_EXPORT_REMOVE_MODE_SAFE being the default */
+    qmp_nbd_server_remove(name, force, BLOCK_EXPORT_REMOVE_MODE_HARD, &err);
     hmp_handle_error(mon, err);
 }
 
diff --git a/block/qcow2-cluster.c b/block/qcow2-cluster.c
index 9acc6ce4ae..aa87d3e99b 100644
--- a/block/qcow2-cluster.c
+++ b/block/qcow2-cluster.c
@@ -240,14 +240,14 @@ int qcow2_write_l1_entry(BlockDriverState *bs, int l1_index)
     }
 
     ret = qcow2_pre_write_overlap_check(bs, QCOW2_OL_ACTIVE_L1,
-            s->l1_table_offset + 8 * l1_start_index, bufsize, false);
+            s->l1_table_offset + L1E_SIZE * l1_start_index, bufsize, false);
     if (ret < 0) {
         return ret;
     }
 
     BLKDBG_EVENT(bs->file, BLKDBG_L1_UPDATE);
     ret = bdrv_pwrite_sync(bs->file,
-                           s->l1_table_offset + 8 * l1_start_index,
+                           s->l1_table_offset + L1E_SIZE * l1_start_index,
                            buf, bufsize);
     if (ret < 0) {
         return ret;
diff --git a/block/sheepdog.c b/block/sheepdog.c
index 2f5c0eb376..25111d5a70 100644
--- a/block/sheepdog.c
+++ b/block/sheepdog.c
@@ -740,7 +740,7 @@ static coroutine_fn void reconnect_to_sdog(void *opaque)
         if (s->fd < 0) {
             trace_sheepdog_reconnect_to_sdog();
             error_report_err(local_err);
-            qemu_co_sleep_ns(QEMU_CLOCK_REALTIME, 1000000000ULL);
+            qemu_co_sleep_ns(QEMU_CLOCK_REALTIME, NANOSECONDS_PER_SECOND);
         }
     };
 
diff --git a/blockdev-nbd.c b/blockdev-nbd.c
index 1a95d89f00..8174023e5c 100644
--- a/blockdev-nbd.c
+++ b/blockdev-nbd.c
@@ -14,7 +14,7 @@
 #include "sysemu/block-backend.h"
 #include "hw/block/block.h"
 #include "qapi/error.h"
-#include "qapi/qapi-commands-block.h"
+#include "qapi/qapi-commands-block-export.h"
 #include "block/nbd.h"
 #include "io/channel-socket.h"
 #include "io/net-listener.h"
@@ -23,23 +23,52 @@ typedef struct NBDServerData {
     QIONetListener *listener;
     QCryptoTLSCreds *tlscreds;
     char *tlsauthz;
+    uint32_t max_connections;
+    uint32_t connections;
 } NBDServerData;
 
 static NBDServerData *nbd_server;
+static bool is_qemu_nbd;
+
+static void nbd_update_server_watch(NBDServerData *s);
+
+void nbd_server_is_qemu_nbd(bool value)
+{
+    is_qemu_nbd = value;
+}
+
+bool nbd_server_is_running(void)
+{
+    return nbd_server || is_qemu_nbd;
+}
 
 static void nbd_blockdev_client_closed(NBDClient *client, bool ignored)
 {
     nbd_client_put(client);
+    assert(nbd_server->connections > 0);
+    nbd_server->connections--;
+    nbd_update_server_watch(nbd_server);
 }
 
 static void nbd_accept(QIONetListener *listener, QIOChannelSocket *cioc,
                        gpointer opaque)
 {
+    nbd_server->connections++;
+    nbd_update_server_watch(nbd_server);
+
     qio_channel_set_name(QIO_CHANNEL(cioc), "nbd-server");
     nbd_client_new(cioc, nbd_server->tlscreds, nbd_server->tlsauthz,
                    nbd_blockdev_client_closed);
 }
 
+static void nbd_update_server_watch(NBDServerData *s)
+{
+    if (!s->max_connections || s->connections < s->max_connections) {
+        qio_net_listener_set_client_func(s->listener, nbd_accept, NULL, NULL);
+    } else {
+        qio_net_listener_set_client_func(s->listener, NULL, NULL, NULL);
+    }
+}
 
 static void nbd_server_free(NBDServerData *server)
 {
@@ -88,7 +117,8 @@ static QCryptoTLSCreds *nbd_get_tls_creds(const char *id, Error **errp)
 
 
 void nbd_server_start(SocketAddress *addr, const char *tls_creds,
-                      const char *tls_authz, Error **errp)
+                      const char *tls_authz, uint32_t max_connections,
+                      Error **errp)
 {
     if (nbd_server) {
         error_setg(errp, "NBD server already running");
@@ -96,6 +126,7 @@ void nbd_server_start(SocketAddress *addr, const char *tls_creds,
     }
 
     nbd_server = g_new0(NBDServerData, 1);
+    nbd_server->max_connections = max_connections;
     nbd_server->listener = qio_net_listener_new();
 
     qio_net_listener_set_name(nbd_server->listener,
@@ -120,10 +151,7 @@ void nbd_server_start(SocketAddress *addr, const char *tls_creds,
 
     nbd_server->tlsauthz = g_strdup(tls_authz);
 
-    qio_net_listener_set_client_func(nbd_server->listener,
-                                     nbd_accept,
-                                     NULL,
-                                     NULL);
+    nbd_update_server_watch(nbd_server);
 
     return;
 
@@ -134,117 +162,100 @@ void nbd_server_start(SocketAddress *addr, const char *tls_creds,
 
 void nbd_server_start_options(NbdServerOptions *arg, Error **errp)
 {
-    nbd_server_start(arg->addr, arg->tls_creds, arg->tls_authz, errp);
+    nbd_server_start(arg->addr, arg->tls_creds, arg->tls_authz,
+                     arg->max_connections, errp);
 }
 
 void qmp_nbd_server_start(SocketAddressLegacy *addr,
                           bool has_tls_creds, const char *tls_creds,
                           bool has_tls_authz, const char *tls_authz,
+                          bool has_max_connections, uint32_t max_connections,
                           Error **errp)
 {
     SocketAddress *addr_flat = socket_address_flatten(addr);
 
-    nbd_server_start(addr_flat, tls_creds, tls_authz, errp);
+    nbd_server_start(addr_flat, tls_creds, tls_authz, max_connections, errp);
     qapi_free_SocketAddress(addr_flat);
 }
 
-void qmp_nbd_server_add(BlockExportNbd *arg, Error **errp)
+void qmp_nbd_server_add(NbdServerAddOptions *arg, Error **errp)
 {
-    BlockDriverState *bs = NULL;
+    BlockExport *export;
+    BlockDriverState *bs;
     BlockBackend *on_eject_blk;
-    NBDExport *exp;
-    int64_t len;
-    AioContext *aio_context;
+    BlockExportOptions *export_opts;
 
-    if (!nbd_server) {
-        error_setg(errp, "NBD server not running");
+    bs = bdrv_lookup_bs(arg->device, arg->device, errp);
+    if (!bs) {
         return;
     }
 
+    /*
+     * block-export-add would default to the node-name, but we may have to use
+     * the device name as a default here for compatibility.
+     */
     if (!arg->has_name) {
         arg->name = arg->device;
     }
 
-    if (strlen(arg->name) > NBD_MAX_STRING_SIZE) {
-        error_setg(errp, "export name '%s' too long", arg->name);
-        return;
-    }
-
-    if (arg->description && strlen(arg->description) > NBD_MAX_STRING_SIZE) {
-        error_setg(errp, "description '%s' too long", arg->description);
-        return;
+    export_opts = g_new(BlockExportOptions, 1);
+    *export_opts = (BlockExportOptions) {
+        .type                   = BLOCK_EXPORT_TYPE_NBD,
+        .id                     = g_strdup(arg->name),
+        .node_name              = g_strdup(bdrv_get_node_name(bs)),
+        .has_writable           = arg->has_writable,
+        .writable               = arg->writable,
+        .u.nbd = {
+            .has_name           = true,
+            .name               = g_strdup(arg->name),
+            .has_description    = arg->has_description,
+            .description        = g_strdup(arg->description),
+            .has_bitmap         = arg->has_bitmap,
+            .bitmap             = g_strdup(arg->bitmap),
+        },
+    };
+
+    /*
+     * nbd-server-add doesn't complain when a read-only device should be
+     * exported as writable, but simply downgrades it. This is an error with
+     * block-export-add.
+     */
+    if (bdrv_is_read_only(bs)) {
+        export_opts->has_writable = true;
+        export_opts->writable = false;
     }
 
-    if (nbd_export_find(arg->name)) {
-        error_setg(errp, "NBD server already has export named '%s'", arg->name);
-        return;
+    export = blk_exp_add(export_opts, errp);
+    if (!export) {
+        goto fail;
     }
 
+    /*
+     * nbd-server-add removes the export when the named BlockBackend used for
+     * @device goes away.
+     */
     on_eject_blk = blk_by_name(arg->device);
-
-    bs = bdrv_lookup_bs(arg->device, arg->device, errp);
-    if (!bs) {
-        return;
-    }
-
-    aio_context = bdrv_get_aio_context(bs);
-    aio_context_acquire(aio_context);
-    len = bdrv_getlength(bs);
-    if (len < 0) {
-        error_setg_errno(errp, -len,
-                         "Failed to determine the NBD export's length");
-        goto out;
-    }
-
-    if (!arg->has_writable) {
-        arg->writable = false;
-    }
-    if (bdrv_is_read_only(bs)) {
-        arg->writable = false;
+    if (on_eject_blk) {
+        nbd_export_set_on_eject_blk(export, on_eject_blk);
     }
 
-    exp = nbd_export_new(bs, 0, len, arg->name, arg->description, arg->bitmap,
-                         !arg->writable, !arg->writable,
-                         NULL, false, on_eject_blk, errp);
-    if (!exp) {
-        goto out;
-    }
-
-    /* The list of named exports has a strong reference to this export now and
-     * our only way of accessing it is through nbd_export_find(), so we can drop
-     * the strong reference that is @exp. */
-    nbd_export_put(exp);
-
- out:
-    aio_context_release(aio_context);
+fail:
+    qapi_free_BlockExportOptions(export_opts);
 }
 
 void qmp_nbd_server_remove(const char *name,
-                           bool has_mode, NbdServerRemoveMode mode,
+                           bool has_mode, BlockExportRemoveMode mode,
                            Error **errp)
 {
-    NBDExport *exp;
-    AioContext *aio_context;
-
-    if (!nbd_server) {
-        error_setg(errp, "NBD server not running");
-        return;
-    }
+    BlockExport *exp;
 
-    exp = nbd_export_find(name);
-    if (exp == NULL) {
-        error_setg(errp, "Export '%s' is not found", name);
+    exp = blk_exp_find(name);
+    if (exp && exp->drv->type != BLOCK_EXPORT_TYPE_NBD) {
+        error_setg(errp, "Block export '%s' is not an NBD export", name);
         return;
     }
 
-    if (!has_mode) {
-        mode = NBD_SERVER_REMOVE_MODE_SAFE;
-    }
-
-    aio_context = nbd_export_aio_context(exp);
-    aio_context_acquire(aio_context);
-    nbd_export_remove(exp, mode, errp);
-    aio_context_release(aio_context);
+    qmp_block_export_del(name, has_mode, mode, errp);
 }
 
 void qmp_nbd_server_stop(Error **errp)
@@ -254,7 +265,7 @@ void qmp_nbd_server_stop(Error **errp)
         return;
     }
 
-    nbd_export_close_all();
+    blk_exp_close_all_type(BLOCK_EXPORT_TYPE_NBD);
 
     nbd_server_free(nbd_server);
     nbd_server = NULL;
diff --git a/docs/system/deprecated.rst b/docs/system/deprecated.rst
index da862201ba..5e8346f7bf 100644
--- a/docs/system/deprecated.rst
+++ b/docs/system/deprecated.rst
@@ -264,6 +264,12 @@ chardev client socket with ``wait`` option (since 4.0)
 Character devices creating sockets in client mode should not specify
 the 'wait' field, which is only applicable to sockets in server mode
 
+``nbd-server-add`` and ``nbd-server-remove`` (since 5.2)
+''''''''''''''''''''''''''''''''''''''''''''''''''''''''
+
+Use the more generic commands ``block-export-add`` and ``block-export-del``
+instead.
+
 Human Monitor Protocol (HMP) commands
 -------------------------------------
 
diff --git a/docs/throttle.txt b/docs/throttle.txt
index cd4e109d39..b5b78b7326 100644
--- a/docs/throttle.txt
+++ b/docs/throttle.txt
@@ -1,6 +1,6 @@
 The QEMU throttling infrastructure
 ==================================
-Copyright (C) 2016 Igalia, S.L.
+Copyright (C) 2016,2020 Igalia, S.L.
 Author: Alberto Garcia <berto@igalia.com>
 
 This work is licensed under the terms of the GNU GPL, version 2 or
@@ -253,3 +253,109 @@ up. After those 60 seconds the bucket will have leaked 60 x 100 =
 
 Also, due to the way the algorithm works, longer burst can be done at
 a lower I/O rate, e.g. 1000 IOPS during 120 seconds.
+
+
+The 'throttle' block filter
+---------------------------
+Since QEMU 2.11 it is possible to configure the I/O limits using a
+'throttle' block filter. This filter uses the exact same throttling
+infrastructure described above but can be used anywhere in the node
+graph, allowing for more flexibility.
+
+The user can create an arbitrary number of filters and each one of
+them must be assigned to a group that contains the actual I/O limits.
+Different filters can use the same group so the limits are shared as
+described earlier in "Applying I/O limits to groups of disks".
+
+A group can be created using the object-add QMP function:
+
+   { "execute": "object-add",
+     "arguments": {
+       "qom-type": "throttle-group",
+       "id": "group0",
+       "props": {
+         "limits" : {
+           "iops-total": 1000
+           "bps-write": 2097152
+         }
+       }
+     }
+   }
+
+throttle-group has a 'limits' property (of type ThrottleLimits as
+defined in qapi/block-core.json) which can be set on creation or later
+with 'qom-set'.
+
+A throttle-group can also be created with the -object command line
+option but at the moment there is no way to pass a 'limits' parameter
+that contains a ThrottleLimits structure. The solution is to set the
+individual values directly, like in this example:
+
+   -object throttle-group,id=group0,x-iops-total=1000,x-bps-write=2097152
+
+Note however that this is not a stable API (hence the 'x-' prefixes) and
+will disappear when -object gains support for structured options and
+enables use of 'limits'.
+
+Once we have a throttle-group we can use the throttle block filter,
+where the 'file' property must be set to the block device that we want
+to filter:
+
+   { "execute": "blockdev-add",
+     "arguments": {
+        "options":  {
+           "driver": "qcow2",
+           "node-name": "disk0",
+           "file": {
+              "driver": "file",
+              "filename": "/path/to/disk.qcow2"
+           }
+        }
+     }
+   }
+
+   { "execute": "blockdev-add",
+     "arguments": {
+        "driver": "throttle",
+        "node-name": "throttle0",
+        "throttle-group": "group0",
+        "file": "disk0"
+     }
+   }
+
+A similar setup can also be done with the command line, for example:
+
+   -drive driver=throttle,throttle-group=group0,
+          file.driver=qcow2,file.file.filename=/path/to/disk.qcow2
+
+The scenario described so far is very simple but the throttle block
+filter allows for more complex configurations. For example, let's say
+that we have three different drives and we want to set I/O limits for
+each one of them and an additional set of limits for the combined I/O
+of all three drives.
+
+First we would define all throttle groups, one for each one of the
+drives and one that would apply to all of them:
+
+   -object throttle-group,id=limits0,x-iops-total=2000
+   -object throttle-group,id=limits1,x-iops-total=2500
+   -object throttle-group,id=limits2,x-iops-total=3000
+   -object throttle-group,id=limits012,x-iops-total=4000
+
+Now we can define the drives, and for each one of them we use two
+chained throttle filters: the drive's own filter and the combined
+filter.
+
+   -drive driver=throttle,throttle-group=limits012,
+          file.driver=throttle,file.throttle-group=limits0
+          file.file.driver=qcow2,file.file.file.filename=/path/to/disk0.qcow2
+   -drive driver=throttle,throttle-group=limits012,
+          file.driver=throttle,file.throttle-group=limits1
+          file.file.driver=qcow2,file.file.file.filename=/path/to/disk1.qcow2
+   -drive driver=throttle,throttle-group=limits012,
+          file.driver=throttle,file.throttle-group=limits2
+          file.file.driver=qcow2,file.file.file.filename=/path/to/disk2.qcow2
+
+In this example the individual drives have IOPS limits of 2000, 2500
+and 3000 respectively but the total combined I/O can never exceed 4000
+IOPS.
diff --git a/include/block/export.h b/include/block/export.h
new file mode 100644
index 0000000000..7feb02e10d
--- /dev/null
+++ b/include/block/export.h
@@ -0,0 +1,89 @@
+/*
+ * Declarations for block exports
+ *
+ * Copyright (c) 2012, 2020 Red Hat, Inc.
+ *
+ * Authors:
+ * Paolo Bonzini <pbonzini@redhat.com>
+ * Kevin Wolf <kwolf@redhat.com>
+ *
+ * This work is licensed under the terms of the GNU GPL, version 2 or
+ * later.  See the COPYING file in the top-level directory.
+ */
+
+#ifndef BLOCK_EXPORT_H
+#define BLOCK_EXPORT_H
+
+#include "qapi/qapi-types-block-export.h"
+#include "qemu/queue.h"
+
+typedef struct BlockExport BlockExport;
+
+typedef struct BlockExportDriver {
+    /* The export type that this driver services */
+    BlockExportType type;
+
+    /*
+     * The size of the driver-specific state that contains BlockExport as its
+     * first field.
+     */
+    size_t instance_size;
+
+    /* Creates and starts a new block export */
+    int (*create)(BlockExport *, BlockExportOptions *, Error **);
+
+    /*
+     * Frees a removed block export. This function is only called after all
+     * references have been dropped.
+     */
+    void (*delete)(BlockExport *);
+
+    /*
+     * Start to disconnect all clients and drop other references held
+     * internally by the export driver. When the function returns, there may
+     * still be active references while the export is in the process of
+     * shutting down.
+     */
+    void (*request_shutdown)(BlockExport *);
+} BlockExportDriver;
+
+struct BlockExport {
+    const BlockExportDriver *drv;
+
+    /* Unique identifier for the export */
+    char *id;
+
+    /*
+     * Reference count for this block export. This includes strong references
+     * both from the owner (qemu-nbd or the monitor) and clients connected to
+     * the export.
+     */
+    int refcount;
+
+    /*
+     * True if one of the references in refcount belongs to the user. After the
+     * user has dropped their reference, they may not e.g. remove the same
+     * export a second time (which would decrease the refcount without having
+     * it incremented first).
+     */
+    bool user_owned;
+
+    /* The AioContext whose lock protects this BlockExport object. */
+    AioContext *ctx;
+
+    /* The block device to export */
+    BlockBackend *blk;
+
+    /* List entry for block_exports */
+    QLIST_ENTRY(BlockExport) next;
+};
+
+BlockExport *blk_exp_add(BlockExportOptions *export, Error **errp);
+BlockExport *blk_exp_find(const char *id);
+void blk_exp_ref(BlockExport *exp);
+void blk_exp_unref(BlockExport *exp);
+void blk_exp_request_shutdown(BlockExport *exp);
+void blk_exp_close_all(void);
+void blk_exp_close_all_type(BlockExportType type);
+
+#endif
diff --git a/include/block/nbd.h b/include/block/nbd.h
index 9bc3bfaeec..3dd9a04546 100644
--- a/include/block/nbd.h
+++ b/include/block/nbd.h
@@ -20,11 +20,13 @@
 #ifndef NBD_H
 #define NBD_H
 
-#include "qapi/qapi-types-block.h"
+#include "block/export.h"
 #include "io/channel-socket.h"
 #include "crypto/tlscreds.h"
 #include "qapi/error.h"
 
+extern const BlockExportDriver blk_exp_nbd;
+
 /* Handshake phase structs - this struct is passed on the wire */
 
 struct NBDOption {
@@ -328,21 +330,10 @@ int nbd_errno_to_system_errno(int err);
 typedef struct NBDExport NBDExport;
 typedef struct NBDClient NBDClient;
 
-NBDExport *nbd_export_new(BlockDriverState *bs, uint64_t dev_offset,
-                          uint64_t size, const char *name, const char *desc,
-                          const char *bitmap, bool readonly, bool shared,
-                          void (*close)(NBDExport *), bool writethrough,
-                          BlockBackend *on_eject_blk, Error **errp);
-void nbd_export_close(NBDExport *exp);
-void nbd_export_remove(NBDExport *exp, NbdServerRemoveMode mode, Error **errp);
-void nbd_export_get(NBDExport *exp);
-void nbd_export_put(NBDExport *exp);
-
-BlockBackend *nbd_export_get_blockdev(NBDExport *exp);
+void nbd_export_set_on_eject_blk(BlockExport *exp, BlockBackend *blk);
 
 AioContext *nbd_export_aio_context(NBDExport *exp);
 NBDExport *nbd_export_find(const char *name);
-void nbd_export_close_all(void);
 
 void nbd_client_new(QIOChannelSocket *sioc,
                     QCryptoTLSCreds *tlscreds,
@@ -351,8 +342,11 @@ void nbd_client_new(QIOChannelSocket *sioc,
 void nbd_client_get(NBDClient *client);
 void nbd_client_put(NBDClient *client);
 
+void nbd_server_is_qemu_nbd(bool value);
+bool nbd_server_is_running(void);
 void nbd_server_start(SocketAddress *addr, const char *tls_creds,
-                      const char *tls_authz, Error **errp);
+                      const char *tls_authz, uint32_t max_connections,
+                      Error **errp);
 void nbd_server_start_options(NbdServerOptions *arg, Error **errp);
 
 /* nbd_read
diff --git a/meson.build b/meson.build
index 3161c1f037..0f0cc21d16 100644
--- a/meson.build
+++ b/meson.build
@@ -970,6 +970,7 @@ subdir('dump')
 
 block_ss.add(files(
   'block.c',
+  'blockdev-nbd.c',
   'blockjob.c',
   'job.c',
   'qemu-io-cmds.c',
@@ -982,7 +983,6 @@ subdir('block')
 
 blockdev_ss.add(files(
   'blockdev.c',
-  'blockdev-nbd.c',
   'iothread.c',
   'job-qmp.c',
 ))
diff --git a/nbd/server.c b/nbd/server.c
index 982de67816..f74766add7 100644
--- a/nbd/server.c
+++ b/nbd/server.c
@@ -18,6 +18,8 @@
  */
 
 #include "qemu/osdep.h"
+
+#include "block/export.h"
 #include "qapi/error.h"
 #include "qemu/queue.h"
 #include "trace.h"
@@ -80,20 +82,15 @@ struct NBDRequestData {
 };
 
 struct NBDExport {
-    int refcount;
-    void (*close)(NBDExport *exp);
+    BlockExport common;
 
-    BlockBackend *blk;
     char *name;
     char *description;
-    uint64_t dev_offset;
     uint64_t size;
     uint16_t nbdflags;
     QTAILQ_HEAD(, NBDClient) clients;
     QTAILQ_ENTRY(NBDExport) next;
 
-    AioContext *ctx;
-
     BlockBackend *eject_notifier_blk;
     Notifier eject_notifier;
 
@@ -102,8 +99,6 @@ struct NBDExport {
 };
 
 static QTAILQ_HEAD(, NBDExport) exports = QTAILQ_HEAD_INITIALIZER(exports);
-static QTAILQ_HEAD(, NBDExport) closed_exports =
-        QTAILQ_HEAD_INITIALIZER(closed_exports);
 
 /* NBDExportMetaContexts represents a list of contexts to be exported,
  * as selected by NBD_OPT_SET_META_CONTEXT. Also used for
@@ -498,7 +493,7 @@ static int nbd_negotiate_handle_export_name(NBDClient *client, bool no_zeroes,
     }
 
     QTAILQ_INSERT_TAIL(&client->exp->clients, client, next);
-    nbd_export_get(client->exp);
+    blk_exp_ref(&client->exp->common);
     nbd_check_meta_export(client);
 
     return 0;
@@ -647,7 +642,7 @@ static int nbd_negotiate_handle_info(NBDClient *client, Error **errp)
      * whether this is OPT_INFO or OPT_GO. */
     /* minimum - 1 for back-compat, or actual if client will obey it. */
     if (client->opt == NBD_OPT_INFO || blocksize) {
-        check_align = sizes[0] = blk_get_request_alignment(exp->blk);
+        check_align = sizes[0] = blk_get_request_alignment(exp->common.blk);
     } else {
         sizes[0] = 1;
     }
@@ -656,7 +651,7 @@ static int nbd_negotiate_handle_info(NBDClient *client, Error **errp)
      * TODO: is blk_bs(blk)->bl.opt_transfer appropriate? */
     sizes[1] = MAX(4096, sizes[0]);
     /* maximum - At most 32M, but smaller as appropriate. */
-    sizes[2] = MIN(blk_get_max_transfer(exp->blk), NBD_MAX_BUFFER_SIZE);
+    sizes[2] = MIN(blk_get_max_transfer(exp->common.blk), NBD_MAX_BUFFER_SIZE);
     trace_nbd_negotiate_handle_info_block_size(sizes[0], sizes[1], sizes[2]);
     sizes[0] = cpu_to_be32(sizes[0]);
     sizes[1] = cpu_to_be32(sizes[1]);
@@ -688,7 +683,7 @@ static int nbd_negotiate_handle_info(NBDClient *client, Error **errp)
      * tolerate all clients, regardless of alignments.
      */
     if (client->opt == NBD_OPT_INFO && !blocksize &&
-        blk_get_request_alignment(exp->blk) > 1) {
+        blk_get_request_alignment(exp->common.blk) > 1) {
         return nbd_negotiate_send_rep_err(client,
                                           NBD_REP_ERR_BLOCK_SIZE_REQD,
                                           errp,
@@ -706,7 +701,7 @@ static int nbd_negotiate_handle_info(NBDClient *client, Error **errp)
         client->exp = exp;
         client->check_align = check_align;
         QTAILQ_INSERT_TAIL(&client->exp->clients, client, next);
-        nbd_export_get(client->exp);
+        blk_exp_ref(&client->exp->common);
         nbd_check_meta_export(client);
         rc = 1;
     }
@@ -1333,8 +1328,8 @@ static coroutine_fn int nbd_negotiate(NBDClient *client, Error **errp)
     }
 
     /* Attach the channel to the same AioContext as the export */
-    if (client->exp && client->exp->ctx) {
-        qio_channel_attach_aio_context(client->ioc, client->exp->ctx);
+    if (client->exp && client->exp->common.ctx) {
+        qio_channel_attach_aio_context(client->ioc, client->exp->common.ctx);
     }
 
     assert(!client->optlen);
@@ -1405,7 +1400,7 @@ void nbd_client_put(NBDClient *client)
         g_free(client->tlsauthz);
         if (client->exp) {
             QTAILQ_REMOVE(&client->exp->clients, client, next);
-            nbd_export_put(client->exp);
+            blk_exp_unref(&client->exp->common);
         }
         g_free(client);
     }
@@ -1466,7 +1461,7 @@ static void blk_aio_attached(AioContext *ctx, void *opaque)
 
     trace_nbd_blk_aio_attached(exp->name, ctx);
 
-    exp->ctx = ctx;
+    exp->common.ctx = ctx;
 
     QTAILQ_FOREACH(client, &exp->clients, next) {
         qio_channel_attach_aio_context(client->ioc, ctx);
@@ -1484,72 +1479,92 @@ static void blk_aio_detach(void *opaque)
     NBDExport *exp = opaque;
     NBDClient *client;
 
-    trace_nbd_blk_aio_detach(exp->name, exp->ctx);
+    trace_nbd_blk_aio_detach(exp->name, exp->common.ctx);
 
     QTAILQ_FOREACH(client, &exp->clients, next) {
         qio_channel_detach_aio_context(client->ioc);
     }
 
-    exp->ctx = NULL;
+    exp->common.ctx = NULL;
 }
 
 static void nbd_eject_notifier(Notifier *n, void *data)
 {
     NBDExport *exp = container_of(n, NBDExport, eject_notifier);
-    AioContext *aio_context;
 
-    aio_context = exp->ctx;
-    aio_context_acquire(aio_context);
-    nbd_export_close(exp);
-    aio_context_release(aio_context);
+    blk_exp_request_shutdown(&exp->common);
+}
+
+void nbd_export_set_on_eject_blk(BlockExport *exp, BlockBackend *blk)
+{
+    NBDExport *nbd_exp = container_of(exp, NBDExport, common);
+    assert(exp->drv == &blk_exp_nbd);
+    assert(nbd_exp->eject_notifier_blk == NULL);
+
+    blk_ref(blk);
+    nbd_exp->eject_notifier_blk = blk;
+    nbd_exp->eject_notifier.notify = nbd_eject_notifier;
+    blk_add_remove_bs_notifier(blk, &nbd_exp->eject_notifier);
 }
 
-NBDExport *nbd_export_new(BlockDriverState *bs, uint64_t dev_offset,
-                          uint64_t size, const char *name, const char *desc,
-                          const char *bitmap, bool readonly, bool shared,
-                          void (*close)(NBDExport *), bool writethrough,
-                          BlockBackend *on_eject_blk, Error **errp)
+static int nbd_export_create(BlockExport *blk_exp, BlockExportOptions *exp_args,
+                             Error **errp)
 {
-    AioContext *ctx;
-    BlockBackend *blk;
-    NBDExport *exp = g_new0(NBDExport, 1);
-    uint64_t perm;
+    NBDExport *exp = container_of(blk_exp, NBDExport, common);
+    BlockExportOptionsNbd *arg = &exp_args->u.nbd;
+    BlockBackend *blk = blk_exp->blk;
+    int64_t size;
+    uint64_t perm, shared_perm;
+    bool readonly = !exp_args->writable;
+    bool shared = !exp_args->writable;
     int ret;
 
-    /*
-     * NBD exports are used for non-shared storage migration.  Make sure
-     * that BDRV_O_INACTIVE is cleared and the image is ready for write
-     * access since the export could be available before migration handover.
-     * ctx was acquired in the caller.
-     */
-    assert(name && strlen(name) <= NBD_MAX_STRING_SIZE);
-    ctx = bdrv_get_aio_context(bs);
-    bdrv_invalidate_cache(bs, NULL);
+    assert(exp_args->type == BLOCK_EXPORT_TYPE_NBD);
+
+    if (!nbd_server_is_running()) {
+        error_setg(errp, "NBD server not running");
+        return -EINVAL;
+    }
+
+    if (!arg->has_name) {
+        arg->name = exp_args->node_name;
+    }
+
+    if (strlen(arg->name) > NBD_MAX_STRING_SIZE) {
+        error_setg(errp, "export name '%s' too long", arg->name);
+        return -EINVAL;
+    }
+
+    if (arg->description && strlen(arg->description) > NBD_MAX_STRING_SIZE) {
+        error_setg(errp, "description '%s' too long", arg->description);
+        return -EINVAL;
+    }
+
+    if (nbd_export_find(arg->name)) {
+        error_setg(errp, "NBD server already has export named '%s'", arg->name);
+        return -EEXIST;
+    }
+
+    size = blk_getlength(blk);
+    if (size < 0) {
+        error_setg_errno(errp, -size,
+                         "Failed to determine the NBD export's length");
+        return size;
+    }
 
     /* Don't allow resize while the NBD server is running, otherwise we don't
      * care what happens with the node. */
-    perm = BLK_PERM_CONSISTENT_READ;
-    if (!readonly) {
-        perm |= BLK_PERM_WRITE;
-    }
-    blk = blk_new(ctx, perm,
-                  BLK_PERM_CONSISTENT_READ | BLK_PERM_WRITE_UNCHANGED |
-                  BLK_PERM_WRITE | BLK_PERM_GRAPH_MOD);
-    ret = blk_insert_bs(blk, bs, errp);
+    blk_get_perm(blk, &perm, &shared_perm);
+    ret = blk_set_perm(blk, perm, shared_perm & ~BLK_PERM_RESIZE, errp);
     if (ret < 0) {
-        goto fail;
+        return ret;
     }
-    blk_set_enable_write_cache(blk, !writethrough);
+
     blk_set_allow_aio_context_change(blk, true);
 
-    exp->refcount = 1;
     QTAILQ_INIT(&exp->clients);
-    exp->blk = blk;
-    assert(dev_offset <= INT64_MAX);
-    exp->dev_offset = dev_offset;
-    exp->name = g_strdup(name);
-    assert(!desc || strlen(desc) <= NBD_MAX_STRING_SIZE);
-    exp->description = g_strdup(desc);
+    exp->name = g_strdup(arg->name);
+    exp->description = g_strdup(arg->description);
     exp->nbdflags = (NBD_FLAG_HAS_FLAGS | NBD_FLAG_SEND_FLUSH |
                      NBD_FLAG_SEND_FUA | NBD_FLAG_SEND_CACHE);
     if (readonly) {
@@ -1561,14 +1576,14 @@ NBDExport *nbd_export_new(BlockDriverState *bs, uint64_t dev_offset,
         exp->nbdflags |= (NBD_FLAG_SEND_TRIM | NBD_FLAG_SEND_WRITE_ZEROES |
                           NBD_FLAG_SEND_FAST_ZERO);
     }
-    assert(size <= INT64_MAX - dev_offset);
     exp->size = QEMU_ALIGN_DOWN(size, BDRV_SECTOR_SIZE);
 
-    if (bitmap) {
+    if (arg->bitmap) {
+        BlockDriverState *bs = blk_bs(blk);
         BdrvDirtyBitmap *bm = NULL;
 
         while (bs) {
-            bm = bdrv_find_dirty_bitmap(bs, bitmap);
+            bm = bdrv_find_dirty_bitmap(bs, arg->bitmap);
             if (bm != NULL) {
                 break;
             }
@@ -1577,50 +1592,43 @@ NBDExport *nbd_export_new(BlockDriverState *bs, uint64_t dev_offset,
         }
 
         if (bm == NULL) {
-            error_setg(errp, "Bitmap '%s' is not found", bitmap);
+            ret = -ENOENT;
+            error_setg(errp, "Bitmap '%s' is not found", arg->bitmap);
             goto fail;
         }
 
         if (bdrv_dirty_bitmap_check(bm, BDRV_BITMAP_ALLOW_RO, errp)) {
+            ret = -EINVAL;
             goto fail;
         }
 
         if (readonly && bdrv_is_writable(bs) &&
             bdrv_dirty_bitmap_enabled(bm)) {
+            ret = -EINVAL;
             error_setg(errp,
                        "Enabled bitmap '%s' incompatible with readonly export",
-                       bitmap);
+                       arg->bitmap);
             goto fail;
         }
 
         bdrv_dirty_bitmap_set_busy(bm, true);
         exp->export_bitmap = bm;
-        assert(strlen(bitmap) <= BDRV_BITMAP_MAX_NAME_SIZE);
+        assert(strlen(arg->bitmap) <= BDRV_BITMAP_MAX_NAME_SIZE);
         exp->export_bitmap_context = g_strdup_printf("qemu:dirty-bitmap:%s",
-                                                     bitmap);
+                                                     arg->bitmap);
         assert(strlen(exp->export_bitmap_context) < NBD_MAX_STRING_SIZE);
     }
 
-    exp->close = close;
-    exp->ctx = ctx;
     blk_add_aio_context_notifier(blk, blk_aio_attached, blk_aio_detach, exp);
 
-    if (on_eject_blk) {
-        blk_ref(on_eject_blk);
-        exp->eject_notifier_blk = on_eject_blk;
-        exp->eject_notifier.notify = nbd_eject_notifier;
-        blk_add_remove_bs_notifier(on_eject_blk, &exp->eject_notifier);
-    }
     QTAILQ_INSERT_TAIL(&exports, exp, next);
-    nbd_export_get(exp);
-    return exp;
+
+    return 0;
 
 fail:
-    blk_unref(blk);
     g_free(exp->name);
     g_free(exp->description);
-    g_free(exp);
-    return NULL;
+    return ret;
 }
 
 NBDExport *nbd_export_find(const char *name)
@@ -1638,14 +1646,15 @@ NBDExport *nbd_export_find(const char *name)
 AioContext *
 nbd_export_aio_context(NBDExport *exp)
 {
-    return exp->ctx;
+    return exp->common.ctx;
 }
 
-void nbd_export_close(NBDExport *exp)
+static void nbd_export_request_shutdown(BlockExport *blk_exp)
 {
+    NBDExport *exp = container_of(blk_exp, NBDExport, common);
     NBDClient *client, *next;
 
-    nbd_export_get(exp);
+    blk_exp_ref(&exp->common);
     /*
      * TODO: Should we expand QMP NbdServerRemoveNode enum to allow a
      * close mode that stops advertising the export to new clients but
@@ -1657,101 +1666,46 @@ void nbd_export_close(NBDExport *exp)
         client_close(client, true);
     }
     if (exp->name) {
-        nbd_export_put(exp);
         g_free(exp->name);
         exp->name = NULL;
         QTAILQ_REMOVE(&exports, exp, next);
-        QTAILQ_INSERT_TAIL(&closed_exports, exp, next);
     }
-    g_free(exp->description);
-    exp->description = NULL;
-    nbd_export_put(exp);
-}
-
-void nbd_export_remove(NBDExport *exp, NbdServerRemoveMode mode, Error **errp)
-{
-    ERRP_GUARD();
-    if (mode == NBD_SERVER_REMOVE_MODE_HARD || QTAILQ_EMPTY(&exp->clients)) {
-        nbd_export_close(exp);
-        return;
-    }
-
-    assert(mode == NBD_SERVER_REMOVE_MODE_SAFE);
-
-    error_setg(errp, "export '%s' still in use", exp->name);
-    error_append_hint(errp, "Use mode='hard' to force client disconnect\n");
-}
-
-void nbd_export_get(NBDExport *exp)
-{
-    assert(exp->refcount > 0);
-    exp->refcount++;
+    blk_exp_unref(&exp->common);
 }
 
-void nbd_export_put(NBDExport *exp)
+static void nbd_export_delete(BlockExport *blk_exp)
 {
-    assert(exp->refcount > 0);
-    if (exp->refcount == 1) {
-        nbd_export_close(exp);
-    }
+    NBDExport *exp = container_of(blk_exp, NBDExport, common);
 
-    /* nbd_export_close() may theoretically reduce refcount to 0. It may happen
-     * if someone calls nbd_export_put() on named export not through
-     * nbd_export_set_name() when refcount is 1. So, let's assert that
-     * it is > 0.
-     */
-    assert(exp->refcount > 0);
-    if (--exp->refcount == 0) {
-        assert(exp->name == NULL);
-        assert(exp->description == NULL);
+    assert(exp->name == NULL);
+    assert(QTAILQ_EMPTY(&exp->clients));
 
-        if (exp->close) {
-            exp->close(exp);
-        }
-
-        if (exp->blk) {
-            if (exp->eject_notifier_blk) {
-                notifier_remove(&exp->eject_notifier);
-                blk_unref(exp->eject_notifier_blk);
-            }
-            blk_remove_aio_context_notifier(exp->blk, blk_aio_attached,
-                                            blk_aio_detach, exp);
-            blk_unref(exp->blk);
-            exp->blk = NULL;
-        }
+    g_free(exp->description);
+    exp->description = NULL;
 
-        if (exp->export_bitmap) {
-            bdrv_dirty_bitmap_set_busy(exp->export_bitmap, false);
-            g_free(exp->export_bitmap_context);
+    if (exp->common.blk) {
+        if (exp->eject_notifier_blk) {
+            notifier_remove(&exp->eject_notifier);
+            blk_unref(exp->eject_notifier_blk);
         }
-
-        QTAILQ_REMOVE(&closed_exports, exp, next);
-        g_free(exp);
-        aio_wait_kick();
+        blk_remove_aio_context_notifier(exp->common.blk, blk_aio_attached,
+                                        blk_aio_detach, exp);
     }
-}
 
-BlockBackend *nbd_export_get_blockdev(NBDExport *exp)
-{
-    return exp->blk;
-}
-
-void nbd_export_close_all(void)
-{
-    NBDExport *exp, *next;
-    AioContext *aio_context;
-
-    QTAILQ_FOREACH_SAFE(exp, &exports, next, next) {
-        aio_context = exp->ctx;
-        aio_context_acquire(aio_context);
-        nbd_export_close(exp);
-        aio_context_release(aio_context);
+    if (exp->export_bitmap) {
+        bdrv_dirty_bitmap_set_busy(exp->export_bitmap, false);
+        g_free(exp->export_bitmap_context);
     }
-
-    AIO_WAIT_WHILE(NULL, !(QTAILQ_EMPTY(&exports) &&
-                           QTAILQ_EMPTY(&closed_exports)));
 }
 
+const BlockExportDriver blk_exp_nbd = {
+    .type               = BLOCK_EXPORT_TYPE_NBD,
+    .instance_size      = sizeof(NBDExport),
+    .create             = nbd_export_create,
+    .delete             = nbd_export_delete,
+    .request_shutdown   = nbd_export_request_shutdown,
+};
+
 static int coroutine_fn nbd_co_send_iov(NBDClient *client, struct iovec *iov,
                                         unsigned niov, Error **errp)
 {
@@ -1888,7 +1842,7 @@ static int coroutine_fn nbd_co_send_sparse_read(NBDClient *client,
 
     while (progress < size) {
         int64_t pnum;
-        int status = bdrv_block_status_above(blk_bs(exp->blk), NULL,
+        int status = bdrv_block_status_above(blk_bs(exp->common.blk), NULL,
                                              offset + progress,
                                              size - progress, &pnum, NULL,
                                              NULL);
@@ -1920,7 +1874,7 @@ static int coroutine_fn nbd_co_send_sparse_read(NBDClient *client,
             stl_be_p(&chunk.length, pnum);
             ret = nbd_co_send_iov(client, iov, 1, errp);
         } else {
-            ret = blk_pread(exp->blk, offset + progress + exp->dev_offset,
+            ret = blk_pread(exp->common.blk, offset + progress,
                             data + progress, pnum);
             if (ret < 0) {
                 error_setg_errno(errp, -ret, "reading from file failed");
@@ -2185,7 +2139,8 @@ static int nbd_co_receive_request(NBDRequestData *req, NBDRequest *request,
         }
 
         if (request->type != NBD_CMD_CACHE) {
-            req->data = blk_try_blockalign(client->exp->blk, request->len);
+            req->data = blk_try_blockalign(client->exp->common.blk,
+                                           request->len);
             if (req->data == NULL) {
                 error_setg(errp, "No memory");
                 return -ENOMEM;
@@ -2281,7 +2236,7 @@ static coroutine_fn int nbd_do_cmd_read(NBDClient *client, NBDRequest *request,
 
     /* XXX: NBD Protocol only documents use of FUA with WRITE */
     if (request->flags & NBD_CMD_FLAG_FUA) {
-        ret = blk_co_flush(exp->blk);
+        ret = blk_co_flush(exp->common.blk);
         if (ret < 0) {
             return nbd_send_generic_reply(client, request->handle, ret,
                                           "flush failed", errp);
@@ -2295,8 +2250,7 @@ static coroutine_fn int nbd_do_cmd_read(NBDClient *client, NBDRequest *request,
                                        data, request->len, errp);
     }
 
-    ret = blk_pread(exp->blk, request->from + exp->dev_offset, data,
-                    request->len);
+    ret = blk_pread(exp->common.blk, request->from, data, request->len);
     if (ret < 0) {
         return nbd_send_generic_reply(client, request->handle, ret,
                                       "reading from file failed", errp);
@@ -2331,7 +2285,7 @@ static coroutine_fn int nbd_do_cmd_cache(NBDClient *client, NBDRequest *request,
 
     assert(request->type == NBD_CMD_CACHE);
 
-    ret = blk_co_preadv(exp->blk, request->from + exp->dev_offset, request->len,
+    ret = blk_co_preadv(exp->common.blk, request->from, request->len,
                         NULL, BDRV_REQ_COPY_ON_READ | BDRV_REQ_PREFETCH);
 
     return nbd_send_generic_reply(client, request->handle, ret,
@@ -2362,8 +2316,8 @@ static coroutine_fn int nbd_handle_request(NBDClient *client,
         if (request->flags & NBD_CMD_FLAG_FUA) {
             flags |= BDRV_REQ_FUA;
         }
-        ret = blk_pwrite(exp->blk, request->from + exp->dev_offset,
-                         data, request->len, flags);
+        ret = blk_pwrite(exp->common.blk, request->from, data, request->len,
+                         flags);
         return nbd_send_generic_reply(client, request->handle, ret,
                                       "writing to file failed", errp);
 
@@ -2384,8 +2338,7 @@ static coroutine_fn int nbd_handle_request(NBDClient *client,
             int align = client->check_align ?: 1;
             int len = MIN(request->len, QEMU_ALIGN_DOWN(BDRV_REQUEST_MAX_BYTES,
                                                         align));
-            ret = blk_pwrite_zeroes(exp->blk, request->from + exp->dev_offset,
-                                    len, flags);
+            ret = blk_pwrite_zeroes(exp->common.blk, request->from, len, flags);
             request->len -= len;
             request->from += len;
         }
@@ -2397,7 +2350,7 @@ static coroutine_fn int nbd_handle_request(NBDClient *client,
         abort();
 
     case NBD_CMD_FLUSH:
-        ret = blk_co_flush(exp->blk);
+        ret = blk_co_flush(exp->common.blk);
         return nbd_send_generic_reply(client, request->handle, ret,
                                       "flush failed", errp);
 
@@ -2408,13 +2361,12 @@ static coroutine_fn int nbd_handle_request(NBDClient *client,
             int align = client->check_align ?: 1;
             int len = MIN(request->len, QEMU_ALIGN_DOWN(BDRV_REQUEST_MAX_BYTES,
                                                         align));
-            ret = blk_co_pdiscard(exp->blk, request->from + exp->dev_offset,
-                                  len);
+            ret = blk_co_pdiscard(exp->common.blk, request->from, len);
             request->len -= len;
             request->from += len;
         }
         if (ret >= 0 && request->flags & NBD_CMD_FLAG_FUA) {
-            ret = blk_co_flush(exp->blk);
+            ret = blk_co_flush(exp->common.blk);
         }
         return nbd_send_generic_reply(client, request->handle, ret,
                                       "discard failed", errp);
@@ -2432,7 +2384,8 @@ static coroutine_fn int nbd_handle_request(NBDClient *client,
 
             if (client->export_meta.base_allocation) {
                 ret = nbd_co_send_block_status(client, request->handle,
-                                               blk_bs(exp->blk), request->from,
+                                               blk_bs(exp->common.blk),
+                                               request->from,
                                                request->len, dont_fragment,
                                                !client->export_meta.bitmap,
                                                NBD_META_ID_BASE_ALLOCATION,
@@ -2546,7 +2499,7 @@ static void nbd_client_receive_next_request(NBDClient *client)
     if (!client->recv_coroutine && client->nb_requests < MAX_NBD_REQUESTS) {
         nbd_client_get(client);
         client->recv_coroutine = qemu_coroutine_create(nbd_trip, client);
-        aio_co_schedule(client->exp->ctx, client->recv_coroutine);
+        aio_co_schedule(client->exp->common.ctx, client->recv_coroutine);
     }
 }
 
diff --git a/qapi/block-core.json b/qapi/block-core.json
index 86ed72ef9f..12a24772b5 100644
--- a/qapi/block-core.json
+++ b/qapi/block-core.json
@@ -5195,172 +5195,6 @@
              '*force': 'bool' } }
 
 ##
-# @NbdServerOptions:
-#
-# @addr: Address on which to listen.
-# @tls-creds: ID of the TLS credentials object (since 2.6).
-# @tls-authz: ID of the QAuthZ authorization object used to validate
-#             the client's x509 distinguished name. This object is
-#             is only resolved at time of use, so can be deleted and
-#             recreated on the fly while the NBD server is active.
-#             If missing, it will default to denying access (since 4.0).
-#
-# Keep this type consistent with the nbd-server-start arguments. The only
-# intended difference is using SocketAddress instead of SocketAddressLegacy.
-#
-# Since: 4.2
-##
-{ 'struct': 'NbdServerOptions',
-  'data': { 'addr': 'SocketAddress',
-            '*tls-creds': 'str',
-            '*tls-authz': 'str'} }
-
-##
-# @nbd-server-start:
-#
-# Start an NBD server listening on the given host and port.  Block
-# devices can then be exported using @nbd-server-add.  The NBD
-# server will present them as named exports; for example, another
-# QEMU instance could refer to them as "nbd:HOST:PORT:exportname=NAME".
-#
-# Keep this type consistent with the NbdServerOptions type. The only intended
-# difference is using SocketAddressLegacy instead of SocketAddress.
-#
-# @addr: Address on which to listen.
-# @tls-creds: ID of the TLS credentials object (since 2.6).
-# @tls-authz: ID of the QAuthZ authorization object used to validate
-#             the client's x509 distinguished name. This object is
-#             is only resolved at time of use, so can be deleted and
-#             recreated on the fly while the NBD server is active.
-#             If missing, it will default to denying access (since 4.0).
-#
-# Returns: error if the server is already running.
-#
-# Since: 1.3.0
-##
-{ 'command': 'nbd-server-start',
-  'data': { 'addr': 'SocketAddressLegacy',
-            '*tls-creds': 'str',
-            '*tls-authz': 'str'} }
-
-##
-# @BlockExportNbd:
-#
-# An NBD block export.
-#
-# @device: The device name or node name of the node to be exported
-#
-# @name: Export name. If unspecified, the @device parameter is used as the
-#        export name. (Since 2.12)
-#
-# @description: Free-form description of the export, up to 4096 bytes.
-#               (Since 5.0)
-#
-# @writable: Whether clients should be able to write to the device via the
-#            NBD connection (default false).
-#
-# @bitmap: Also export the dirty bitmap reachable from @device, so the
-#          NBD client can use NBD_OPT_SET_META_CONTEXT with
-#          "qemu:dirty-bitmap:NAME" to inspect the bitmap. (since 4.0)
-#
-# Since: 5.0
-##
-{ 'struct': 'BlockExportNbd',
-  'data': {'device': 'str', '*name': 'str', '*description': 'str',
-           '*writable': 'bool', '*bitmap': 'str' } }
-
-##
-# @nbd-server-add:
-#
-# Export a block node to QEMU's embedded NBD server.
-#
-# Returns: error if the server is not running, or export with the same name
-#          already exists.
-#
-# Since: 1.3.0
-##
-{ 'command': 'nbd-server-add',
-  'data': 'BlockExportNbd', 'boxed': true }
-
-##
-# @NbdServerRemoveMode:
-#
-# Mode for removing an NBD export.
-#
-# @safe: Remove export if there are no existing connections, fail otherwise.
-#
-# @hard: Drop all connections immediately and remove export.
-#
-# Potential additional modes to be added in the future:
-#
-# hide: Just hide export from new clients, leave existing connections as is.
-# Remove export after all clients are disconnected.
-#
-# soft: Hide export from new clients, answer with ESHUTDOWN for all further
-# requests from existing clients.
-#
-# Since: 2.12
-##
-{'enum': 'NbdServerRemoveMode', 'data': ['safe', 'hard']}
-
-##
-# @nbd-server-remove:
-#
-# Remove NBD export by name.
-#
-# @name: Export name.
-#
-# @mode: Mode of command operation. See @NbdServerRemoveMode description.
-#        Default is 'safe'.
-#
-# Returns: error if
-#            - the server is not running
-#            - export is not found
-#            - mode is 'safe' and there are existing connections
-#
-# Since: 2.12
-##
-{ 'command': 'nbd-server-remove',
-  'data': {'name': 'str', '*mode': 'NbdServerRemoveMode'} }
-
-##
-# @nbd-server-stop:
-#
-# Stop QEMU's embedded NBD server, and unregister all devices previously
-# added via @nbd-server-add.
-#
-# Since: 1.3.0
-##
-{ 'command': 'nbd-server-stop' }
-
-##
-# @BlockExportType:
-#
-# An enumeration of block export types
-#
-# @nbd: NBD export
-#
-# Since: 4.2
-##
-{ 'enum': 'BlockExportType',
-  'data': [ 'nbd' ] }
-
-##
-# @BlockExport:
-#
-# Describes a block export, i.e. how single node should be exported on an
-# external interface.
-#
-# Since: 4.2
-##
-{ 'union': 'BlockExport',
-  'base': { 'type': 'BlockExportType' },
-  'discriminator': 'type',
-  'data': {
-      'nbd': 'BlockExportNbd'
-   } }
-
-##
 # @QuorumOpType:
 #
 # An enumeration of the quorum operation types
diff --git a/qapi/block-export.json b/qapi/block-export.json
new file mode 100644
index 0000000000..65804834d9
--- /dev/null
+++ b/qapi/block-export.json
@@ -0,0 +1,291 @@
+# -*- Mode: Python -*-
+# vim: filetype=python
+
+##
+# == Block device exports
+##
+
+{ 'include': 'sockets.json' }
+
+##
+# @NbdServerOptions:
+#
+# Keep this type consistent with the nbd-server-start arguments. The only
+# intended difference is using SocketAddress instead of SocketAddressLegacy.
+#
+# @addr: Address on which to listen.
+# @tls-creds: ID of the TLS credentials object (since 2.6).
+# @tls-authz: ID of the QAuthZ authorization object used to validate
+#             the client's x509 distinguished name. This object is
+#             is only resolved at time of use, so can be deleted and
+#             recreated on the fly while the NBD server is active.
+#             If missing, it will default to denying access (since 4.0).
+# @max-connections: The maximum number of connections to allow at the same
+#                   time, 0 for unlimited. (since 5.2; default: 0)
+#
+# Since: 4.2
+##
+{ 'struct': 'NbdServerOptions',
+  'data': { 'addr': 'SocketAddress',
+            '*tls-creds': 'str',
+            '*tls-authz': 'str',
+            '*max-connections': 'uint32' } }
+
+##
+# @nbd-server-start:
+#
+# Start an NBD server listening on the given host and port.  Block
+# devices can then be exported using @nbd-server-add.  The NBD
+# server will present them as named exports; for example, another
+# QEMU instance could refer to them as "nbd:HOST:PORT:exportname=NAME".
+#
+# Keep this type consistent with the NbdServerOptions type. The only intended
+# difference is using SocketAddressLegacy instead of SocketAddress.
+#
+# @addr: Address on which to listen.
+# @tls-creds: ID of the TLS credentials object (since 2.6).
+# @tls-authz: ID of the QAuthZ authorization object used to validate
+#             the client's x509 distinguished name. This object is
+#             is only resolved at time of use, so can be deleted and
+#             recreated on the fly while the NBD server is active.
+#             If missing, it will default to denying access (since 4.0).
+# @max-connections: The maximum number of connections to allow at the same
+#                   time, 0 for unlimited. (since 5.2; default: 0)
+#
+# Returns: error if the server is already running.
+#
+# Since: 1.3.0
+##
+{ 'command': 'nbd-server-start',
+  'data': { 'addr': 'SocketAddressLegacy',
+            '*tls-creds': 'str',
+            '*tls-authz': 'str',
+            '*max-connections': 'uint32' } }
+
+##
+# @BlockExportOptionsNbd:
+#
+# An NBD block export (options shared between nbd-server-add and the NBD branch
+# of block-export-add).
+#
+# @name: Export name. If unspecified, the @device parameter is used as the
+#        export name. (Since 2.12)
+#
+# @description: Free-form description of the export, up to 4096 bytes.
+#               (Since 5.0)
+#
+# @bitmap: Also export the dirty bitmap reachable from @device, so the
+#          NBD client can use NBD_OPT_SET_META_CONTEXT with
+#          "qemu:dirty-bitmap:NAME" to inspect the bitmap. (since 4.0)
+#
+# Since: 5.0
+##
+{ 'struct': 'BlockExportOptionsNbd',
+  'data': { '*name': 'str', '*description': 'str',
+            '*bitmap': 'str' } }
+
+##
+# @NbdServerAddOptions:
+#
+# An NBD block export.
+#
+# @device: The device name or node name of the node to be exported
+#
+# @writable: Whether clients should be able to write to the device via the
+#            NBD connection (default false).
+#
+# Since: 5.0
+##
+{ 'struct': 'NbdServerAddOptions',
+  'base': 'BlockExportOptionsNbd',
+  'data': { 'device': 'str',
+            '*writable': 'bool' } }
+
+##
+# @nbd-server-add:
+#
+# Export a block node to QEMU's embedded NBD server.
+#
+# The export name will be used as the id for the resulting block export.
+#
+# Features:
+# @deprecated: This command is deprecated. Use @block-export-add instead.
+#
+# Returns: error if the server is not running, or export with the same name
+#          already exists.
+#
+# Since: 1.3.0
+##
+{ 'command': 'nbd-server-add',
+  'data': 'NbdServerAddOptions', 'boxed': true, 'features': ['deprecated'] }
+
+##
+# @BlockExportRemoveMode:
+#
+# Mode for removing a block export.
+#
+# @safe: Remove export if there are no existing connections, fail otherwise.
+#
+# @hard: Drop all connections immediately and remove export.
+#
+# Potential additional modes to be added in the future:
+#
+# hide: Just hide export from new clients, leave existing connections as is.
+# Remove export after all clients are disconnected.
+#
+# soft: Hide export from new clients, answer with ESHUTDOWN for all further
+# requests from existing clients.
+#
+# Since: 2.12
+##
+{'enum': 'BlockExportRemoveMode', 'data': ['safe', 'hard']}
+
+##
+# @nbd-server-remove:
+#
+# Remove NBD export by name.
+#
+# @name: Block export id.
+#
+# @mode: Mode of command operation. See @BlockExportRemoveMode description.
+#        Default is 'safe'.
+#
+# Features:
+# @deprecated: This command is deprecated. Use @block-export-del instead.
+#
+# Returns: error if
+#            - the server is not running
+#            - export is not found
+#            - mode is 'safe' and there are existing connections
+#
+# Since: 2.12
+##
+{ 'command': 'nbd-server-remove',
+  'data': {'name': 'str', '*mode': 'BlockExportRemoveMode'},
+  'features': ['deprecated'] }
+
+##
+# @nbd-server-stop:
+#
+# Stop QEMU's embedded NBD server, and unregister all devices previously
+# added via @nbd-server-add.
+#
+# Since: 1.3.0
+##
+{ 'command': 'nbd-server-stop' }
+
+##
+# @BlockExportType:
+#
+# An enumeration of block export types
+#
+# @nbd: NBD export
+#
+# Since: 4.2
+##
+{ 'enum': 'BlockExportType',
+  'data': [ 'nbd' ] }
+
+##
+# @BlockExportOptions:
+#
+# Describes a block export, i.e. how single node should be exported on an
+# external interface.
+#
+# @id: A unique identifier for the block export (across all export types)
+#
+# @node-name: The node name of the block node to be exported (since: 5.2)
+#
+# @writable: True if clients should be able to write to the export
+#            (default false)
+#
+# @writethrough: If true, caches are flushed after every write request to the
+#                export before completion is signalled. (since: 5.2;
+#                default: false)
+#
+# Since: 4.2
+##
+{ 'union': 'BlockExportOptions',
+  'base': { 'type': 'BlockExportType',
+            'id': 'str',
+            'node-name': 'str',
+            '*writable': 'bool',
+            '*writethrough': 'bool' },
+  'discriminator': 'type',
+  'data': {
+      'nbd': 'BlockExportOptionsNbd'
+   } }
+
+##
+# @block-export-add:
+#
+# Creates a new block export.
+#
+# Since: 5.2
+##
+{ 'command': 'block-export-add',
+  'data': 'BlockExportOptions', 'boxed': true }
+
+##
+# @block-export-del:
+#
+# Request to remove a block export. This drops the user's reference to the
+# export, but the export may still stay around after this command returns until
+# the shutdown of the export has completed.
+#
+# @id: Block export id.
+#
+# @mode: Mode of command operation. See @BlockExportRemoveMode description.
+#        Default is 'safe'.
+#
+# Returns: Error if the export is not found or @mode is 'safe' and the export
+#          is still in use (e.g. by existing client connections)
+#
+# Since: 5.2
+##
+{ 'command': 'block-export-del',
+  'data': { 'id': 'str', '*mode': 'BlockExportRemoveMode' } }
+
+##
+# @BLOCK_EXPORT_DELETED:
+#
+# Emitted when a block export is removed and its id can be reused.
+#
+# @id: Block export id.
+#
+# Since: 5.2
+##
+{ 'event': 'BLOCK_EXPORT_DELETED',
+  'data': { 'id': 'str' } }
+
+##
+# @BlockExportInfo:
+#
+# Information about a single block export.
+#
+# @id: The unique identifier for the block export
+#
+# @type: The block export type
+#
+# @node-name: The node name of the block node that is exported
+#
+# @shutting-down: True if the export is shutting down (e.g. after a
+#                 block-export-del command, but before the shutdown has
+#                 completed)
+#
+# Since:  5.2
+##
+{ 'struct': 'BlockExportInfo',
+  'data': { 'id': 'str',
+            'type': 'BlockExportType',
+            'node-name': 'str',
+            'shutting-down': 'bool' } }
+
+##
+# @query-block-exports:
+#
+# Returns: A list of BlockExportInfo describing all block exports
+#
+# Since: 5.2
+##
+{ 'command': 'query-block-exports', 'returns': ['BlockExportInfo'] }
diff --git a/qapi/meson.build b/qapi/meson.build
index 7c4a89a882..ea359a0148 100644
--- a/qapi/meson.build
+++ b/qapi/meson.build
@@ -17,8 +17,9 @@ qapi_all_modules = [
   'acpi',
   'audio',
   'authz',
-  'block-core',
   'block',
+  'block-core',
+  'block-export',
   'char',
   'common',
   'control',
@@ -49,6 +50,7 @@ qapi_all_modules = [
 
 qapi_storage_daemon_modules = [
   'block-core',
+  'block-export',
   'char',
   'common',
   'control',
diff --git a/qapi/qapi-schema.json b/qapi/qapi-schema.json
index 0c6ca5c000..8d567e1386 100644
--- a/qapi/qapi-schema.json
+++ b/qapi/qapi-schema.json
@@ -66,6 +66,7 @@
 { 'include': 'run-state.json' }
 { 'include': 'crypto.json' }
 { 'include': 'block.json' }
+{ 'include': 'block-export.json' }
 { 'include': 'char.json' }
 { 'include': 'dump.json' }
 { 'include': 'job.json' }
diff --git a/qemu-io-cmds.c b/qemu-io-cmds.c
index baeae86d8c..4153f1c0b0 100644
--- a/qemu-io-cmds.c
+++ b/qemu-io-cmds.c
@@ -2383,14 +2383,7 @@ static const cmdinfo_t sleep_cmd = {
 
 static void help_oneline(const char *cmd, const cmdinfo_t *ct)
 {
-    if (cmd) {
-        printf("%s ", cmd);
-    } else {
-        printf("%s ", ct->name);
-        if (ct->altname) {
-            printf("(or %s) ", ct->altname);
-        }
-    }
+    printf("%s ", cmd);
 
     if (ct->args) {
         printf("%s ", ct->args);
@@ -2420,7 +2413,7 @@ static int help_f(BlockBackend *blk, int argc, char **argv)
 {
     const cmdinfo_t *ct;
 
-    if (argc == 1) {
+    if (argc < 2) {
         help_all();
         return 0;
     }
diff --git a/qemu-nbd.c b/qemu-nbd.c
index e39d3b23c1..bacb69b089 100644
--- a/qemu-nbd.c
+++ b/qemu-nbd.c
@@ -65,12 +65,11 @@
 
 #define MBR_SIZE 512
 
-static NBDExport *export;
 static int verbose;
 static char *srcpath;
 static SocketAddress *saddr;
 static int persistent = 0;
-static enum { RUNNING, TERMINATE, TERMINATING, TERMINATED } state;
+static enum { RUNNING, TERMINATE, TERMINATED } state;
 static int shared = 1;
 static int nb_fds;
 static QIONetListener *server;
@@ -332,12 +331,6 @@ static int nbd_can_accept(void)
     return state == RUNNING && nb_fds < shared;
 }
 
-static void nbd_export_closed(NBDExport *export)
-{
-    assert(state == TERMINATING);
-    state = TERMINATED;
-}
-
 static void nbd_update_server_watch(void);
 
 static void nbd_client_closed(NBDClient *client, bool negotiated)
@@ -524,7 +517,6 @@ int main(int argc, char **argv)
     const char *port = NULL;
     char *sockpath = NULL;
     char *device = NULL;
-    int64_t fd_size;
     QemuOpts *sn_opts = NULL;
     const char *sn_id_or_name = NULL;
     const char *sopt = "hVb:o:p:rsnc:dvk:e:f:tl:x:T:D:B:L";
@@ -587,6 +579,7 @@ int main(int argc, char **argv)
     int old_stderr = -1;
     unsigned socket_activation;
     const char *pid_file_name = NULL;
+    BlockExportOptions *export_opts;
 
 #if HAVE_NBD_DEVICE
     /* The client thread uses SIGTERM to interrupt the server.  A signal
@@ -1037,6 +1030,17 @@ int main(int argc, char **argv)
     }
     bs = blk_bs(blk);
 
+    if (dev_offset) {
+        QDict *raw_opts = qdict_new();
+        qdict_put_str(raw_opts, "driver", "raw");
+        qdict_put_str(raw_opts, "file", bs->node_name);
+        qdict_put_int(raw_opts, "offset", dev_offset);
+        bs = bdrv_open(NULL, NULL, raw_opts, flags, &error_fatal);
+        blk_remove_bs(blk);
+        blk_insert_bs(blk, bs, &error_fatal);
+        bdrv_unref(bs);
+    }
+
     blk_set_enable_write_cache(blk, !writethrough);
 
     if (sn_opts) {
@@ -1054,24 +1058,29 @@ int main(int argc, char **argv)
     }
 
     bs->detect_zeroes = detect_zeroes;
-    fd_size = blk_getlength(blk);
-    if (fd_size < 0) {
-        error_report("Failed to determine the image length: %s",
-                     strerror(-fd_size));
-        exit(EXIT_FAILURE);
-    }
 
-    if (dev_offset >= fd_size) {
-        error_report("Offset (%" PRIu64 ") has to be smaller than the image "
-                     "size (%" PRId64 ")", dev_offset, fd_size);
-        exit(EXIT_FAILURE);
-    }
-    fd_size -= dev_offset;
-
-    export = nbd_export_new(bs, dev_offset, fd_size, export_name,
-                            export_description, bitmap, readonly, shared > 1,
-                            nbd_export_closed, writethrough, NULL,
-                            &error_fatal);
+    nbd_server_is_qemu_nbd(true);
+
+    export_opts = g_new(BlockExportOptions, 1);
+    *export_opts = (BlockExportOptions) {
+        .type               = BLOCK_EXPORT_TYPE_NBD,
+        .id                 = g_strdup("qemu-nbd-export"),
+        .node_name          = g_strdup(bdrv_get_node_name(bs)),
+        .has_writethrough   = true,
+        .writethrough       = writethrough,
+        .has_writable       = true,
+        .writable           = !readonly,
+        .u.nbd = {
+            .has_name           = true,
+            .name               = g_strdup(export_name),
+            .has_description    = !!export_description,
+            .description        = g_strdup(export_description),
+            .has_bitmap         = !!bitmap,
+            .bitmap             = g_strdup(bitmap),
+        },
+    };
+    blk_exp_add(export_opts, &error_fatal);
+    qapi_free_BlockExportOptions(export_opts);
 
     if (device) {
 #if HAVE_NBD_DEVICE
@@ -1111,10 +1120,8 @@ int main(int argc, char **argv)
     do {
         main_loop_wait(false);
         if (state == TERMINATE) {
-            state = TERMINATING;
-            nbd_export_close(export);
-            nbd_export_put(export);
-            export = NULL;
+            blk_exp_close_all();
+            state = TERMINATED;
         }
     } while (state != TERMINATED);
 
diff --git a/storage-daemon/qapi/qapi-schema.json b/storage-daemon/qapi/qapi-schema.json
index 6100d1f0c9..c6ad5ae1e3 100644
--- a/storage-daemon/qapi/qapi-schema.json
+++ b/storage-daemon/qapi/qapi-schema.json
@@ -16,6 +16,7 @@
 { 'include': '../../qapi/pragma.json' }
 
 { 'include': '../../qapi/block-core.json' }
+{ 'include': '../../qapi/block-export.json' }
 { 'include': '../../qapi/char.json' }
 { 'include': '../../qapi/common.json' }
 { 'include': '../../qapi/control.json' }
diff --git a/storage-daemon/qemu-storage-daemon.c b/storage-daemon/qemu-storage-daemon.c
index 7e9b0e0d3f..1ae1cda481 100644
--- a/storage-daemon/qemu-storage-daemon.c
+++ b/storage-daemon/qemu-storage-daemon.c
@@ -35,8 +35,8 @@
 #include "monitor/monitor-internal.h"
 
 #include "qapi/error.h"
-#include "qapi/qapi-visit-block.h"
 #include "qapi/qapi-visit-block-core.h"
+#include "qapi/qapi-visit-block-export.h"
 #include "qapi/qapi-visit-control.h"
 #include "qapi/qmp/qdict.h"
 #include "qapi/qmp/qstring.h"
@@ -92,7 +92,7 @@ static void help(void)
 "  --chardev <options>    configure a character device backend\n"
 "                         (see the qemu(1) man page for possible options)\n"
 "\n"
-"  --export [type=]nbd,device=<node-name>[,name=<export-name>]\n"
+"  --export [type=]nbd,id=<id>,node-name=<node-name>[,name=<export-name>]\n"
 "           [,writable=on|off][,bitmap=<name>]\n"
 "                         export the specified block node over NBD\n"
 "                         (requires --nbd-server)\n"
@@ -101,9 +101,9 @@ static void help(void)
 "                         configure a QMP monitor\n"
 "\n"
 "  --nbd-server addr.type=inet,addr.host=<host>,addr.port=<port>\n"
-"               [,tls-creds=<id>][,tls-authz=<id>]\n"
+"               [,tls-creds=<id>][,tls-authz=<id>][,max-connections=<n>]\n"
 "  --nbd-server addr.type=unix,addr.path=<path>\n"
-"               [,tls-creds=<id>][,tls-authz=<id>]\n"
+"               [,tls-creds=<id>][,tls-authz=<id>][,max-connections=<n>]\n"
 "                         start an NBD server for exporting block nodes\n"
 "\n"
 "  --object help          list object types that can be added\n"
@@ -150,17 +150,6 @@ static void init_qmp_commands(void)
                          qmp_marshal_qmp_capabilities, QCO_ALLOW_PRECONFIG);
 }
 
-static void init_export(BlockExport *export, Error **errp)
-{
-    switch (export->type) {
-    case BLOCK_EXPORT_TYPE_NBD:
-        qmp_nbd_server_add(&export->u.nbd, errp);
-        break;
-    default:
-        g_assert_not_reached();
-    }
-}
-
 static void process_options(int argc, char *argv[])
 {
     int c;
@@ -235,14 +224,14 @@ static void process_options(int argc, char *argv[])
         case OPTION_EXPORT:
             {
                 Visitor *v;
-                BlockExport *export;
+                BlockExportOptions *export;
 
                 v = qobject_input_visitor_new_str(optarg, "type", &error_fatal);
-                visit_type_BlockExport(v, NULL, &export, &error_fatal);
+                visit_type_BlockExportOptions(v, NULL, &export, &error_fatal);
                 visit_free(v);
 
-                init_export(export, &error_fatal);
-                qapi_free_BlockExport(export);
+                qmp_block_export_add(export, &error_fatal);
+                qapi_free_BlockExportOptions(export);
                 break;
             }
         case OPTION_MONITOR:
diff --git a/tests/check-block.sh b/tests/check-block.sh
index a5a69060e1..f6b1bda7b9 100755
--- a/tests/check-block.sh
+++ b/tests/check-block.sh
@@ -46,6 +46,11 @@ if ! command -v bash >/dev/null 2>&1 ; then
     exit 0
 fi
 
+if LANG=C bash --version | grep -q 'GNU bash, version [123]' ; then
+    echo "bash version too old ==> Not running the qemu-iotests."
+    exit 0
+fi
+
 if ! (sed --version | grep 'GNU sed') > /dev/null 2>&1 ; then
     if ! command -v gsed >/dev/null 2>&1; then
         echo "GNU sed not available ==> Not running the qemu-iotests."
diff --git a/tests/qemu-iotests/140 b/tests/qemu-iotests/140
index 8d2ce5d9e3..309b177e77 100755
--- a/tests/qemu-iotests/140
+++ b/tests/qemu-iotests/140
@@ -81,10 +81,17 @@ $QEMU_IO_PROG -f raw -r -c 'read -P 42 0 64k' \
     "nbd+unix:///drv?socket=$SOCK_DIR/nbd" 2>&1 \
     | _filter_qemu_io | _filter_nbd
 
+# The order of 'return' and the BLOCK_EXPORT_DELETED event is undefined. Just
+# wait until we've twice seen one of them. Filter the 'return' line out so that
+# the output is defined.
 _send_qemu_cmd $QEMU_HANDLE \
     "{ 'execute': 'eject',
        'arguments': { 'device': 'drv' }}" \
-    'return'
+    'return\|BLOCK_EXPORT_DELETED' |
+    grep -v 'return'
+
+_send_qemu_cmd $QEMU_HANDLE '' 'return\|BLOCK_EXPORT_DELETED' |
+    grep -v 'return'
 
 $QEMU_IO_PROG -f raw -r -c close \
     "nbd+unix:///drv?socket=$SOCK_DIR/nbd" 2>&1 \
diff --git a/tests/qemu-iotests/140.out b/tests/qemu-iotests/140.out
index 86b985da75..62d9c3ab3c 100644
--- a/tests/qemu-iotests/140.out
+++ b/tests/qemu-iotests/140.out
@@ -11,7 +11,7 @@ wrote 65536/65536 bytes at offset 0
 read 65536/65536 bytes at offset 0
 64 KiB, X ops; XX:XX:XX.X (XXX YYY/sec and XXX ops/sec)
 { 'execute': 'eject', 'arguments': { 'device': 'drv' }}
-{"return": {}}
+{"timestamp": {"seconds":  TIMESTAMP, "microseconds":  TIMESTAMP}, "event": "BLOCK_EXPORT_DELETED", "data": {"id": "drv"}}
 qemu-io: can't open device nbd+unix:///drv?socket=SOCK_DIR/nbd: Requested export not available
 server reported: export 'drv' not present
 { 'execute': 'quit' }
diff --git a/tests/qemu-iotests/223.out b/tests/qemu-iotests/223.out
index e1eaaedb55..f6eac23f04 100644
--- a/tests/qemu-iotests/223.out
+++ b/tests/qemu-iotests/223.out
@@ -45,7 +45,7 @@ exports available: 0
 {"execute":"nbd-server-add", "arguments":{"device":"nosuch"}}
 {"error": {"class": "GenericError", "desc": "Cannot find device=nosuch nor node_name=nosuch"}}
 {"execute":"nbd-server-add", "arguments":{"device":"n"}}
-{"error": {"class": "GenericError", "desc": "NBD server already has export named 'n'"}}
+{"error": {"class": "GenericError", "desc": "Block export id 'n' is already in use"}}
 {"execute":"nbd-server-add", "arguments":{"device":"n", "name":"n2", "bitmap":"b2"}}
 {"error": {"class": "GenericError", "desc": "Enabled bitmap 'b2' incompatible with readonly export"}}
 {"execute":"nbd-server-add", "arguments":{"device":"n", "name":"n2", "bitmap":"b3"}}
@@ -102,8 +102,10 @@ read 2097152/2097152 bytes at offset 2097152
 {"execute":"nbd-server-remove", "arguments":{"name":"n"}}
 {"return": {}}
 {"execute":"nbd-server-remove", "arguments":{"name":"n2"}}
+{"timestamp": {"seconds":  TIMESTAMP, "microseconds":  TIMESTAMP}, "event": "BLOCK_EXPORT_DELETED", "data": {"id": "n"}}
 {"return": {}}
 {"execute":"nbd-server-remove", "arguments":{"name":"n2"}}
+{"timestamp": {"seconds":  TIMESTAMP, "microseconds":  TIMESTAMP}, "event": "BLOCK_EXPORT_DELETED", "data": {"id": "n2"}}
 {"error": {"class": "GenericError", "desc": "Export 'n2' is not found"}}
 {"execute":"nbd-server-stop"}
 {"return": {}}
@@ -126,7 +128,7 @@ exports available: 0
 {"execute":"nbd-server-add", "arguments":{"device":"nosuch"}}
 {"error": {"class": "GenericError", "desc": "Cannot find device=nosuch nor node_name=nosuch"}}
 {"execute":"nbd-server-add", "arguments":{"device":"n"}}
-{"error": {"class": "GenericError", "desc": "NBD server already has export named 'n'"}}
+{"error": {"class": "GenericError", "desc": "Block export id 'n' is already in use"}}
 {"execute":"nbd-server-add", "arguments":{"device":"n", "name":"n2", "bitmap":"b2"}}
 {"error": {"class": "GenericError", "desc": "Enabled bitmap 'b2' incompatible with readonly export"}}
 {"execute":"nbd-server-add", "arguments":{"device":"n", "name":"n2", "bitmap":"b3"}}
@@ -183,8 +185,10 @@ read 2097152/2097152 bytes at offset 2097152
 {"execute":"nbd-server-remove", "arguments":{"name":"n"}}
 {"return": {}}
 {"execute":"nbd-server-remove", "arguments":{"name":"n2"}}
+{"timestamp": {"seconds":  TIMESTAMP, "microseconds":  TIMESTAMP}, "event": "BLOCK_EXPORT_DELETED", "data": {"id": "n"}}
 {"return": {}}
 {"execute":"nbd-server-remove", "arguments":{"name":"n2"}}
+{"timestamp": {"seconds":  TIMESTAMP, "microseconds":  TIMESTAMP}, "event": "BLOCK_EXPORT_DELETED", "data": {"id": "n2"}}
 {"error": {"class": "GenericError", "desc": "Export 'n2' is not found"}}
 {"execute":"nbd-server-stop"}
 {"return": {}}
diff --git a/tests/qemu-iotests/307 b/tests/qemu-iotests/307
new file mode 100755
index 0000000000..de7c25fcfc
--- /dev/null
+++ b/tests/qemu-iotests/307
@@ -0,0 +1,132 @@
+#!/usr/bin/env python3
+#
+# Copyright (C) 2020 Red Hat, Inc.
+#
+# This program is free software; you can redistribute it and/or modify
+# it under the terms of the GNU General Public License as published by
+# the Free Software Foundation; either version 2 of the License, or
+# (at your option) any later version.
+#
+# This program is distributed in the hope that it will be useful,
+# but WITHOUT ANY WARRANTY; without even the implied warranty of
+# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+# GNU General Public License for more details.
+#
+# You should have received a copy of the GNU General Public License
+# along with this program.  If not, see <http://www.gnu.org/licenses/>.
+#
+# Creator/Owner: Kevin Wolf <kwolf@redhat.com>
+#
+# Test the block export QAPI interfaces
+
+import iotests
+import os
+
+# Need a writable image format (but not vpc, which rounds the image size, nor
+# luks which requires special command lines)
+iotests.script_initialize(
+    supported_fmts=['generic'],
+    unsupported_fmts=['luks', 'vpc'],
+    supported_platforms=['linux'],
+)
+
+with iotests.FilePath('image') as img, \
+     iotests.FilePath('socket', base_dir=iotests.sock_dir) as socket, \
+     iotests.VM() as vm:
+
+    iotests.qemu_img('create', '-f', iotests.imgfmt, img, '64M')
+    iotests.qemu_io_log('-f', iotests.imgfmt, '-c', 'write -P 0x11 0 4k', img)
+
+    iotests.log('=== Launch VM ===')
+
+    virtio_scsi_device = iotests.get_virtio_scsi_device()
+
+    vm.add_object('iothread,id=iothread0')
+    vm.add_blockdev(f'file,filename={img},node-name=file')
+    vm.add_blockdev(f'{iotests.imgfmt},file=file,node-name=fmt')
+    vm.add_blockdev('raw,file=file,node-name=ro,read-only=on')
+    vm.add_device(f'id=scsi0,driver={virtio_scsi_device},iothread=iothread0')
+    vm.launch()
+
+    vm.qmp_log('nbd-server-start',
+               addr={'type': 'unix', 'data': {'path': socket}},
+               filters=(iotests.filter_qmp_testfiles, ))
+    vm.qmp_log('query-block-exports')
+
+    iotests.log('\n=== Create a read-only NBD export ===')
+
+    vm.qmp_log('block-export-add', id='export0', type='nbd', node_name='fmt')
+    vm.qmp_log('query-block-exports')
+
+    iotests.qemu_nbd_list_log('-k', socket)
+
+    iotests.log('\n=== Try a few invalid things ===')
+
+    vm.qmp_log('block-export-add', id='#invalid', type='nbd', node_name='fmt')
+    vm.qmp_log('block-export-add', id='export0', type='nbd', node_name='fmt')
+    vm.qmp_log('block-export-add', id='export1', type='nbd', node_name='ro',
+               writable=True)
+    vm.qmp_log('block-export-del', id='export1')
+    vm.qmp_log('query-block-exports')
+
+    iotests.log('\n=== Move export to an iothread ===')
+
+    vm.qmp_log('device_add', id='sda', driver='scsi-hd', drive='fmt')
+    vm.qmp_log('query-block-exports')
+    iotests.qemu_nbd_list_log('-k', socket)
+
+    iotests.log('\n=== Add a writable export ===')
+
+    # This fails because share-rw=off
+    vm.qmp_log('block-export-add', id='export1', type='nbd', node_name='fmt',
+               name='export1', writable=True, writethrough=True,
+               description='This is the writable second export')
+
+    vm.qmp_log('device_del', id='sda')
+    event = vm.event_wait(name="DEVICE_DELETED",
+                          match={'data': {'device': 'sda'}})
+    iotests.log(event, filters=[iotests.filter_qmp_event])
+    vm.qmp_log('device_add', id='sda', driver='scsi-hd', drive='fmt',
+               share_rw=True)
+
+    # Now it should work
+    vm.qmp_log('block-export-add', id='export1', type='nbd', node_name='fmt',
+               name='export1', writable=True, writethrough=True,
+               description='This is the writable second export')
+
+    vm.qmp_log('query-block-exports')
+    iotests.qemu_nbd_list_log('-k', socket)
+
+    iotests.log('\n=== Connect qemu-io to export1, try removing exports ===')
+
+    nbd_url = f'nbd+unix:///export1?socket={socket}'
+    qemu_io = iotests.QemuIoInteractive('-f', 'raw', nbd_url)
+
+    iotests.log(qemu_io.cmd('read -P 0x11 0 4k'),
+                filters=[iotests.filter_qemu_io])
+    iotests.log(qemu_io.cmd('write -P 0x22 4k 4k'),
+                filters=[iotests.filter_qemu_io])
+
+    vm.qmp_log('block-export-del', id='export1')
+    vm.qmp_log('block-export-del', id='export0')
+    iotests.log(vm.get_qmp_events_filtered())
+    qemu_io.close()
+
+    vm.qmp_log('query-block-exports')
+    iotests.qemu_nbd_list_log('-k', socket)
+
+    iotests.log('\n=== Connect qemu-io again, try force removing ===')
+
+    qemu_io = iotests.QemuIoInteractive('-f', 'raw', nbd_url)
+    vm.qmp_log('block-export-del', id='export1')
+    vm.qmp_log('block-export-del', id='export1', mode='hard')
+
+    # This should fail now
+    iotests.log(qemu_io.cmd('read -P 0x11 0 4k'))
+    qemu_io.close()
+
+    vm.qmp_log('query-block-exports')
+    iotests.qemu_nbd_list_log('-k', socket)
+
+    iotests.log('\n=== Shut down QEMU ===')
+    vm.shutdown()
diff --git a/tests/qemu-iotests/307.out b/tests/qemu-iotests/307.out
new file mode 100644
index 0000000000..daa8ad2da0
--- /dev/null
+++ b/tests/qemu-iotests/307.out
@@ -0,0 +1,124 @@
+wrote 4096/4096 bytes at offset 0
+4 KiB, X ops; XX:XX:XX.X (XXX YYY/sec and XXX ops/sec)
+
+=== Launch VM ===
+{"execute": "nbd-server-start", "arguments": {"addr": {"data": {"path": "SOCK_DIR/PID-socket"}, "type": "unix"}}}
+{"return": {}}
+{"execute": "query-block-exports", "arguments": {}}
+{"return": []}
+
+=== Create a read-only NBD export ===
+{"execute": "block-export-add", "arguments": {"id": "export0", "node-name": "fmt", "type": "nbd"}}
+{"return": {}}
+{"execute": "query-block-exports", "arguments": {}}
+{"return": [{"id": "export0", "node-name": "fmt", "shutting-down": false, "type": "nbd"}]}
+exports available: 1
+ export: 'fmt'
+  size:  67108864
+  flags: 0x58f ( readonly flush fua df multi cache )
+  min block: XXX
+  opt block: XXX
+  max block: XXX
+  available meta contexts: 1
+   base:allocation
+
+
+=== Try a few invalid things ===
+{"execute": "block-export-add", "arguments": {"id": "#invalid", "node-name": "fmt", "type": "nbd"}}
+{"error": {"class": "GenericError", "desc": "Invalid block export id"}}
+{"execute": "block-export-add", "arguments": {"id": "export0", "node-name": "fmt", "type": "nbd"}}
+{"error": {"class": "GenericError", "desc": "Block export id 'export0' is already in use"}}
+{"execute": "block-export-add", "arguments": {"id": "export1", "node-name": "ro", "type": "nbd", "writable": true}}
+{"error": {"class": "GenericError", "desc": "Cannot export read-only node as writable"}}
+{"execute": "block-export-del", "arguments": {"id": "export1"}}
+{"error": {"class": "GenericError", "desc": "Export 'export1' is not found"}}
+{"execute": "query-block-exports", "arguments": {}}
+{"return": [{"id": "export0", "node-name": "fmt", "shutting-down": false, "type": "nbd"}]}
+
+=== Move export to an iothread ===
+{"execute": "device_add", "arguments": {"drive": "fmt", "driver": "scsi-hd", "id": "sda"}}
+{"return": {}}
+{"execute": "query-block-exports", "arguments": {}}
+{"return": [{"id": "export0", "node-name": "fmt", "shutting-down": false, "type": "nbd"}]}
+exports available: 1
+ export: 'fmt'
+  size:  67108864
+  flags: 0x58f ( readonly flush fua df multi cache )
+  min block: XXX
+  opt block: XXX
+  max block: XXX
+  available meta contexts: 1
+   base:allocation
+
+
+=== Add a writable export ===
+{"execute": "block-export-add", "arguments": {"description": "This is the writable second export", "id": "export1", "name": "export1", "node-name": "fmt", "type": "nbd", "writable": true, "writethrough": true}}
+{"error": {"class": "GenericError", "desc": "Conflicts with use by sda as 'root', which does not allow 'write' on fmt"}}
+{"execute": "device_del", "arguments": {"id": "sda"}}
+{"return": {}}
+{"data": {"device": "sda", "path": "/machine/peripheral/sda"}, "event": "DEVICE_DELETED", "timestamp": {"microseconds": "USECS", "seconds": "SECS"}}
+{"execute": "device_add", "arguments": {"drive": "fmt", "driver": "scsi-hd", "id": "sda", "share-rw": true}}
+{"return": {}}
+{"execute": "block-export-add", "arguments": {"description": "This is the writable second export", "id": "export1", "name": "export1", "node-name": "fmt", "type": "nbd", "writable": true, "writethrough": true}}
+{"return": {}}
+{"execute": "query-block-exports", "arguments": {}}
+{"return": [{"id": "export1", "node-name": "fmt", "shutting-down": false, "type": "nbd"}, {"id": "export0", "node-name": "fmt", "shutting-down": false, "type": "nbd"}]}
+exports available: 2
+ export: 'fmt'
+  size:  67108864
+  flags: 0x58f ( readonly flush fua df multi cache )
+  min block: XXX
+  opt block: XXX
+  max block: XXX
+  available meta contexts: 1
+   base:allocation
+ export: 'export1'
+  description: This is the writable second export
+  size:  67108864
+  flags: 0xced ( flush fua trim zeroes df cache fast-zero )
+  min block: XXX
+  opt block: XXX
+  max block: XXX
+  available meta contexts: 1
+   base:allocation
+
+
+=== Connect qemu-io to export1, try removing exports ===
+read 4096/4096 bytes at offset 0
+4 KiB, X ops; XX:XX:XX.X (XXX YYY/sec and XXX ops/sec)
+
+wrote 4096/4096 bytes at offset 4096
+4 KiB, X ops; XX:XX:XX.X (XXX YYY/sec and XXX ops/sec)
+
+{"execute": "block-export-del", "arguments": {"id": "export1"}}
+{"error": {"class": "GenericError", "desc": "export 'export1' still in use"}}
+{"execute": "block-export-del", "arguments": {"id": "export0"}}
+{"return": {}}
+[{"data": {"id": "export0"}, "event": "BLOCK_EXPORT_DELETED", "timestamp": {"microseconds": "USECS", "seconds": "SECS"}}]
+{"execute": "query-block-exports", "arguments": {}}
+{"return": [{"id": "export1", "node-name": "fmt", "shutting-down": false, "type": "nbd"}]}
+exports available: 1
+ export: 'export1'
+  description: This is the writable second export
+  size:  67108864
+  flags: 0xced ( flush fua trim zeroes df cache fast-zero )
+  min block: XXX
+  opt block: XXX
+  max block: XXX
+  available meta contexts: 1
+   base:allocation
+
+
+=== Connect qemu-io again, try force removing ===
+{"execute": "block-export-del", "arguments": {"id": "export1"}}
+{"error": {"class": "GenericError", "desc": "export 'export1' still in use"}}
+{"execute": "block-export-del", "arguments": {"id": "export1", "mode": "hard"}}
+{"return": {}}
+read failed: Input/output error
+
+{"execute": "query-block-exports", "arguments": {}}
+{"return": []}
+exports available: 0
+
+
+=== Shut down QEMU ===
diff --git a/tests/qemu-iotests/group b/tests/qemu-iotests/group
index ff59cfd2d4..9e4f7c0153 100644
--- a/tests/qemu-iotests/group
+++ b/tests/qemu-iotests/group
@@ -314,3 +314,4 @@
 303 rw quick
 304 rw quick
 305 rw quick
+307 rw quick export
diff --git a/tests/qemu-iotests/iotests.py b/tests/qemu-iotests/iotests.py
index f48460480a..f212cec446 100644
--- a/tests/qemu-iotests/iotests.py
+++ b/tests/qemu-iotests/iotests.py
@@ -65,7 +65,8 @@ if os.environ.get('QEMU_IO_OPTIONS_NO_FMT'):
     qemu_io_args_no_fmt += \
         os.environ['QEMU_IO_OPTIONS_NO_FMT'].strip().split(' ')
 
-qemu_nbd_args = [os.environ.get('QEMU_NBD_PROG', 'qemu-nbd')]
+qemu_nbd_prog = os.environ.get('QEMU_NBD_PROG', 'qemu-nbd')
+qemu_nbd_args = [qemu_nbd_prog]
 if os.environ.get('QEMU_NBD_OPTIONS'):
     qemu_nbd_args += os.environ['QEMU_NBD_OPTIONS'].strip().split(' ')
 
@@ -88,21 +89,30 @@ luks_default_secret_object = 'secret,id=keysec0,data=' + \
 luks_default_key_secret_opt = 'key-secret=keysec0'
 
 
-def qemu_img_pipe_and_status(*args: str) -> Tuple[str, int]:
+def qemu_tool_pipe_and_status(tool: str, args: Sequence[str],
+                              connect_stderr: bool = True) -> Tuple[str, int]:
     """
-    Run qemu-img and return both its output and its exit code
+    Run a tool and return both its output and its exit code
     """
-    subp = subprocess.Popen(qemu_img_args + list(args),
+    stderr = subprocess.STDOUT if connect_stderr else None
+    subp = subprocess.Popen(args,
                             stdout=subprocess.PIPE,
-                            stderr=subprocess.STDOUT,
+                            stderr=stderr,
                             universal_newlines=True)
     output = subp.communicate()[0]
     if subp.returncode < 0:
-        sys.stderr.write('qemu-img received signal %i: %s\n'
-                         % (-subp.returncode,
+        sys.stderr.write('%s received signal %i: %s\n'
+                         % (tool, -subp.returncode,
                             ' '.join(qemu_img_args + list(args))))
     return (output, subp.returncode)
 
+def qemu_img_pipe_and_status(*args: str) -> Tuple[str, int]:
+    """
+    Run qemu-img and return both its output and its exit code
+    """
+    full_args = qemu_img_args + list(args)
+    return qemu_tool_pipe_and_status('qemu-img', full_args)
+
 def qemu_img(*args: str) -> int:
     '''Run qemu-img and return the exit code'''
     return qemu_img_pipe_and_status(*args)[1]
@@ -263,19 +273,20 @@ def qemu_nbd(*args):
     '''Run qemu-nbd in daemon mode and return the parent's exit code'''
     return subprocess.call(qemu_nbd_args + ['--fork'] + list(args))
 
-def qemu_nbd_early_pipe(*args):
+def qemu_nbd_early_pipe(*args: str) -> Tuple[int, str]:
     '''Run qemu-nbd in daemon mode and return both the parent's exit code
        and its output in case of an error'''
-    subp = subprocess.Popen(qemu_nbd_args + ['--fork'] + list(args),
-                            stdout=subprocess.PIPE,
-                            universal_newlines=True)
-    output = subp.communicate()[0]
-    if subp.returncode < 0:
-        sys.stderr.write('qemu-nbd received signal %i: %s\n' %
-                         (-subp.returncode,
-                          ' '.join(qemu_nbd_args + ['--fork'] + list(args))))
-
-    return subp.returncode, output if subp.returncode else ''
+    full_args = qemu_nbd_args + ['--fork'] + list(args)
+    output, returncode = qemu_tool_pipe_and_status('qemu-nbd', full_args,
+                                                   connect_stderr=False)
+    return returncode, output if returncode else ''
+
+def qemu_nbd_list_log(*args: str) -> str:
+    '''Run qemu-nbd to list remote exports'''
+    full_args = [qemu_nbd_prog, '-L'] + list(args)
+    output, _ = qemu_tool_pipe_and_status('qemu-nbd', full_args)
+    log(output, filters=[filter_testfiles, filter_nbd_exports])
+    return output
 
 @contextmanager
 def qemu_nbd_popen(*args):
@@ -410,6 +421,9 @@ def filter_qmp_imgfmt(qmsg):
         return value
     return filter_qmp(qmsg, _filter)
 
+def filter_nbd_exports(output: str) -> str:
+    return re.sub(r'((min|opt|max) block): [0-9]+', r'\1: XXX', output)
+
 
 Msg = TypeVar('Msg', Dict[str, Any], List[Any], str)
 
@@ -1048,16 +1062,12 @@ def case_notrun(reason):
 
 def _verify_image_format(supported_fmts: Sequence[str] = (),
                          unsupported_fmts: Sequence[str] = ()) -> None:
-    assert not (supported_fmts and unsupported_fmts)
-
     if 'generic' in supported_fmts and \
             os.environ.get('IMGFMT_GENERIC', 'true') == 'true':
         # similar to
         #   _supported_fmt generic
         # for bash tests
-        if imgfmt == 'luks':
-            verify_working_luks()
-        return
+        supported_fmts = ()
 
     not_sup = supported_fmts and (imgfmt not in supported_fmts)
     if not_sup or (imgfmt in unsupported_fmts):
@@ -1141,20 +1151,14 @@ def verify_working_luks():
     if not working:
         notrun(reason)
 
-def qemu_pipe(*args):
+def qemu_pipe(*args: str) -> str:
     """
     Run qemu with an option to print something and exit (e.g. a help option).
 
     :return: QEMU's stdout output.
     """
-    args = [qemu_prog] + qemu_opts + list(args)
-    subp = subprocess.Popen(args, stdout=subprocess.PIPE,
-                            stderr=subprocess.STDOUT,
-                            universal_newlines=True)
-    output = subp.communicate()[0]
-    if subp.returncode < 0:
-        sys.stderr.write('qemu received signal %i: %s\n' %
-                         (-subp.returncode, ' '.join(args)))
+    full_args = [qemu_prog] + qemu_opts + list(args)
+    output, _ = qemu_tool_pipe_and_status('qemu', full_args)
     return output
 
 def supported_formats(read_only=False):