block/nbd-client.c

   1 /*
   2  * QEMU Block driver for  NBD
   3  *
   4  * Copyright (C) 2008 Bull S.A.S.
   5  *     Author: Laurent Vivier <Laurent.Vivier@bull.net>
   6  *
   7  * Some parts:
   8  *    Copyright (C) 2007 Anthony Liguori <anthony@codemonkey.ws>
   9  *
  10  * Permission is hereby granted, free of charge, to any person obtaining a copy
  11  * of this software and associated documentation files (the "Software"), to deal
  12  * in the Software without restriction, including without limitation the rights
  13  * to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
  14  * copies of the Software, and to permit persons to whom the Software is
  15  * furnished to do so, subject to the following conditions:
  16  *
  17  * The above copyright notice and this permission notice shall be included in
  18  * all copies or substantial portions of the Software.
  19  *
  20  * THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
  21  * IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
  22  * FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL
  23  * THE AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
  24  * LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
  25  * OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN
  26  * THE SOFTWARE.
  27  */
  28
  29 #include "nbd-client.h"
  30 #include "qemu/sockets.h"
  31
  32 #define HANDLE_TO_INDEX(bs, handle) ((handle) ^ ((uint64_t)(intptr_t)bs))
  33 #define INDEX_TO_HANDLE(bs, index)  ((index)  ^ ((uint64_t)(intptr_t)bs))
  34
  35 static void nbd_reply_ready(void *opaque)
  36 {
  37     NbdClientSession *s = opaque;
  38     uint64_t i;
  39     int ret;
  40
  41     if (s->reply.handle == 0) {
  42         /* No reply already in flight.  Fetch a header.  It is possible
  43          * that another thread has done the same thing in parallel, so
  44          * the socket is not readable anymore.
  45          */
  46         ret = nbd_receive_reply(s->sock, &s->reply);
  47         if (ret == -EAGAIN) {
  48             return;
  49         }
  50         if (ret < 0) {
  51             s->reply.handle = 0;
  52             goto fail;
  53         }
  54     }
  55
  56     /* There's no need for a mutex on the receive side, because the
  57      * handler acts as a synchronization point and ensures that only
  58      * one coroutine is called until the reply finishes.  */
  59     i = HANDLE_TO_INDEX(s, s->reply.handle);
  60     if (i >= MAX_NBD_REQUESTS) {
  61         goto fail;
  62     }
  63
  64     if (s->recv_coroutine[i]) {
  65         qemu_coroutine_enter(s->recv_coroutine[i], NULL);
  66         return;
  67     }
  68
  69 fail:
  70     for (i = 0; i < MAX_NBD_REQUESTS; i++) {
  71         if (s->recv_coroutine[i]) {
  72             qemu_coroutine_enter(s->recv_coroutine[i], NULL);
  73         }
  74     }
  75 }
  76
  77 static void nbd_restart_write(void *opaque)
  78 {
  79     NbdClientSession *s = opaque;
  80
  81     qemu_coroutine_enter(s->send_coroutine, NULL);
  82 }
  83
  84 static int nbd_co_send_request(NbdClientSession *s,
  85     struct nbd_request *request,
  86     QEMUIOVector *qiov, int offset)
  87 {
  88     int rc, ret;
  89
  90     qemu_co_mutex_lock(&s->send_mutex);
  91     s->send_coroutine = qemu_coroutine_self();
  92     qemu_aio_set_fd_handler(s->sock, nbd_reply_ready, nbd_restart_write, s);
  93     if (qiov) {
  94         if (!s->is_unix) {
  95             socket_set_cork(s->sock, 1);
  96         }
  97         rc = nbd_send_request(s->sock, request);
  98         if (rc >= 0) {
  99             ret = qemu_co_sendv(s->sock, qiov->iov, qiov->niov,
 100                                 offset, request->len);
 101             if (ret != request->len) {
 102                 rc = -EIO;
 103             }
 104         }
 105         if (!s->is_unix) {
 106             socket_set_cork(s->sock, 0);
 107         }
 108     } else {
 109         rc = nbd_send_request(s->sock, request);
 110     }
 111     qemu_aio_set_fd_handler(s->sock, nbd_reply_ready, NULL, s);
 112     s->send_coroutine = NULL;
 113     qemu_co_mutex_unlock(&s->send_mutex);
 114     return rc;
 115 }
 116
 117 static void nbd_co_receive_reply(NbdClientSession *s,
 118     struct nbd_request *request, struct nbd_reply *reply,
 119     QEMUIOVector *qiov, int offset)
 120 {
 121     int ret;
 122
 123     /* Wait until we're woken up by the read handler.  TODO: perhaps
 124      * peek at the next reply and avoid yielding if it's ours?  */
 125     qemu_coroutine_yield();
 126     *reply = s->reply;
 127     if (reply->handle != request->handle) {
 128         reply->error = EIO;
 129     } else {
 130         if (qiov && reply->error == 0) {
 131             ret = qemu_co_recvv(s->sock, qiov->iov, qiov->niov,
 132                                 offset, request->len);
 133             if (ret != request->len) {
 134                 reply->error = EIO;
 135             }
 136         }
 137
 138         /* Tell the read handler to read another header.  */
 139         s->reply.handle = 0;
 140     }
 141 }
 142
 143 static void nbd_coroutine_start(NbdClientSession *s,
 144    struct nbd_request *request)
 145 {
 146     int i;
 147
 148     /* Poor man semaphore.  The free_sema is locked when no other request
 149      * can be accepted, and unlocked after receiving one reply.  */
 150     if (s->in_flight >= MAX_NBD_REQUESTS - 1) {
 151         qemu_co_mutex_lock(&s->free_sema);
 152         assert(s->in_flight < MAX_NBD_REQUESTS);
 153     }
 154     s->in_flight++;
 155
 156     for (i = 0; i < MAX_NBD_REQUESTS; i++) {
 157         if (s->recv_coroutine[i] == NULL) {
 158             s->recv_coroutine[i] = qemu_coroutine_self();
 159             break;
 160         }
 161     }
 162
 163     assert(i < MAX_NBD_REQUESTS);
 164     request->handle = INDEX_TO_HANDLE(s, i);
 165 }
 166
 167 static void nbd_coroutine_end(NbdClientSession *s,
 168     struct nbd_request *request)
 169 {
 170     int i = HANDLE_TO_INDEX(s, request->handle);
 171     s->recv_coroutine[i] = NULL;
 172     if (s->in_flight-- == MAX_NBD_REQUESTS) {
 173         qemu_co_mutex_unlock(&s->free_sema);
 174     }
 175 }
 176
 177 static int nbd_co_readv_1(NbdClientSession *client, int64_t sector_num,
 178                           int nb_sectors, QEMUIOVector *qiov,
 179                           int offset)
 180 {
 181     struct nbd_request request;
 182     struct nbd_reply reply;
 183     ssize_t ret;
 184
 185     request.type = NBD_CMD_READ;
 186     request.from = sector_num * 512;
 187     request.len = nb_sectors * 512;
 188
 189     nbd_coroutine_start(client, &request);
 190     ret = nbd_co_send_request(client, &request, NULL, 0);
 191     if (ret < 0) {
 192         reply.error = -ret;
 193     } else {
 194         nbd_co_receive_reply(client, &request, &reply, qiov, offset);
 195     }
 196     nbd_coroutine_end(client, &request);
 197     return -reply.error;
 198
 199 }
 200
 201 static int nbd_co_writev_1(NbdClientSession *client, int64_t sector_num,
 202                            int nb_sectors, QEMUIOVector *qiov,
 203                            int offset)
 204 {
 205     struct nbd_request request;
 206     struct nbd_reply reply;
 207     ssize_t ret;
 208
 209     request.type = NBD_CMD_WRITE;
 210     if (!bdrv_enable_write_cache(client->bs) &&
 211         (client->nbdflags & NBD_FLAG_SEND_FUA)) {
 212         request.type |= NBD_CMD_FLAG_FUA;
 213     }
 214
 215     request.from = sector_num * 512;
 216     request.len = nb_sectors * 512;
 217
 218     nbd_coroutine_start(client, &request);
 219     ret = nbd_co_send_request(client, &request, qiov, offset);
 220     if (ret < 0) {
 221         reply.error = -ret;
 222     } else {
 223         nbd_co_receive_reply(client, &request, &reply, NULL, 0);
 224     }
 225     nbd_coroutine_end(client, &request);
 226     return -reply.error;
 227 }
 228
 229 /* qemu-nbd has a limit of slightly less than 1M per request.  Try to
 230  * remain aligned to 4K. */
 231 #define NBD_MAX_SECTORS 2040
 232
 233 int nbd_client_session_co_readv(NbdClientSession *client, int64_t sector_num,
 234     int nb_sectors, QEMUIOVector *qiov)
 235 {
 236     int offset = 0;
 237     int ret;
 238     while (nb_sectors > NBD_MAX_SECTORS) {
 239         ret = nbd_co_readv_1(client, sector_num,
 240                              NBD_MAX_SECTORS, qiov, offset);
 241         if (ret < 0) {
 242             return ret;
 243         }
 244         offset += NBD_MAX_SECTORS * 512;
 245         sector_num += NBD_MAX_SECTORS;
 246         nb_sectors -= NBD_MAX_SECTORS;
 247     }
 248     return nbd_co_readv_1(client, sector_num, nb_sectors, qiov, offset);
 249 }
 250
 251 int nbd_client_session_co_writev(NbdClientSession *client, int64_t sector_num,
 252                                  int nb_sectors, QEMUIOVector *qiov)
 253 {
 254     int offset = 0;
 255     int ret;
 256     while (nb_sectors > NBD_MAX_SECTORS) {
 257         ret = nbd_co_writev_1(client, sector_num,
 258                               NBD_MAX_SECTORS, qiov, offset);
 259         if (ret < 0) {
 260             return ret;
 261         }
 262         offset += NBD_MAX_SECTORS * 512;
 263         sector_num += NBD_MAX_SECTORS;
 264         nb_sectors -= NBD_MAX_SECTORS;
 265     }
 266     return nbd_co_writev_1(client, sector_num, nb_sectors, qiov, offset);
 267 }
 268
 269 int nbd_client_session_co_flush(NbdClientSession *client)
 270 {
 271     struct nbd_request request;
 272     struct nbd_reply reply;
 273     ssize_t ret;
 274
 275     if (!(client->nbdflags & NBD_FLAG_SEND_FLUSH)) {
 276         return 0;
 277     }
 278
 279     request.type = NBD_CMD_FLUSH;
 280     if (client->nbdflags & NBD_FLAG_SEND_FUA) {
 281         request.type |= NBD_CMD_FLAG_FUA;
 282     }
 283
 284     request.from = 0;
 285     request.len = 0;
 286
 287     nbd_coroutine_start(client, &request);
 288     ret = nbd_co_send_request(client, &request, NULL, 0);
 289     if (ret < 0) {
 290         reply.error = -ret;
 291     } else {
 292         nbd_co_receive_reply(client, &request, &reply, NULL, 0);
 293     }
 294     nbd_coroutine_end(client, &request);
 295     return -reply.error;
 296 }
 297
 298 int nbd_client_session_co_discard(NbdClientSession *client, int64_t sector_num,
 299     int nb_sectors)
 300 {
 301     struct nbd_request request;
 302     struct nbd_reply reply;
 303     ssize_t ret;
 304
 305     if (!(client->nbdflags & NBD_FLAG_SEND_TRIM)) {
 306         return 0;
 307     }
 308     request.type = NBD_CMD_TRIM;
 309     request.from = sector_num * 512;
 310     request.len = nb_sectors * 512;
 311
 312     nbd_coroutine_start(client, &request);
 313     ret = nbd_co_send_request(client, &request, NULL, 0);
 314     if (ret < 0) {
 315         reply.error = -ret;
 316     } else {
 317         nbd_co_receive_reply(client, &request, &reply, NULL, 0);
 318     }
 319     nbd_coroutine_end(client, &request);
 320     return -reply.error;
 321
 322 }
 323
 324 static void nbd_teardown_connection(NbdClientSession *client)
 325 {
 326     struct nbd_request request;
 327
 328     request.type = NBD_CMD_DISC;
 329     request.from = 0;
 330     request.len = 0;
 331     nbd_send_request(client->sock, &request);
 332
 333     qemu_aio_set_fd_handler(client->sock, NULL, NULL, NULL);
 334     closesocket(client->sock);
 335     client->sock = -1;
 336 }
 337
 338 void nbd_client_session_close(NbdClientSession *client)
 339 {
 340     if (!client->bs) {
 341         return;
 342     }
 343
 344     nbd_teardown_connection(client);
 345     client->bs = NULL;
 346 }
 347
 348 int nbd_client_session_init(NbdClientSession *client, BlockDriverState *bs,
 349     int sock, const char *export)
 350 {
 351     int ret;
 352
 353     /* NBD handshake */
 354     logout("session init %s\n", export);
 355     qemu_set_block(sock);
 356     ret = nbd_receive_negotiate(sock, export,
 357                                 &client->nbdflags, &client->size,
 358                                 &client->blocksize);
 359     if (ret < 0) {
 360         logout("Failed to negotiate with the NBD server\n");
 361         closesocket(sock);
 362         return ret;
 363     }
 364
 365     qemu_co_mutex_init(&client->send_mutex);
 366     qemu_co_mutex_init(&client->free_sema);
 367     client->bs = bs;
 368     client->sock = sock;
 369
 370     /* Now that we're connected, set the socket to be non-blocking and
 371      * kick the reply mechanism.  */
 372     qemu_set_nonblock(sock);
 373     qemu_aio_set_fd_handler(sock, nbd_reply_ready, NULL, client);
 374
 375     logout("Established connection with NBD server\n");
 376     return 0;
 377 }