block/nbd-client.c

   1 /*
   2  * QEMU Block driver for  NBD
   3  *
   4  * Copyright (C) 2016 Red Hat, Inc.
   5  * Copyright (C) 2008 Bull S.A.S.
   6  *     Author: Laurent Vivier <[email protected]>
   7  *
   8  * Some parts:
   9  *    Copyright (C) 2007 Anthony Liguori <[email protected]>
  10  *
  11  * Permission is hereby granted, free of charge, to any person obtaining a copy
  12  * of this software and associated documentation files (the "Software"), to deal
  13  * in the Software without restriction, including without limitation the rights
  14  * to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
  15  * copies of the Software, and to permit persons to whom the Software is
  16  * furnished to do so, subject to the following conditions:
  17  *
  18  * The above copyright notice and this permission notice shall be included in
  19  * all copies or substantial portions of the Software.
  20  *
  21  * THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
  22  * IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
  23  * FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL
  24  * THE AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
  25  * LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
  26  * OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN
  27  * THE SOFTWARE.
  28  */
  29
  30 #include "qemu/osdep.h"
  31 #include "nbd-client.h"
  32
  33 #define HANDLE_TO_INDEX(bs, handle) ((handle) ^ ((uint64_t)(intptr_t)bs))
  34 #define INDEX_TO_HANDLE(bs, index)  ((index)  ^ ((uint64_t)(intptr_t)bs))
  35
  36 static void nbd_recv_coroutines_enter_all(NBDClientSession *s)
  37 {
  38     int i;
  39
  40     for (i = 0; i < MAX_NBD_REQUESTS; i++) {
  41         if (s->recv_coroutine[i]) {
  42             aio_co_wake(s->recv_coroutine[i]);
  43         }
  44     }
  45 }
  46
  47 static void nbd_teardown_connection(BlockDriverState *bs)
  48 {
  49     NBDClientSession *client = nbd_get_client_session(bs);
  50
  51     if (!client->ioc) { /* Already closed */
  52         return;
  53     }
  54
  55     /* finish any pending coroutines */
  56     qio_channel_shutdown(client->ioc,
  57                          QIO_CHANNEL_SHUTDOWN_BOTH,
  58                          NULL);
  59     BDRV_POLL_WHILE(bs, client->read_reply_co);
  60
  61     nbd_client_detach_aio_context(bs);
  62     object_unref(OBJECT(client->sioc));
  63     client->sioc = NULL;
  64     object_unref(OBJECT(client->ioc));
  65     client->ioc = NULL;
  66 }
  67
  68 static coroutine_fn void nbd_read_reply_entry(void *opaque)
  69 {
  70     NBDClientSession *s = opaque;
  71     uint64_t i;
  72     int ret;
  73
  74     for (;;) {
  75         assert(s->reply.handle == 0);
  76         ret = nbd_receive_reply(s->ioc, &s->reply);
  77         if (ret <= 0) {
  78             break;
  79         }
  80
  81         /* There's no need for a mutex on the receive side, because the
  82          * handler acts as a synchronization point and ensures that only
  83          * one coroutine is called until the reply finishes.
  84          */
  85         i = HANDLE_TO_INDEX(s, s->reply.handle);
  86         if (i >= MAX_NBD_REQUESTS || !s->recv_coroutine[i]) {
  87             break;
  88         }
  89
  90         /* We're woken up by the recv_coroutine itself.  Note that there
  91          * is no race between yielding and reentering read_reply_co.  This
  92          * is because:
  93          *
  94          * - if recv_coroutine[i] runs on the same AioContext, it is only
  95          *   entered after we yield
  96          *
  97          * - if recv_coroutine[i] runs on a different AioContext, reentering
  98          *   read_reply_co happens through a bottom half, which can only
  99          *   run after we yield.
 100          */
 101         aio_co_wake(s->recv_coroutine[i]);
 102         qemu_coroutine_yield();
 103     }
 104
 105     nbd_recv_coroutines_enter_all(s);
 106     s->read_reply_co = NULL;
 107 }
 108
 109 static int nbd_co_send_request(BlockDriverState *bs,
 110                                NBDRequest *request,
 111                                QEMUIOVector *qiov)
 112 {
 113     NBDClientSession *s = nbd_get_client_session(bs);
 114     int rc, ret, i;
 115
 116     qemu_co_mutex_lock(&s->send_mutex);
 117
 118     for (i = 0; i < MAX_NBD_REQUESTS; i++) {
 119         if (s->recv_coroutine[i] == NULL) {
 120             s->recv_coroutine[i] = qemu_coroutine_self();
 121             break;
 122         }
 123     }
 124
 125     g_assert(qemu_in_coroutine());
 126     assert(i < MAX_NBD_REQUESTS);
 127     request->handle = INDEX_TO_HANDLE(s, i);
 128
 129     if (!s->ioc) {
 130         qemu_co_mutex_unlock(&s->send_mutex);
 131         return -EPIPE;
 132     }
 133
 134     if (qiov) {
 135         qio_channel_set_cork(s->ioc, true);
 136         rc = nbd_send_request(s->ioc, request);
 137         if (rc >= 0) {
 138             ret = nbd_wr_syncv(s->ioc, qiov->iov, qiov->niov, request->len,
 139                                false);
 140             if (ret != request->len) {
 141                 rc = -EIO;
 142             }
 143         }
 144         qio_channel_set_cork(s->ioc, false);
 145     } else {
 146         rc = nbd_send_request(s->ioc, request);
 147     }
 148     qemu_co_mutex_unlock(&s->send_mutex);
 149     return rc;
 150 }
 151
 152 static void nbd_co_receive_reply(NBDClientSession *s,
 153                                  NBDRequest *request,
 154                                  NBDReply *reply,
 155                                  QEMUIOVector *qiov)
 156 {
 157     int ret;
 158
 159     /* Wait until we're woken up by nbd_read_reply_entry.  */
 160     qemu_coroutine_yield();
 161     *reply = s->reply;
 162     if (reply->handle != request->handle ||
 163         !s->ioc) {
 164         reply->error = EIO;
 165     } else {
 166         if (qiov && reply->error == 0) {
 167             ret = nbd_wr_syncv(s->ioc, qiov->iov, qiov->niov, request->len,
 168                                true);
 169             if (ret != request->len) {
 170                 reply->error = EIO;
 171             }
 172         }
 173
 174         /* Tell the read handler to read another header.  */
 175         s->reply.handle = 0;
 176     }
 177 }
 178
 179 static void nbd_coroutine_start(NBDClientSession *s,
 180                                 NBDRequest *request)
 181 {
 182     /* Poor man semaphore.  The free_sema is locked when no other request
 183      * can be accepted, and unlocked after receiving one reply.  */
 184     if (s->in_flight == MAX_NBD_REQUESTS) {
 185         qemu_co_queue_wait(&s->free_sema, NULL);
 186         assert(s->in_flight < MAX_NBD_REQUESTS);
 187     }
 188     s->in_flight++;
 189
 190     /* s->recv_coroutine[i] is set as soon as we get the send_lock.  */
 191 }
 192
 193 static void nbd_coroutine_end(BlockDriverState *bs,
 194                               NBDRequest *request)
 195 {
 196     NBDClientSession *s = nbd_get_client_session(bs);
 197     int i = HANDLE_TO_INDEX(s, request->handle);
 198
 199     s->recv_coroutine[i] = NULL;
 200     s->in_flight--;
 201     qemu_co_queue_next(&s->free_sema);
 202
 203     /* Kick the read_reply_co to get the next reply.  */
 204     if (s->read_reply_co) {
 205         aio_co_wake(s->read_reply_co);
 206     }
 207 }
 208
 209 int nbd_client_co_preadv(BlockDriverState *bs, uint64_t offset,
 210                          uint64_t bytes, QEMUIOVector *qiov, int flags)
 211 {
 212     NBDClientSession *client = nbd_get_client_session(bs);
 213     NBDRequest request = {
 214         .type = NBD_CMD_READ,
 215         .from = offset,
 216         .len = bytes,
 217     };
 218     NBDReply reply;
 219     ssize_t ret;
 220
 221     assert(bytes <= NBD_MAX_BUFFER_SIZE);
 222     assert(!flags);
 223
 224     nbd_coroutine_start(client, &request);
 225     ret = nbd_co_send_request(bs, &request, NULL);
 226     if (ret < 0) {
 227         reply.error = -ret;
 228     } else {
 229         nbd_co_receive_reply(client, &request, &reply, qiov);
 230     }
 231     nbd_coroutine_end(bs, &request);
 232     return -reply.error;
 233 }
 234
 235 int nbd_client_co_pwritev(BlockDriverState *bs, uint64_t offset,
 236                           uint64_t bytes, QEMUIOVector *qiov, int flags)
 237 {
 238     NBDClientSession *client = nbd_get_client_session(bs);
 239     NBDRequest request = {
 240         .type = NBD_CMD_WRITE,
 241         .from = offset,
 242         .len = bytes,
 243     };
 244     NBDReply reply;
 245     ssize_t ret;
 246
 247     if (flags & BDRV_REQ_FUA) {
 248         assert(client->nbdflags & NBD_FLAG_SEND_FUA);
 249         request.flags |= NBD_CMD_FLAG_FUA;
 250     }
 251
 252     assert(bytes <= NBD_MAX_BUFFER_SIZE);
 253
 254     nbd_coroutine_start(client, &request);
 255     ret = nbd_co_send_request(bs, &request, qiov);
 256     if (ret < 0) {
 257         reply.error = -ret;
 258     } else {
 259         nbd_co_receive_reply(client, &request, &reply, NULL);
 260     }
 261     nbd_coroutine_end(bs, &request);
 262     return -reply.error;
 263 }
 264
 265 int nbd_client_co_pwrite_zeroes(BlockDriverState *bs, int64_t offset,
 266                                 int count, BdrvRequestFlags flags)
 267 {
 268     ssize_t ret;
 269     NBDClientSession *client = nbd_get_client_session(bs);
 270     NBDRequest request = {
 271         .type = NBD_CMD_WRITE_ZEROES,
 272         .from = offset,
 273         .len = count,
 274     };
 275     NBDReply reply;
 276
 277     if (!(client->nbdflags & NBD_FLAG_SEND_WRITE_ZEROES)) {
 278         return -ENOTSUP;
 279     }
 280
 281     if (flags & BDRV_REQ_FUA) {
 282         assert(client->nbdflags & NBD_FLAG_SEND_FUA);
 283         request.flags |= NBD_CMD_FLAG_FUA;
 284     }
 285     if (!(flags & BDRV_REQ_MAY_UNMAP)) {
 286         request.flags |= NBD_CMD_FLAG_NO_HOLE;
 287     }
 288
 289     nbd_coroutine_start(client, &request);
 290     ret = nbd_co_send_request(bs, &request, NULL);
 291     if (ret < 0) {
 292         reply.error = -ret;
 293     } else {
 294         nbd_co_receive_reply(client, &request, &reply, NULL);
 295     }
 296     nbd_coroutine_end(bs, &request);
 297     return -reply.error;
 298 }
 299
 300 int nbd_client_co_flush(BlockDriverState *bs)
 301 {
 302     NBDClientSession *client = nbd_get_client_session(bs);
 303     NBDRequest request = { .type = NBD_CMD_FLUSH };
 304     NBDReply reply;
 305     ssize_t ret;
 306
 307     if (!(client->nbdflags & NBD_FLAG_SEND_FLUSH)) {
 308         return 0;
 309     }
 310
 311     request.from = 0;
 312     request.len = 0;
 313
 314     nbd_coroutine_start(client, &request);
 315     ret = nbd_co_send_request(bs, &request, NULL);
 316     if (ret < 0) {
 317         reply.error = -ret;
 318     } else {
 319         nbd_co_receive_reply(client, &request, &reply, NULL);
 320     }
 321     nbd_coroutine_end(bs, &request);
 322     return -reply.error;
 323 }
 324
 325 int nbd_client_co_pdiscard(BlockDriverState *bs, int64_t offset, int count)
 326 {
 327     NBDClientSession *client = nbd_get_client_session(bs);
 328     NBDRequest request = {
 329         .type = NBD_CMD_TRIM,
 330         .from = offset,
 331         .len = count,
 332     };
 333     NBDReply reply;
 334     ssize_t ret;
 335
 336     if (!(client->nbdflags & NBD_FLAG_SEND_TRIM)) {
 337         return 0;
 338     }
 339
 340     nbd_coroutine_start(client, &request);
 341     ret = nbd_co_send_request(bs, &request, NULL);
 342     if (ret < 0) {
 343         reply.error = -ret;
 344     } else {
 345         nbd_co_receive_reply(client, &request, &reply, NULL);
 346     }
 347     nbd_coroutine_end(bs, &request);
 348     return -reply.error;
 349
 350 }
 351
 352 void nbd_client_detach_aio_context(BlockDriverState *bs)
 353 {
 354     NBDClientSession *client = nbd_get_client_session(bs);
 355     qio_channel_detach_aio_context(QIO_CHANNEL(client->sioc));
 356 }
 357
 358 void nbd_client_attach_aio_context(BlockDriverState *bs,
 359                                    AioContext *new_context)
 360 {
 361     NBDClientSession *client = nbd_get_client_session(bs);
 362     qio_channel_attach_aio_context(QIO_CHANNEL(client->sioc), new_context);
 363     aio_co_schedule(new_context, client->read_reply_co);
 364 }
 365
 366 void nbd_client_close(BlockDriverState *bs)
 367 {
 368     NBDClientSession *client = nbd_get_client_session(bs);
 369     NBDRequest request = { .type = NBD_CMD_DISC };
 370
 371     if (client->ioc == NULL) {
 372         return;
 373     }
 374
 375     nbd_send_request(client->ioc, &request);
 376
 377     nbd_teardown_connection(bs);
 378 }
 379
 380 int nbd_client_init(BlockDriverState *bs,
 381                     QIOChannelSocket *sioc,
 382                     const char *export,
 383                     QCryptoTLSCreds *tlscreds,
 384                     const char *hostname,
 385                     Error **errp)
 386 {
 387     NBDClientSession *client = nbd_get_client_session(bs);
 388     int ret;
 389
 390     /* NBD handshake */
 391     logout("session init %s\n", export);
 392     qio_channel_set_blocking(QIO_CHANNEL(sioc), true, NULL);
 393
 394     ret = nbd_receive_negotiate(QIO_CHANNEL(sioc), export,
 395                                 &client->nbdflags,
 396                                 tlscreds, hostname,
 397                                 &client->ioc,
 398                                 &client->size, errp);
 399     if (ret < 0) {
 400         logout("Failed to negotiate with the NBD server\n");
 401         return ret;
 402     }
 403     if (client->nbdflags & NBD_FLAG_SEND_FUA) {
 404         bs->supported_write_flags = BDRV_REQ_FUA;
 405         bs->supported_zero_flags |= BDRV_REQ_FUA;
 406     }
 407     if (client->nbdflags & NBD_FLAG_SEND_WRITE_ZEROES) {
 408         bs->supported_zero_flags |= BDRV_REQ_MAY_UNMAP;
 409     }
 410
 411     qemu_co_mutex_init(&client->send_mutex);
 412     qemu_co_queue_init(&client->free_sema);
 413     client->sioc = sioc;
 414     object_ref(OBJECT(client->sioc));
 415
 416     if (!client->ioc) {
 417         client->ioc = QIO_CHANNEL(sioc);
 418         object_ref(OBJECT(client->ioc));
 419     }
 420
 421     /* Now that we're connected, set the socket to be non-blocking and
 422      * kick the reply mechanism.  */
 423     qio_channel_set_blocking(QIO_CHANNEL(sioc), false, NULL);
 424     client->read_reply_co = qemu_coroutine_create(nbd_read_reply_entry, client);
 425     nbd_client_attach_aio_context(bs, bdrv_get_aio_context(bs));
 426
 427     logout("Established connection with NBD server\n");
 428     return 0;
 429 }