[Date Prev][Date Next][Thread Prev][Thread Next][Date Index][Thread Index]
[PATCH v7 2/8] block/nbd.c: Add yank feature
|
From: |
Lukas Straub |
|
Subject: |
[PATCH v7 2/8] block/nbd.c: Add yank feature |
|
Date: |
Tue, 4 Aug 2020 10:11:37 +0200 |
Register a yank function which shuts down the socket and sets
s->state = NBD_CLIENT_QUIT. This is the same behaviour as if an
error occured.
Signed-off-by: Lukas Straub <lukasstraub2@web.de>
Acked-by: Stefan Hajnoczi <stefanha@redhat.com>
---
block/nbd.c | 129 ++++++++++++++++++++++++++++++++--------------------
1 file changed, 80 insertions(+), 49 deletions(-)
diff --git a/block/nbd.c b/block/nbd.c
index 7bb881fef4..8632cf5340 100644
--- a/block/nbd.c
+++ b/block/nbd.c
@@ -35,6 +35,7 @@
#include "qemu/option.h"
#include "qemu/cutils.h"
#include "qemu/main-loop.h"
+#include "qemu/atomic.h"
#include "qapi/qapi-visit-sockets.h"
#include "qapi/qmp/qstring.h"
@@ -43,6 +44,8 @@
#include "block/nbd.h"
#include "block/block_int.h"
+#include "qemu/yank.h"
+
#define EN_OPTSTR ":exportname="
#define MAX_NBD_REQUESTS 16
@@ -84,6 +87,8 @@ typedef struct BDRVNBDState {
NBDReply reply;
BlockDriverState *bs;
+ char *yank_name;
+
/* Connection parameters */
uint32_t reconnect_delay;
SocketAddress *saddr;
@@ -93,10 +98,10 @@ typedef struct BDRVNBDState {
char *x_dirty_bitmap;
} BDRVNBDState;
-static QIOChannelSocket *nbd_establish_connection(SocketAddress *saddr,
- Error **errp);
-static int nbd_client_handshake(BlockDriverState *bs, QIOChannelSocket *sioc,
- Error **errp);
+static int nbd_establish_connection(BlockDriverState *bs, SocketAddress *saddr,
+ Error **errp);
+static int nbd_client_handshake(BlockDriverState *bs, Error **errp);
+static void nbd_yank(void *opaque);
static void nbd_clear_bdrvstate(BDRVNBDState *s)
{
@@ -109,17 +114,19 @@ static void nbd_clear_bdrvstate(BDRVNBDState *s)
s->tlscredsid = NULL;
g_free(s->x_dirty_bitmap);
s->x_dirty_bitmap = NULL;
+ g_free(s->yank_name);
+ s->yank_name = NULL;
}
static void nbd_channel_error(BDRVNBDState *s, int ret)
{
if (ret == -EIO) {
- if (s->state == NBD_CLIENT_CONNECTED) {
+ if (atomic_load_acquire(&s->state) == NBD_CLIENT_CONNECTED) {
s->state = s->reconnect_delay ? NBD_CLIENT_CONNECTING_WAIT :
NBD_CLIENT_CONNECTING_NOWAIT;
}
} else {
- if (s->state == NBD_CLIENT_CONNECTED) {
+ if (atomic_load_acquire(&s->state) == NBD_CLIENT_CONNECTED) {
qio_channel_shutdown(s->ioc, QIO_CHANNEL_SHUTDOWN_BOTH, NULL);
}
s->state = NBD_CLIENT_QUIT;
@@ -170,7 +177,7 @@ static void nbd_client_attach_aio_context(BlockDriverState
*bs,
* s->connection_co is either yielded from nbd_receive_reply or from
* nbd_co_reconnect_loop()
*/
- if (s->state == NBD_CLIENT_CONNECTED) {
+ if (atomic_load_acquire(&s->state) == NBD_CLIENT_CONNECTED) {
qio_channel_attach_aio_context(QIO_CHANNEL(s->ioc), new_context);
}
@@ -237,20 +244,20 @@ static void nbd_teardown_connection(BlockDriverState *bs)
static bool nbd_client_connecting(BDRVNBDState *s)
{
- return s->state == NBD_CLIENT_CONNECTING_WAIT ||
- s->state == NBD_CLIENT_CONNECTING_NOWAIT;
+ NBDClientState state = atomic_load_acquire(&s->state);
+ return state == NBD_CLIENT_CONNECTING_WAIT ||
+ state == NBD_CLIENT_CONNECTING_NOWAIT;
}
static bool nbd_client_connecting_wait(BDRVNBDState *s)
{
- return s->state == NBD_CLIENT_CONNECTING_WAIT;
+ return atomic_load_acquire(&s->state) == NBD_CLIENT_CONNECTING_WAIT;
}
static coroutine_fn void nbd_reconnect_attempt(BDRVNBDState *s)
{
int ret;
Error *local_err = NULL;
- QIOChannelSocket *sioc;
if (!nbd_client_connecting(s)) {
return;
@@ -283,21 +290,21 @@ static coroutine_fn void
nbd_reconnect_attempt(BDRVNBDState *s)
/* Finalize previous connection if any */
if (s->ioc) {
nbd_client_detach_aio_context(s->bs);
+ yank_unregister_function(s->yank_name, nbd_yank, s->bs);
object_unref(OBJECT(s->sioc));
s->sioc = NULL;
object_unref(OBJECT(s->ioc));
s->ioc = NULL;
}
- sioc = nbd_establish_connection(s->saddr, &local_err);
- if (!sioc) {
+ if (nbd_establish_connection(s->bs, s->saddr, &local_err) < 0) {
ret = -ECONNREFUSED;
goto out;
}
bdrv_dec_in_flight(s->bs);
- ret = nbd_client_handshake(s->bs, sioc, &local_err);
+ ret = nbd_client_handshake(s->bs, &local_err);
if (s->drained) {
s->wait_drained_end = true;
@@ -334,7 +341,7 @@ static coroutine_fn void nbd_co_reconnect_loop(BDRVNBDState
*s)
nbd_reconnect_attempt(s);
while (nbd_client_connecting(s)) {
- if (s->state == NBD_CLIENT_CONNECTING_WAIT &&
+ if (atomic_load_acquire(&s->state) == NBD_CLIENT_CONNECTING_WAIT &&
qemu_clock_get_ns(QEMU_CLOCK_REALTIME) - start_time_ns > delay_ns)
{
s->state = NBD_CLIENT_CONNECTING_NOWAIT;
@@ -371,7 +378,7 @@ static coroutine_fn void nbd_connection_entry(void *opaque)
int ret = 0;
Error *local_err = NULL;
- while (s->state != NBD_CLIENT_QUIT) {
+ while (atomic_load_acquire(&s->state) != NBD_CLIENT_QUIT) {
/*
* The NBD client can only really be considered idle when it has
* yielded from qio_channel_readv_all_eof(), waiting for data. This is
@@ -386,7 +393,7 @@ static coroutine_fn void nbd_connection_entry(void *opaque)
nbd_co_reconnect_loop(s);
}
- if (s->state != NBD_CLIENT_CONNECTED) {
+ if (atomic_load_acquire(&s->state) != NBD_CLIENT_CONNECTED) {
continue;
}
@@ -441,6 +448,7 @@ static coroutine_fn void nbd_connection_entry(void *opaque)
s->connection_co = NULL;
if (s->ioc) {
nbd_client_detach_aio_context(s->bs);
+ yank_unregister_function(s->yank_name, nbd_yank, s->bs);
object_unref(OBJECT(s->sioc));
s->sioc = NULL;
object_unref(OBJECT(s->ioc));
@@ -465,7 +473,7 @@ static int nbd_co_send_request(BlockDriverState *bs,
qemu_co_queue_wait(&s->free_sema, &s->send_mutex);
}
- if (s->state != NBD_CLIENT_CONNECTED) {
+ if (atomic_load_acquire(&s->state) != NBD_CLIENT_CONNECTED) {
rc = -EIO;
goto err;
}
@@ -492,7 +500,7 @@ static int nbd_co_send_request(BlockDriverState *bs,
if (qiov) {
qio_channel_set_cork(s->ioc, true);
rc = nbd_send_request(s->ioc, request);
- if (rc >= 0 && s->state == NBD_CLIENT_CONNECTED) {
+ if (rc >= 0 && atomic_load_acquire(&s->state) == NBD_CLIENT_CONNECTED)
{
if (qio_channel_writev_all(s->ioc, qiov->iov, qiov->niov,
NULL) < 0) {
rc = -EIO;
@@ -807,7 +815,7 @@ static coroutine_fn int nbd_co_do_receive_one_chunk(
s->requests[i].receiving = true;
qemu_coroutine_yield();
s->requests[i].receiving = false;
- if (s->state != NBD_CLIENT_CONNECTED) {
+ if (atomic_load_acquire(&s->state) != NBD_CLIENT_CONNECTED) {
error_setg(errp, "Connection closed");
return -EIO;
}
@@ -966,7 +974,7 @@ static bool nbd_reply_chunk_iter_receive(BDRVNBDState *s,
NBDReply local_reply;
NBDStructuredReplyChunk *chunk;
Error *local_err = NULL;
- if (s->state != NBD_CLIENT_CONNECTED) {
+ if (atomic_load_acquire(&s->state) != NBD_CLIENT_CONNECTED) {
error_setg(&local_err, "Connection closed");
nbd_iter_channel_error(iter, -EIO, &local_err);
goto break_loop;
@@ -991,7 +999,8 @@ static bool nbd_reply_chunk_iter_receive(BDRVNBDState *s,
}
/* Do not execute the body of NBD_FOREACH_REPLY_CHUNK for simple reply. */
- if (nbd_reply_is_simple(reply) || s->state != NBD_CLIENT_CONNECTED) {
+ if (nbd_reply_is_simple(reply) ||
+ atomic_load_acquire(&s->state) != NBD_CLIENT_CONNECTED) {
goto break_loop;
}
@@ -1423,6 +1432,15 @@ static int nbd_client_reopen_prepare(BDRVReopenState
*state,
return 0;
}
+static void nbd_yank(void *opaque)
+{
+ BlockDriverState *bs = opaque;
+ BDRVNBDState *s = (BDRVNBDState *)bs->opaque;
+
+ atomic_store_release(&s->state, NBD_CLIENT_QUIT);
+ qio_channel_shutdown(QIO_CHANNEL(s->sioc), QIO_CHANNEL_SHUTDOWN_BOTH,
NULL);
+}
+
static void nbd_client_close(BlockDriverState *bs)
{
BDRVNBDState *s = (BDRVNBDState *)bs->opaque;
@@ -1435,52 +1453,53 @@ static void nbd_client_close(BlockDriverState *bs)
nbd_teardown_connection(bs);
}
-static QIOChannelSocket *nbd_establish_connection(SocketAddress *saddr,
- Error **errp)
+static int nbd_establish_connection(BlockDriverState *bs,
+ SocketAddress *saddr,
+ Error **errp)
{
ERRP_GUARD();
- QIOChannelSocket *sioc;
+ BDRVNBDState *s = (BDRVNBDState *)bs->opaque;
- sioc = qio_channel_socket_new();
- qio_channel_set_name(QIO_CHANNEL(sioc), "nbd-client");
+ s->sioc = qio_channel_socket_new();
+ qio_channel_set_name(QIO_CHANNEL(s->sioc), "nbd-client");
+ yank_register_function(s->yank_name, nbd_yank, bs);
- qio_channel_socket_connect_sync(sioc, saddr, errp);
+ qio_channel_socket_connect_sync(s->sioc, saddr, errp);
if (*errp) {
- object_unref(OBJECT(sioc));
- return NULL;
+ yank_unregister_function(s->yank_name, nbd_yank, bs);
+ object_unref(OBJECT(s->sioc));
+ s->sioc = NULL;
+ return -1;
}
- qio_channel_set_delay(QIO_CHANNEL(sioc), false);
+ qio_channel_set_delay(QIO_CHANNEL(s->sioc), false);
- return sioc;
+ return 0;
}
-/* nbd_client_handshake takes ownership on sioc. On failure it is unref'ed. */
-static int nbd_client_handshake(BlockDriverState *bs, QIOChannelSocket *sioc,
- Error **errp)
+/* nbd_client_handshake takes ownership on s->sioc. On failure it's unref'ed.
*/
+static int nbd_client_handshake(BlockDriverState *bs, Error **errp)
{
BDRVNBDState *s = (BDRVNBDState *)bs->opaque;
AioContext *aio_context = bdrv_get_aio_context(bs);
int ret;
trace_nbd_client_handshake(s->export);
-
- s->sioc = sioc;
-
- qio_channel_set_blocking(QIO_CHANNEL(sioc), false, NULL);
- qio_channel_attach_aio_context(QIO_CHANNEL(sioc), aio_context);
+ qio_channel_set_blocking(QIO_CHANNEL(s->sioc), false, NULL);
+ qio_channel_attach_aio_context(QIO_CHANNEL(s->sioc), aio_context);
s->info.request_sizes = true;
s->info.structured_reply = true;
s->info.base_allocation = true;
s->info.x_dirty_bitmap = g_strdup(s->x_dirty_bitmap);
s->info.name = g_strdup(s->export ?: "");
- ret = nbd_receive_negotiate(aio_context, QIO_CHANNEL(sioc), s->tlscreds,
+ ret = nbd_receive_negotiate(aio_context, QIO_CHANNEL(s->sioc), s->tlscreds,
s->hostname, &s->ioc, &s->info, errp);
g_free(s->info.x_dirty_bitmap);
g_free(s->info.name);
if (ret < 0) {
- object_unref(OBJECT(sioc));
+ yank_unregister_function(s->yank_name, nbd_yank, bs);
+ object_unref(OBJECT(s->sioc));
s->sioc = NULL;
return ret;
}
@@ -1508,7 +1527,7 @@ static int nbd_client_handshake(BlockDriverState *bs,
QIOChannelSocket *sioc,
}
if (!s->ioc) {
- s->ioc = QIO_CHANNEL(sioc);
+ s->ioc = QIO_CHANNEL(s->sioc);
object_ref(OBJECT(s->ioc));
}
@@ -1524,9 +1543,10 @@ static int nbd_client_handshake(BlockDriverState *bs,
QIOChannelSocket *sioc,
{
NBDRequest request = { .type = NBD_CMD_DISC };
- nbd_send_request(s->ioc ?: QIO_CHANNEL(sioc), &request);
+ nbd_send_request(s->ioc ?: QIO_CHANNEL(s->sioc), &request);
- object_unref(OBJECT(sioc));
+ yank_unregister_function(s->yank_name, nbd_yank, bs);
+ object_unref(OBJECT(s->sioc));
s->sioc = NULL;
return ret;
@@ -1918,7 +1938,6 @@ static int nbd_open(BlockDriverState *bs, QDict *options,
int flags,
{
int ret;
BDRVNBDState *s = (BDRVNBDState *)bs->opaque;
- QIOChannelSocket *sioc;
ret = nbd_process_options(bs, options, errp);
if (ret < 0) {
@@ -1929,17 +1948,28 @@ static int nbd_open(BlockDriverState *bs, QDict
*options, int flags,
qemu_co_mutex_init(&s->send_mutex);
qemu_co_queue_init(&s->free_sema);
+ s->yank_name = g_strconcat("blockdev:", bs->node_name, NULL);
+ yank_register_instance(s->yank_name, errp);
+ if (*errp) {
+ g_free(s->yank_name);
+ s->yank_name = NULL;
+ return -EEXIST;
+ }
+
/*
* establish TCP connection, return error if it fails
* TODO: Configurable retry-until-timeout behaviour.
*/
- sioc = nbd_establish_connection(s->saddr, errp);
- if (!sioc) {
+ if (nbd_establish_connection(bs, s->saddr, errp) < 0) {
+ yank_unregister_instance(s->yank_name);
+ g_free(s->yank_name);
+ s->yank_name = NULL;
return -ECONNREFUSED;
}
- ret = nbd_client_handshake(bs, sioc, errp);
+ ret = nbd_client_handshake(bs, errp);
if (ret < 0) {
+ yank_unregister_instance(s->yank_name);
nbd_clear_bdrvstate(s);
return ret;
}
@@ -1997,6 +2027,7 @@ static void nbd_close(BlockDriverState *bs)
BDRVNBDState *s = bs->opaque;
nbd_client_close(bs);
+ yank_unregister_instance(s->yank_name);
nbd_clear_bdrvstate(s);
}
--
2.20.1
pgplT36fPKz9H.pgp
Description: OpenPGP digital signature
- [PATCH v7 0/8] Introduce 'yank' oob qmp command to recover from hanging qemu, Lukas Straub, 2020/08/04
- [PATCH v7 2/8] block/nbd.c: Add yank feature,
Lukas Straub <=
- [PATCH v7 3/8] chardev/char-socket.c: Add yank feature, Lukas Straub, 2020/08/04
- [PATCH v7 4/8] migration: Add yank feature, Lukas Straub, 2020/08/04
- [PATCH v7 5/8] io/channel-tls.c: make qio_channel_tls_shutdown thread-safe, Lukas Straub, 2020/08/04
- [PATCH v7 6/8] io: Document thread-safety of qio_channel_shutdown, Lukas Straub, 2020/08/04
- [PATCH v7 7/8] MAINTAINERS: Add myself as maintainer for yank feature, Lukas Straub, 2020/08/04
- [PATCH v7 8/8] tests/test-char.c: Wait for the chardev to connect in char_socket_client_dupid_test, Lukas Straub, 2020/08/04