mirrors/qemu
1
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

nbd patches for 2019-08-15

Browse Source 
- Addition of InetSocketAddress keep-alive
 - Addition of BDRV_REQ_PREFETCH for more efficient copy-on-read
 - Initial refactoring in preparation of NBD reconnect
 -----BEGIN PGP SIGNATURE-----
 
 iQEcBAABCAAGBQJdVaRZAAoJEKeha0olJ0NqrGoIAJSvVLMDeWZIkHr3CQ5AbMHy
 6IHUntBwv4PEHw0FyyDU7lLgEWubTwe/7RfvyJ69kQYSJLjvHa3KEic0aa7SOETK
 hGUlSoIFHEugi+XDcYyy9EG+ItUR7jnunkwomxvFRm4XzjEHFO9ck8fOS+uq/23e
 LGDHwdoZI6vawUPftbBuRAlB3egCEcBtTWXYMk8lm3MXHOHL7O18DRkfWvwcHfl6
 mNIKgTVMtl1gYoJznCUmC5VLHL4jQy+kSNXnyHBQOEEvTcORu0EztJS81H+BODni
 sxa9seem7JL9NLUTmkJsbGfSM6RKdfypX34oik9yakqUnXRrlxkxI+IX26XfdQ4=
 =2MAO
 -----END PGP SIGNATURE-----

Merge remote-tracking branch 'remotes/ericb/tags/pull-nbd-2019-08-15' into staging

nbd patches for 2019-08-15

- Addition of InetSocketAddress keep-alive
- Addition of BDRV_REQ_PREFETCH for more efficient copy-on-read
- Initial refactoring in preparation of NBD reconnect

# gpg: Signature made Thu 15 Aug 2019 19:28:41 BST
# gpg:                using RSA key A7A16B4A2527436A
# gpg: Good signature from "Eric Blake <eblake@redhat.com>" [full]
# gpg:                 aka "Eric Blake (Free Software Programmer) <ebb9@byu.net>" [full]
# gpg:                 aka "[jpeg image of size 6874]" [full]
# Primary key fingerprint: 71C2 CC22 B1C4 6029 27D2  F3AA A7A1 6B4A 2527 436A

* remotes/ericb/tags/pull-nbd-2019-08-15:
  block/nbd: refactor nbd connection parameters
  block/nbd: add cmdline and qapi parameter reconnect-delay
  block/nbd: move from quit to state
  block/nbd: use non-blocking io channel for nbd negotiation
  block/nbd: split connection_co start out of nbd_client_connect
  nbd: improve CMD_CACHE: use BDRV_REQ_PREFETCH
  block/stream: use BDRV_REQ_PREFETCH
  block: implement BDRV_REQ_PREFETCH
  qapi: Add InetSocketAddress member keep-alive

Signed-off-by: Peter Maydell <peter.maydell@linaro.org>
This commit is contained in:
Peter Maydell 2019-08-16 15:53:37 +01:00
commit c6a2225a5a
11 changed files with 233 additions and 121 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

14
block/io.c
View File

@ -1168,7 +1168,8 @@ bdrv_driver_pwritev_compressed(BlockDriverState *bs, uint64_t offset,
} }
static int coroutine_fn bdrv_co_do_copy_on_readv(BdrvChild *child, static int coroutine_fn bdrv_co_do_copy_on_readv(BdrvChild *child,
int64_t offset, unsigned int bytes, QEMUIOVector *qiov) int64_t offset, unsigned int bytes, QEMUIOVector *qiov,
int flags)
{ {
BlockDriverState *bs = child->bs; BlockDriverState *bs = child->bs;
@ -1279,9 +1280,11 @@ static int coroutine_fn bdrv_co_do_copy_on_readv(BdrvChild *child,
goto err; goto err;
} }
if (!(flags & BDRV_REQ_PREFETCH)) {
qemu_iovec_from_buf(qiov, progress, bounce_buffer + skip_bytes, qemu_iovec_from_buf(qiov, progress, bounce_buffer + skip_bytes,
pnum - skip_bytes); pnum - skip_bytes);
} else { }
} else if (!(flags & BDRV_REQ_PREFETCH)) {
/* Read directly into the destination */ /* Read directly into the destination */
qemu_iovec_init(&local_qiov, qiov->niov); qemu_iovec_init(&local_qiov, qiov->niov);
qemu_iovec_concat(&local_qiov, qiov, progress, pnum - skip_bytes); qemu_iovec_concat(&local_qiov, qiov, progress, pnum - skip_bytes);
@ -1332,7 +1335,8 @@ static int coroutine_fn bdrv_aligned_preadv(BdrvChild *child,
* potential fallback support, if we ever implement any read flags * potential fallback support, if we ever implement any read flags
* to pass through to drivers. For now, there aren't any * to pass through to drivers. For now, there aren't any
* passthrough flags. */ * passthrough flags. */
assert(!(flags & ~(BDRV_REQ_NO_SERIALISING | BDRV_REQ_COPY_ON_READ))); assert(!(flags & ~(BDRV_REQ_NO_SERIALISING | BDRV_REQ_COPY_ON_READ |
BDRV_REQ_PREFETCH)));
/* Handle Copy on Read and associated serialisation */ /* Handle Copy on Read and associated serialisation */
if (flags & BDRV_REQ_COPY_ON_READ) { if (flags & BDRV_REQ_COPY_ON_READ) {
@ -1360,7 +1364,9 @@ static int coroutine_fn bdrv_aligned_preadv(BdrvChild *child,
} }
if (!ret || pnum != bytes) { if (!ret || pnum != bytes) {
ret = bdrv_co_do_copy_on_readv(child, offset, bytes, qiov); ret = bdrv_co_do_copy_on_readv(child, offset, bytes, qiov, flags);
goto out;
} else if (flags & BDRV_REQ_PREFETCH) {
goto out; goto out;
} }
} }

189
block/nbd.c
View File

@ -54,6 +54,11 @@ typedef struct {
bool receiving; /* waiting for connection_co? */ bool receiving; /* waiting for connection_co? */
} NBDClientRequest; } NBDClientRequest;
typedef enum NBDClientState {
NBD_CLIENT_CONNECTED,
NBD_CLIENT_QUIT
} NBDClientState;
typedef struct BDRVNBDState { typedef struct BDRVNBDState {
QIOChannelSocket *sioc; /* The master data channel */ QIOChannelSocket *sioc; /* The master data channel */
QIOChannel *ioc; /* The current I/O channel which may differ (eg TLS) */ QIOChannel *ioc; /* The current I/O channel which may differ (eg TLS) */
@ -63,17 +68,27 @@ typedef struct BDRVNBDState {
CoQueue free_sema; CoQueue free_sema;
Coroutine *connection_co; Coroutine *connection_co;
int in_flight; int in_flight;
NBDClientState state;
NBDClientRequest requests[MAX_NBD_REQUESTS]; NBDClientRequest requests[MAX_NBD_REQUESTS];
NBDReply reply; NBDReply reply;
BlockDriverState *bs; BlockDriverState *bs;
bool quit;
/* For nbd_refresh_filename() */ /* Connection parameters */
uint32_t reconnect_delay;
SocketAddress *saddr; SocketAddress *saddr;
char *export, *tlscredsid; char *export, *tlscredsid;
QCryptoTLSCreds *tlscreds;
const char *hostname;
char *x_dirty_bitmap;
} BDRVNBDState; } BDRVNBDState;
/* @ret will be used for reconnect in future */
static void nbd_channel_error(BDRVNBDState *s, int ret)
{
s->state = NBD_CLIENT_QUIT;
}
static void nbd_recv_coroutines_wake_all(BDRVNBDState *s) static void nbd_recv_coroutines_wake_all(BDRVNBDState *s)
{ {
int i; int i;
@ -152,7 +167,7 @@ static coroutine_fn void nbd_connection_entry(void *opaque)
int ret = 0; int ret = 0;
Error *local_err = NULL; Error *local_err = NULL;
while (!s->quit) { while (s->state != NBD_CLIENT_QUIT) {
/* /*
* The NBD client can only really be considered idle when it has * The NBD client can only really be considered idle when it has
* yielded from qio_channel_readv_all_eof(), waiting for data. This is * yielded from qio_channel_readv_all_eof(), waiting for data. This is
@ -170,6 +185,7 @@ static coroutine_fn void nbd_connection_entry(void *opaque)
error_free(local_err); error_free(local_err);
} }
if (ret <= 0) { if (ret <= 0) {
nbd_channel_error(s, ret ? ret : -EIO);
break; break;
} }
@ -184,6 +200,7 @@ static coroutine_fn void nbd_connection_entry(void *opaque)
!s->requests[i].receiving || !s->requests[i].receiving ||
(nbd_reply_is_structured(&s->reply) && !s->info.structured_reply)) (nbd_reply_is_structured(&s->reply) && !s->info.structured_reply))
{ {
nbd_channel_error(s, -EINVAL);
break; break;
} }
@ -203,7 +220,6 @@ static coroutine_fn void nbd_connection_entry(void *opaque)
qemu_coroutine_yield(); qemu_coroutine_yield();
} }
s->quit = true;
nbd_recv_coroutines_wake_all(s); nbd_recv_coroutines_wake_all(s);
bdrv_dec_in_flight(s->bs); bdrv_dec_in_flight(s->bs);
@ -216,12 +232,18 @@ static int nbd_co_send_request(BlockDriverState *bs,
QEMUIOVector *qiov) QEMUIOVector *qiov)
{ {
BDRVNBDState *s = (BDRVNBDState *)bs->opaque; BDRVNBDState *s = (BDRVNBDState *)bs->opaque;
int rc, i; int rc, i = -1;
qemu_co_mutex_lock(&s->send_mutex); qemu_co_mutex_lock(&s->send_mutex);
while (s->in_flight == MAX_NBD_REQUESTS) { while (s->in_flight == MAX_NBD_REQUESTS) {
qemu_co_queue_wait(&s->free_sema, &s->send_mutex); qemu_co_queue_wait(&s->free_sema, &s->send_mutex);
} }
if (s->state != NBD_CLIENT_CONNECTED) {
rc = -EIO;
goto err;
}
s->in_flight++; s->in_flight++;
for (i = 0; i < MAX_NBD_REQUESTS; i++) { for (i = 0; i < MAX_NBD_REQUESTS; i++) {
@ -239,16 +261,12 @@ static int nbd_co_send_request(BlockDriverState *bs,
request->handle = INDEX_TO_HANDLE(s, i); request->handle = INDEX_TO_HANDLE(s, i);
if (s->quit) {
rc = -EIO;
goto err;
}
assert(s->ioc); assert(s->ioc);
if (qiov) { if (qiov) {
qio_channel_set_cork(s->ioc, true); qio_channel_set_cork(s->ioc, true);
rc = nbd_send_request(s->ioc, request); rc = nbd_send_request(s->ioc, request);
if (rc >= 0 && !s->quit) { if (rc >= 0 && s->state == NBD_CLIENT_CONNECTED) {
if (qio_channel_writev_all(s->ioc, qiov->iov, qiov->niov, if (qio_channel_writev_all(s->ioc, qiov->iov, qiov->niov,
NULL) < 0) { NULL) < 0) {
rc = -EIO; rc = -EIO;
@ -263,9 +281,11 @@ static int nbd_co_send_request(BlockDriverState *bs,
err: err:
if (rc < 0) { if (rc < 0) {
s->quit = true; nbd_channel_error(s, rc);
if (i != -1) {
s->requests[i].coroutine = NULL; s->requests[i].coroutine = NULL;
s->in_flight--; s->in_flight--;
}
qemu_co_queue_next(&s->free_sema); qemu_co_queue_next(&s->free_sema);
} }
qemu_co_mutex_unlock(&s->send_mutex); qemu_co_mutex_unlock(&s->send_mutex);
@ -557,7 +577,7 @@ static coroutine_fn int nbd_co_do_receive_one_chunk(
s->requests[i].receiving = true; s->requests[i].receiving = true;
qemu_coroutine_yield(); qemu_coroutine_yield();
s->requests[i].receiving = false; s->requests[i].receiving = false;
if (s->quit) { if (s->state != NBD_CLIENT_CONNECTED) {
error_setg(errp, "Connection closed"); error_setg(errp, "Connection closed");
return -EIO; return -EIO;
} }
@ -642,7 +662,7 @@ static coroutine_fn int nbd_co_receive_one_chunk(
if (ret < 0) { if (ret < 0) {
memset(reply, 0, sizeof(*reply)); memset(reply, 0, sizeof(*reply));
s->quit = true; nbd_channel_error(s, ret);
} else { } else {
/* For assert at loop start in nbd_connection_entry */ /* For assert at loop start in nbd_connection_entry */
*reply = s->reply; *reply = s->reply;
@ -710,7 +730,7 @@ static bool nbd_reply_chunk_iter_receive(BDRVNBDState *s,
NBDReply local_reply; NBDReply local_reply;
NBDStructuredReplyChunk *chunk; NBDStructuredReplyChunk *chunk;
Error *local_err = NULL; Error *local_err = NULL;
if (s->quit) { if (s->state != NBD_CLIENT_CONNECTED) {
error_setg(&local_err, "Connection closed"); error_setg(&local_err, "Connection closed");
nbd_iter_channel_error(iter, -EIO, &local_err); nbd_iter_channel_error(iter, -EIO, &local_err);
goto break_loop; goto break_loop;
@ -735,7 +755,7 @@ static bool nbd_reply_chunk_iter_receive(BDRVNBDState *s,
} }
/* Do not execute the body of NBD_FOREACH_REPLY_CHUNK for simple reply. */ /* Do not execute the body of NBD_FOREACH_REPLY_CHUNK for simple reply. */
if (nbd_reply_is_simple(reply) || s->quit) { if (nbd_reply_is_simple(reply) || s->state != NBD_CLIENT_CONNECTED) {
goto break_loop; goto break_loop;
} }
@ -809,14 +829,14 @@ static int nbd_co_receive_cmdread_reply(BDRVNBDState *s, uint64_t handle,
ret = nbd_parse_offset_hole_payload(s, &reply.structured, payload, ret = nbd_parse_offset_hole_payload(s, &reply.structured, payload,
offset, qiov, &local_err); offset, qiov, &local_err);
if (ret < 0) { if (ret < 0) {
s->quit = true; nbd_channel_error(s, ret);
nbd_iter_channel_error(&iter, ret, &local_err); nbd_iter_channel_error(&iter, ret, &local_err);
} }
break; break;
default: default:
if (!nbd_reply_type_is_error(chunk->type)) { if (!nbd_reply_type_is_error(chunk->type)) {
/* not allowed reply type */ /* not allowed reply type */
s->quit = true; nbd_channel_error(s, -EINVAL);
error_setg(&local_err, error_setg(&local_err,
"Unexpected reply type: %d (%s) for CMD_READ", "Unexpected reply type: %d (%s) for CMD_READ",
chunk->type, nbd_reply_type_lookup(chunk->type)); chunk->type, nbd_reply_type_lookup(chunk->type));
@ -854,7 +874,7 @@ static int nbd_co_receive_blockstatus_reply(BDRVNBDState *s,
switch (chunk->type) { switch (chunk->type) {
case NBD_REPLY_TYPE_BLOCK_STATUS: case NBD_REPLY_TYPE_BLOCK_STATUS:
if (received) { if (received) {
s->quit = true; nbd_channel_error(s, -EINVAL);
error_setg(&local_err, "Several BLOCK_STATUS chunks in reply"); error_setg(&local_err, "Several BLOCK_STATUS chunks in reply");
nbd_iter_channel_error(&iter, -EINVAL, &local_err); nbd_iter_channel_error(&iter, -EINVAL, &local_err);
} }
@ -864,13 +884,13 @@ static int nbd_co_receive_blockstatus_reply(BDRVNBDState *s,
payload, length, extent, payload, length, extent,
&local_err); &local_err);
if (ret < 0) { if (ret < 0) {
s->quit = true; nbd_channel_error(s, ret);
nbd_iter_channel_error(&iter, ret, &local_err); nbd_iter_channel_error(&iter, ret, &local_err);
} }
break; break;
default: default:
if (!nbd_reply_type_is_error(chunk->type)) { if (!nbd_reply_type_is_error(chunk->type)) {
s->quit = true; nbd_channel_error(s, -EINVAL);
error_setg(&local_err, error_setg(&local_err,
"Unexpected reply type: %d (%s) " "Unexpected reply type: %d (%s) "
"for CMD_BLOCK_STATUS", "for CMD_BLOCK_STATUS",
@ -1167,47 +1187,43 @@ static QIOChannelSocket *nbd_establish_connection(SocketAddress *saddr,
return sioc; return sioc;
} }
static int nbd_client_connect(BlockDriverState *bs, static int nbd_client_connect(BlockDriverState *bs, Error **errp)
SocketAddress *saddr,
const char *export,
QCryptoTLSCreds *tlscreds,
const char *hostname,
const char *x_dirty_bitmap,
Error **errp)
{ {
BDRVNBDState *s = (BDRVNBDState *)bs->opaque; BDRVNBDState *s = (BDRVNBDState *)bs->opaque;
AioContext *aio_context = bdrv_get_aio_context(bs);
int ret; int ret;
/* /*
* establish TCP connection, return error if it fails * establish TCP connection, return error if it fails
* TODO: Configurable retry-until-timeout behaviour. * TODO: Configurable retry-until-timeout behaviour.
*/ */
QIOChannelSocket *sioc = nbd_establish_connection(saddr, errp); QIOChannelSocket *sioc = nbd_establish_connection(s->saddr, errp);
if (!sioc) { if (!sioc) {
return -ECONNREFUSED; return -ECONNREFUSED;
} }
/* NBD handshake */ /* NBD handshake */
trace_nbd_client_connect(export); trace_nbd_client_connect(s->export);
qio_channel_set_blocking(QIO_CHANNEL(sioc), true, NULL); qio_channel_set_blocking(QIO_CHANNEL(sioc), false, NULL);
qio_channel_attach_aio_context(QIO_CHANNEL(sioc), aio_context);
s->info.request_sizes = true; s->info.request_sizes = true;
s->info.structured_reply = true; s->info.structured_reply = true;
s->info.base_allocation = true; s->info.base_allocation = true;
s->info.x_dirty_bitmap = g_strdup(x_dirty_bitmap); s->info.x_dirty_bitmap = g_strdup(s->x_dirty_bitmap);
s->info.name = g_strdup(export ?: ""); s->info.name = g_strdup(s->export ?: "");
ret = nbd_receive_negotiate(QIO_CHANNEL(sioc), tlscreds, hostname, ret = nbd_receive_negotiate(aio_context, QIO_CHANNEL(sioc), s->tlscreds,
&s->ioc, &s->info, errp); s->hostname, &s->ioc, &s->info, errp);
g_free(s->info.x_dirty_bitmap); g_free(s->info.x_dirty_bitmap);
g_free(s->info.name); g_free(s->info.name);
if (ret < 0) { if (ret < 0) {
object_unref(OBJECT(sioc)); object_unref(OBJECT(sioc));
return ret; return ret;
} }
if (x_dirty_bitmap && !s->info.base_allocation) { if (s->x_dirty_bitmap && !s->info.base_allocation) {
error_setg(errp, "requested x-dirty-bitmap %s not found", error_setg(errp, "requested x-dirty-bitmap %s not found",
x_dirty_bitmap); s->x_dirty_bitmap);
ret = -EINVAL; ret = -EINVAL;
goto fail; goto fail;
} }
@ -1232,24 +1248,14 @@ static int nbd_client_connect(BlockDriverState *bs,
object_ref(OBJECT(s->ioc)); object_ref(OBJECT(s->ioc));
} }
/* trace_nbd_client_connect_success(s->export);
* Now that we're connected, set the socket to be non-blocking and
* kick the reply mechanism.
*/
qio_channel_set_blocking(QIO_CHANNEL(sioc), false, NULL);
s->connection_co = qemu_coroutine_create(nbd_connection_entry, s);
bdrv_inc_in_flight(bs);
nbd_client_attach_aio_context(bs, bdrv_get_aio_context(bs));
trace_nbd_client_connect_success(export);
return 0; return 0;
fail: fail:
/* /*
* We have connected, but must fail for other reasons. The * We have connected, but must fail for other reasons.
* connection is still blocking; send NBD_CMD_DISC as a courtesy * Send NBD_CMD_DISC as a courtesy to the server.
* to the server.
*/ */
{ {
NBDRequest request = { .type = NBD_CMD_DISC }; NBDRequest request = { .type = NBD_CMD_DISC };
@ -1262,23 +1268,9 @@ static int nbd_client_connect(BlockDriverState *bs,
} }
} }
static int nbd_client_init(BlockDriverState *bs, /*
SocketAddress *saddr, * Parse nbd_open options
const char *export, */
QCryptoTLSCreds *tlscreds,
const char *hostname,
const char *x_dirty_bitmap,
Error **errp)
{
BDRVNBDState *s = (BDRVNBDState *)bs->opaque;
s->bs = bs;
qemu_co_mutex_init(&s->send_mutex);
qemu_co_queue_init(&s->free_sema);
return nbd_client_connect(bs, saddr, export, tlscreds, hostname,
x_dirty_bitmap, errp);
}
static int nbd_parse_uri(const char *filename, QDict *options) static int nbd_parse_uri(const char *filename, QDict *options)
{ {
@ -1583,18 +1575,27 @@ static QemuOptsList nbd_runtime_opts = {
.help = "experimental: expose named dirty bitmap in place of " .help = "experimental: expose named dirty bitmap in place of "
"block status", "block status",
}, },
{
.name = "reconnect-delay",
.type = QEMU_OPT_NUMBER,
.help = "On an unexpected disconnect, the nbd client tries to "
"connect again until succeeding or encountering a serious "
"error. During the first @reconnect-delay seconds, all "
"requests are paused and will be rerun on a successful "
"reconnect. After that time, any delayed requests and all "
"future requests before a successful reconnect will "
"immediately fail. Default 0",
},
{ /* end of list */ } { /* end of list */ }
}, },
}; };
static int nbd_open(BlockDriverState *bs, QDict *options, int flags, static int nbd_process_options(BlockDriverState *bs, QDict *options,
Error **errp) Error **errp)
{ {
BDRVNBDState *s = bs->opaque; BDRVNBDState *s = bs->opaque;
QemuOpts *opts = NULL; QemuOpts *opts;
Error *local_err = NULL; Error *local_err = NULL;
QCryptoTLSCreds *tlscreds = NULL;
const char *hostname = NULL;
int ret = -EINVAL; int ret = -EINVAL;
opts = qemu_opts_create(&nbd_runtime_opts, NULL, 0, &error_abort); opts = qemu_opts_create(&nbd_runtime_opts, NULL, 0, &error_abort);
@ -1619,8 +1620,8 @@ static int nbd_open(BlockDriverState *bs, QDict *options, int flags,
s->tlscredsid = g_strdup(qemu_opt_get(opts, "tls-creds")); s->tlscredsid = g_strdup(qemu_opt_get(opts, "tls-creds"));
if (s->tlscredsid) { if (s->tlscredsid) {
tlscreds = nbd_get_tls_creds(s->tlscredsid, errp); s->tlscreds = nbd_get_tls_creds(s->tlscredsid, errp);
if (!tlscreds) { if (!s->tlscreds) {
goto error; goto error;
} }
@ -1629,18 +1630,17 @@ static int nbd_open(BlockDriverState *bs, QDict *options, int flags,
error_setg(errp, "TLS only supported over IP sockets"); error_setg(errp, "TLS only supported over IP sockets");
goto error; goto error;
} }
hostname = s->saddr->u.inet.host; s->hostname = s->saddr->u.inet.host;
} }
/* NBD handshake */ s->x_dirty_bitmap = g_strdup(qemu_opt_get(opts, "x-dirty-bitmap"));
ret = nbd_client_init(bs, s->saddr, s->export, tlscreds, hostname, s->reconnect_delay = qemu_opt_get_number(opts, "reconnect-delay", 0);
qemu_opt_get(opts, "x-dirty-bitmap"), errp);
ret = 0;
error: error:
if (tlscreds) {
object_unref(OBJECT(tlscreds));
}
if (ret < 0) { if (ret < 0) {
object_unref(OBJECT(s->tlscreds));
qapi_free_SocketAddress(s->saddr); qapi_free_SocketAddress(s->saddr);
g_free(s->export); g_free(s->export);
g_free(s->tlscredsid); g_free(s->tlscredsid);
@ -1649,6 +1649,35 @@ static int nbd_open(BlockDriverState *bs, QDict *options, int flags,
return ret; return ret;
} }
static int nbd_open(BlockDriverState *bs, QDict *options, int flags,
Error **errp)
{
int ret;
BDRVNBDState *s = (BDRVNBDState *)bs->opaque;
ret = nbd_process_options(bs, options, errp);
if (ret < 0) {
return ret;
}
s->bs = bs;
qemu_co_mutex_init(&s->send_mutex);
qemu_co_queue_init(&s->free_sema);
ret = nbd_client_connect(bs, errp);
if (ret < 0) {
return ret;
}
/* successfully connected */
s->state = NBD_CLIENT_CONNECTED;
s->connection_co = qemu_coroutine_create(nbd_connection_entry, s);
bdrv_inc_in_flight(bs);
aio_co_schedule(bdrv_get_aio_context(bs), s->connection_co);
return 0;
}
static int nbd_co_flush(BlockDriverState *bs) static int nbd_co_flush(BlockDriverState *bs)
{ {
return nbd_client_co_flush(bs); return nbd_client_co_flush(bs);
@ -1694,9 +1723,11 @@ static void nbd_close(BlockDriverState *bs)
nbd_client_close(bs); nbd_client_close(bs);
object_unref(OBJECT(s->tlscreds));
qapi_free_SocketAddress(s->saddr); qapi_free_SocketAddress(s->saddr);
g_free(s->export); g_free(s->export);
g_free(s->tlscredsid); g_free(s->tlscredsid);
g_free(s->x_dirty_bitmap);
} }
static int64_t nbd_getlength(BlockDriverState *bs) static int64_t nbd_getlength(BlockDriverState *bs)

24
block/stream.c
View File

@ -22,11 +22,11 @@
enum { enum {
/* /*
* Size of data buffer for populating the image file. This should be large * Maximum chunk size to feed to copy-on-read. This should be
* enough to process multiple clusters in a single call, so that populating * large enough to process multiple clusters in a single call, so
* contiguous regions of the image is efficient. * that populating contiguous regions of the image is efficient.
*/ */
STREAM_BUFFER_SIZE = 512 * 1024, /* in bytes */ STREAM_CHUNK = 512 * 1024, /* in bytes */
}; };
typedef struct StreamBlockJob { typedef struct StreamBlockJob {
@ -39,13 +39,12 @@ typedef struct StreamBlockJob {
} StreamBlockJob; } StreamBlockJob;
static int coroutine_fn stream_populate(BlockBackend *blk, static int coroutine_fn stream_populate(BlockBackend *blk,
int64_t offset, uint64_t bytes, int64_t offset, uint64_t bytes)
void *buf)
{ {
assert(bytes < SIZE_MAX); assert(bytes < SIZE_MAX);
/* Copy-on-read the unallocated clusters */ return blk_co_preadv(blk, offset, bytes, NULL,
return blk_co_pread(blk, offset, bytes, buf, BDRV_REQ_COPY_ON_READ); BDRV_REQ_COPY_ON_READ | BDRV_REQ_PREFETCH);
} }
static void stream_abort(Job *job) static void stream_abort(Job *job)
@ -117,7 +116,6 @@ static int coroutine_fn stream_run(Job *job, Error **errp)
int error = 0; int error = 0;
int ret = 0; int ret = 0;
int64_t n = 0; /* bytes */ int64_t n = 0; /* bytes */
void *buf;
if (bs == s->bottom) { if (bs == s->bottom) {
/* Nothing to stream */ /* Nothing to stream */
@ -130,8 +128,6 @@ static int coroutine_fn stream_run(Job *job, Error **errp)
} }
job_progress_set_remaining(&s->common.job, len); job_progress_set_remaining(&s->common.job, len);
buf = qemu_blockalign(bs, STREAM_BUFFER_SIZE);
/* Turn on copy-on-read for the whole block device so that guest read /* Turn on copy-on-read for the whole block device so that guest read
* requests help us make progress. Only do this when copying the entire * requests help us make progress. Only do this when copying the entire
* backing chain since the copy-on-read operation does not take base into * backing chain since the copy-on-read operation does not take base into
@ -154,7 +150,7 @@ static int coroutine_fn stream_run(Job *job, Error **errp)
copy = false; copy = false;
ret = bdrv_is_allocated(bs, offset, STREAM_BUFFER_SIZE, &n); ret = bdrv_is_allocated(bs, offset, STREAM_CHUNK, &n);
if (ret == 1) { if (ret == 1) {
/* Allocated in the top, no need to copy. */ /* Allocated in the top, no need to copy. */
} else if (ret >= 0) { } else if (ret >= 0) {
@ -171,7 +167,7 @@ static int coroutine_fn stream_run(Job *job, Error **errp)
} }
trace_stream_one_iteration(s, offset, n, ret); trace_stream_one_iteration(s, offset, n, ret);
if (copy) { if (copy) {
ret = stream_populate(blk, offset, n, buf); ret = stream_populate(blk, offset, n);
} }
if (ret < 0) { if (ret < 0) {
BlockErrorAction action = BlockErrorAction action =
@ -202,8 +198,6 @@ static int coroutine_fn stream_run(Job *job, Error **errp)
bdrv_disable_copy_on_read(bs); bdrv_disable_copy_on_read(bs);
} }
qemu_vfree(buf);
/* Do not remove the backing file if an error was there but ignored. */ /* Do not remove the backing file if an error was there but ignored. */
return error; return error;
} }

8
include/block/block.h
View File

@ -87,8 +87,14 @@ typedef enum {
* fallback. */ * fallback. */
BDRV_REQ_NO_FALLBACK = 0x100, BDRV_REQ_NO_FALLBACK = 0x100,
/*
* BDRV_REQ_PREFETCH may be used only together with BDRV_REQ_COPY_ON_READ
* on read request and means that caller doesn't really need data to be
* written to qiov parameter which may be NULL.
*/
BDRV_REQ_PREFETCH = 0x200,
/* Mask of valid flags */ /* Mask of valid flags */
BDRV_REQ_MASK = 0x1ff, BDRV_REQ_MASK = 0x3ff,
} BdrvRequestFlags; } BdrvRequestFlags;
typedef struct BlockSizes { typedef struct BlockSizes {

3
include/block/nbd.h
View File

@ -304,7 +304,8 @@ struct NBDExportInfo {
}; };
typedef struct NBDExportInfo NBDExportInfo; typedef struct NBDExportInfo NBDExportInfo;
int nbd_receive_negotiate(QIOChannel *ioc, QCryptoTLSCreds *tlscreds, int nbd_receive_negotiate(AioContext *aio_context, QIOChannel *ioc,
QCryptoTLSCreds *tlscreds,
const char *hostname, QIOChannel **outioc, const char *hostname, QIOChannel **outioc,
NBDExportInfo *info, Error **errp); NBDExportInfo *info, Error **errp);
void nbd_free_export_list(NBDExportInfo *info, int count); void nbd_free_export_list(NBDExportInfo *info, int count);

16
nbd/client.c
View File

@ -868,7 +868,8 @@ static int nbd_list_meta_contexts(QIOChannel *ioc,
* 2: server is newstyle, but lacks structured replies * 2: server is newstyle, but lacks structured replies
* 3: server is newstyle and set up for structured replies * 3: server is newstyle and set up for structured replies
*/ */
static int nbd_start_negotiate(QIOChannel *ioc, QCryptoTLSCreds *tlscreds, static int nbd_start_negotiate(AioContext *aio_context, QIOChannel *ioc,
QCryptoTLSCreds *tlscreds,
const char *hostname, QIOChannel **outioc, const char *hostname, QIOChannel **outioc,
bool structured_reply, bool *zeroes, bool structured_reply, bool *zeroes,
Error **errp) Error **errp)
@ -935,6 +936,10 @@ static int nbd_start_negotiate(QIOChannel *ioc, QCryptoTLSCreds *tlscreds,
return -EINVAL; return -EINVAL;
} }
ioc = *outioc; ioc = *outioc;
if (aio_context) {
qio_channel_set_blocking(ioc, false, NULL);
qio_channel_attach_aio_context(ioc, aio_context);
}
} else { } else {
error_setg(errp, "Server does not support STARTTLS"); error_setg(errp, "Server does not support STARTTLS");
return -EINVAL; return -EINVAL;
@ -999,7 +1004,8 @@ static int nbd_negotiate_finish_oldstyle(QIOChannel *ioc, NBDExportInfo *info,
* Returns: negative errno: failure talking to server * Returns: negative errno: failure talking to server
* 0: server is connected * 0: server is connected
*/ */
int nbd_receive_negotiate(QIOChannel *ioc, QCryptoTLSCreds *tlscreds, int nbd_receive_negotiate(AioContext *aio_context, QIOChannel *ioc,
QCryptoTLSCreds *tlscreds,
const char *hostname, QIOChannel **outioc, const char *hostname, QIOChannel **outioc,
NBDExportInfo *info, Error **errp) NBDExportInfo *info, Error **errp)
{ {
@ -1010,7 +1016,7 @@ int nbd_receive_negotiate(QIOChannel *ioc, QCryptoTLSCreds *tlscreds,
assert(info->name); assert(info->name);
trace_nbd_receive_negotiate_name(info->name); trace_nbd_receive_negotiate_name(info->name);
result = nbd_start_negotiate(ioc, tlscreds, hostname, outioc, result = nbd_start_negotiate(aio_context, ioc, tlscreds, hostname, outioc,
info->structured_reply, &zeroes, errp); info->structured_reply, &zeroes, errp);
info->structured_reply = false; info->structured_reply = false;
@ -1130,8 +1136,8 @@ int nbd_receive_export_list(QIOChannel *ioc, QCryptoTLSCreds *tlscreds,
QIOChannel *sioc = NULL; QIOChannel *sioc = NULL;
*info = NULL; *info = NULL;
result = nbd_start_negotiate(ioc, tlscreds, hostname, &sioc, true, NULL, result = nbd_start_negotiate(NULL, ioc, tlscreds, hostname, &sioc, true,
errp); NULL, errp);
if (tlscreds && sioc) { if (tlscreds && sioc) {
ioc = sioc; ioc = sioc;
} }

35
nbd/server.c
View File

@ -2105,12 +2105,15 @@ static int nbd_co_receive_request(NBDRequestData *req, NBDRequest *request,
return -EINVAL; return -EINVAL;
} }
if (request->type != NBD_CMD_CACHE) {
req->data = blk_try_blockalign(client->exp->blk, request->len); req->data = blk_try_blockalign(client->exp->blk, request->len);
if (req->data == NULL) { if (req->data == NULL) {
error_setg(errp, "No memory"); error_setg(errp, "No memory");
return -ENOMEM; return -ENOMEM;
} }
} }
}
if (request->type == NBD_CMD_WRITE) { if (request->type == NBD_CMD_WRITE) {
if (nbd_read(client->ioc, req->data, request->len, "CMD_WRITE data", if (nbd_read(client->ioc, req->data, request->len, "CMD_WRITE data",
errp) < 0) errp) < 0)
@ -2195,7 +2198,7 @@ static coroutine_fn int nbd_do_cmd_read(NBDClient *client, NBDRequest *request,
int ret; int ret;
NBDExport *exp = client->exp; NBDExport *exp = client->exp;
assert(request->type == NBD_CMD_READ || request->type == NBD_CMD_CACHE); assert(request->type == NBD_CMD_READ);
/* XXX: NBD Protocol only documents use of FUA with WRITE */ /* XXX: NBD Protocol only documents use of FUA with WRITE */
if (request->flags & NBD_CMD_FLAG_FUA) { if (request->flags & NBD_CMD_FLAG_FUA) {
@ -2207,7 +2210,7 @@ static coroutine_fn int nbd_do_cmd_read(NBDClient *client, NBDRequest *request,
} }
if (client->structured_reply && !(request->flags & NBD_CMD_FLAG_DF) && if (client->structured_reply && !(request->flags & NBD_CMD_FLAG_DF) &&
request->len && request->type != NBD_CMD_CACHE) request->len)
{ {
return nbd_co_send_sparse_read(client, request->handle, request->from, return nbd_co_send_sparse_read(client, request->handle, request->from,
data, request->len, errp); data, request->len, errp);
@ -2215,7 +2218,7 @@ static coroutine_fn int nbd_do_cmd_read(NBDClient *client, NBDRequest *request,
ret = blk_pread(exp->blk, request->from + exp->dev_offset, data, ret = blk_pread(exp->blk, request->from + exp->dev_offset, data,
request->len); request->len);
if (ret < 0 || request->type == NBD_CMD_CACHE) { if (ret < 0) {
return nbd_send_generic_reply(client, request->handle, ret, return nbd_send_generic_reply(client, request->handle, ret,
"reading from file failed", errp); "reading from file failed", errp);
} }
@ -2234,6 +2237,28 @@ static coroutine_fn int nbd_do_cmd_read(NBDClient *client, NBDRequest *request,
} }
} }
/*
* nbd_do_cmd_cache
*
* Handle NBD_CMD_CACHE request.
* Return -errno if sending fails. Other errors are reported directly to the
* client as an error reply.
*/
static coroutine_fn int nbd_do_cmd_cache(NBDClient *client, NBDRequest *request,
Error **errp)
{
int ret;
NBDExport *exp = client->exp;
assert(request->type == NBD_CMD_CACHE);
ret = blk_co_preadv(exp->blk, request->from + exp->dev_offset, request->len,
NULL, BDRV_REQ_COPY_ON_READ | BDRV_REQ_PREFETCH);
return nbd_send_generic_reply(client, request->handle, ret,
"caching data failed", errp);
}
/* Handle NBD request. /* Handle NBD request.
* Return -errno if sending fails. Other errors are reported directly to the * Return -errno if sending fails. Other errors are reported directly to the
* client as an error reply. */ * client as an error reply. */
@ -2247,8 +2272,10 @@ static coroutine_fn int nbd_handle_request(NBDClient *client,
char *msg; char *msg;
switch (request->type) { switch (request->type) {
case NBD_CMD_READ:
case NBD_CMD_CACHE: case NBD_CMD_CACHE:
return nbd_do_cmd_cache(client, request, errp);
case NBD_CMD_READ:
return nbd_do_cmd_read(client, request, data, errp); return nbd_do_cmd_read(client, request, data, errp);
case NBD_CMD_WRITE: case NBD_CMD_WRITE:

11
qapi/block-core.json
View File

@ -3860,13 +3860,22 @@
# traditional "base:allocation" block status (see # traditional "base:allocation" block status (see
# NBD_OPT_LIST_META_CONTEXT in the NBD protocol) (since 3.0) # NBD_OPT_LIST_META_CONTEXT in the NBD protocol) (since 3.0)
# #
# @reconnect-delay: On an unexpected disconnect, the nbd client tries to
# connect again until succeeding or encountering a serious
# error. During the first @reconnect-delay seconds, all
# requests are paused and will be rerun on a successful
# reconnect. After that time, any delayed requests and all
# future requests before a successful reconnect will
# immediately fail. Default 0 (Since 4.2)
#
# Since: 2.9 # Since: 2.9
## ##
{ 'struct': 'BlockdevOptionsNbd', { 'struct': 'BlockdevOptionsNbd',
'data': { 'server': 'SocketAddress', 'data': { 'server': 'SocketAddress',
'*export': 'str', '*export': 'str',
'*tls-creds': 'str', '*tls-creds': 'str',
'*x-dirty-bitmap': 'str' } } '*x-dirty-bitmap': 'str',
'*reconnect-delay': 'uint32' } }
## ##
# @BlockdevOptionsRaw: # @BlockdevOptionsRaw:

6
qapi/sockets.json
View File

@ -53,6 +53,9 @@
# #
# @ipv6: whether to accept IPv6 addresses, default try both IPv4 and IPv6 # @ipv6: whether to accept IPv6 addresses, default try both IPv4 and IPv6
# #
# @keep-alive: enable keep-alive when connecting to this socket. Not supported
# for passive sockets. (Since 4.2)
#
# Since: 1.3 # Since: 1.3
## ##
{ 'struct': 'InetSocketAddress', { 'struct': 'InetSocketAddress',
@ -61,7 +64,8 @@
'*numeric': 'bool', '*numeric': 'bool',
'*to': 'uint16', '*to': 'uint16',
'*ipv4': 'bool', '*ipv4': 'bool',
'*ipv6': 'bool' } } '*ipv6': 'bool',
'*keep-alive': 'bool' } }
## ##
# @UnixSocketAddress: # @UnixSocketAddress:

2
qemu-nbd.c
View File

@ -362,7 +362,7 @@ static void *nbd_client_thread(void *arg)
goto out; goto out;
} }
ret = nbd_receive_negotiate(QIO_CHANNEL(sioc), ret = nbd_receive_negotiate(NULL, QIO_CHANNEL(sioc),
NULL, NULL, NULL, &info, &local_error); NULL, NULL, NULL, &info, &local_error);
if (ret < 0) { if (ret < 0) {
if (local_error) { if (local_error) {

28
util/qemu-sockets.c
View File

@ -219,6 +219,12 @@ static int inet_listen_saddr(InetSocketAddress *saddr,
bool socket_created = false; bool socket_created = false;
Error *err = NULL; Error *err = NULL;
if (saddr->keep_alive) {
error_setg(errp, "keep-alive option is not supported for passive "
"sockets");
return -1;
}
memset(&ai,0, sizeof(ai)); memset(&ai,0, sizeof(ai));
ai.ai_flags = AI_PASSIVE; ai.ai_flags = AI_PASSIVE;
if (saddr->has_numeric && saddr->numeric) { if (saddr->has_numeric && saddr->numeric) {
@ -458,6 +464,19 @@ int inet_connect_saddr(InetSocketAddress *saddr, Error **errp)
} }
freeaddrinfo(res); freeaddrinfo(res);
if (saddr->keep_alive) {
int val = 1;
int ret = qemu_setsockopt(sock, SOL_SOCKET, SO_KEEPALIVE,
&val, sizeof(val));
if (ret < 0) {
error_setg_errno(errp, errno, "Unable to set KEEPALIVE");
close(sock);
return -1;
}
}
return sock; return sock;
} }
@ -653,6 +672,15 @@ int inet_parse(InetSocketAddress *addr, const char *str, Error **errp)
} }
addr->has_ipv6 = true; addr->has_ipv6 = true;
} }
begin = strstr(optstr, ",keep-alive");
if (begin) {
if (inet_parse_flag("keep-alive", begin + strlen(",keep-alive"),
&addr->keep_alive, errp) < 0)
{
return -1;
}
addr->has_keep_alive = true;
}
return 0; return 0;
} }