Merge "migration: do cleanup operation after completion"
[kvmfornfv.git] / qemu / 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_recv_coroutines_enter_all(NbdClientSession *s)
36 {
37     int i;
38
39     for (i = 0; i < MAX_NBD_REQUESTS; i++) {
40         if (s->recv_coroutine[i]) {
41             qemu_coroutine_enter(s->recv_coroutine[i], NULL);
42         }
43     }
44 }
45
46 static void nbd_teardown_connection(BlockDriverState *bs)
47 {
48     NbdClientSession *client = nbd_get_client_session(bs);
49
50     /* finish any pending coroutines */
51     shutdown(client->sock, 2);
52     nbd_recv_coroutines_enter_all(client);
53
54     nbd_client_detach_aio_context(bs);
55     closesocket(client->sock);
56     client->sock = -1;
57 }
58
59 static void nbd_reply_ready(void *opaque)
60 {
61     BlockDriverState *bs = opaque;
62     NbdClientSession *s = nbd_get_client_session(bs);
63     uint64_t i;
64     int ret;
65
66     if (s->reply.handle == 0) {
67         /* No reply already in flight.  Fetch a header.  It is possible
68          * that another thread has done the same thing in parallel, so
69          * the socket is not readable anymore.
70          */
71         ret = nbd_receive_reply(s->sock, &s->reply);
72         if (ret == -EAGAIN) {
73             return;
74         }
75         if (ret < 0) {
76             s->reply.handle = 0;
77             goto fail;
78         }
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     i = HANDLE_TO_INDEX(s, s->reply.handle);
85     if (i >= MAX_NBD_REQUESTS) {
86         goto fail;
87     }
88
89     if (s->recv_coroutine[i]) {
90         qemu_coroutine_enter(s->recv_coroutine[i], NULL);
91         return;
92     }
93
94 fail:
95     nbd_teardown_connection(bs);
96 }
97
98 static void nbd_restart_write(void *opaque)
99 {
100     BlockDriverState *bs = opaque;
101
102     qemu_coroutine_enter(nbd_get_client_session(bs)->send_coroutine, NULL);
103 }
104
105 static int nbd_co_send_request(BlockDriverState *bs,
106                                struct nbd_request *request,
107                                QEMUIOVector *qiov, int offset)
108 {
109     NbdClientSession *s = nbd_get_client_session(bs);
110     AioContext *aio_context;
111     int rc, ret, i;
112
113     qemu_co_mutex_lock(&s->send_mutex);
114
115     for (i = 0; i < MAX_NBD_REQUESTS; i++) {
116         if (s->recv_coroutine[i] == NULL) {
117             s->recv_coroutine[i] = qemu_coroutine_self();
118             break;
119         }
120     }
121
122     assert(i < MAX_NBD_REQUESTS);
123     request->handle = INDEX_TO_HANDLE(s, i);
124     s->send_coroutine = qemu_coroutine_self();
125     aio_context = bdrv_get_aio_context(bs);
126
127     aio_set_fd_handler(aio_context, s->sock,
128                        nbd_reply_ready, nbd_restart_write, bs);
129     if (qiov) {
130         if (!s->is_unix) {
131             socket_set_cork(s->sock, 1);
132         }
133         rc = nbd_send_request(s->sock, request);
134         if (rc >= 0) {
135             ret = qemu_co_sendv(s->sock, qiov->iov, qiov->niov,
136                                 offset, request->len);
137             if (ret != request->len) {
138                 rc = -EIO;
139             }
140         }
141         if (!s->is_unix) {
142             socket_set_cork(s->sock, 0);
143         }
144     } else {
145         rc = nbd_send_request(s->sock, request);
146     }
147     aio_set_fd_handler(aio_context, s->sock, nbd_reply_ready, NULL, bs);
148     s->send_coroutine = NULL;
149     qemu_co_mutex_unlock(&s->send_mutex);
150     return rc;
151 }
152
153 static void nbd_co_receive_reply(NbdClientSession *s,
154     struct nbd_request *request, struct nbd_reply *reply,
155     QEMUIOVector *qiov, int offset)
156 {
157     int ret;
158
159     /* Wait until we're woken up by the read handler.  TODO: perhaps
160      * peek at the next reply and avoid yielding if it's ours?  */
161     qemu_coroutine_yield();
162     *reply = s->reply;
163     if (reply->handle != request->handle) {
164         reply->error = EIO;
165     } else {
166         if (qiov && reply->error == 0) {
167             ret = qemu_co_recvv(s->sock, qiov->iov, qiov->niov,
168                                 offset, request->len);
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    struct nbd_request *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 - 1) {
185         qemu_co_mutex_lock(&s->free_sema);
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(NbdClientSession *s,
194     struct nbd_request *request)
195 {
196     int i = HANDLE_TO_INDEX(s, request->handle);
197     s->recv_coroutine[i] = NULL;
198     if (s->in_flight-- == MAX_NBD_REQUESTS) {
199         qemu_co_mutex_unlock(&s->free_sema);
200     }
201 }
202
203 static int nbd_co_readv_1(BlockDriverState *bs, int64_t sector_num,
204                           int nb_sectors, QEMUIOVector *qiov,
205                           int offset)
206 {
207     NbdClientSession *client = nbd_get_client_session(bs);
208     struct nbd_request request = { .type = NBD_CMD_READ };
209     struct nbd_reply reply;
210     ssize_t ret;
211
212     request.from = sector_num * 512;
213     request.len = nb_sectors * 512;
214
215     nbd_coroutine_start(client, &request);
216     ret = nbd_co_send_request(bs, &request, NULL, 0);
217     if (ret < 0) {
218         reply.error = -ret;
219     } else {
220         nbd_co_receive_reply(client, &request, &reply, qiov, offset);
221     }
222     nbd_coroutine_end(client, &request);
223     return -reply.error;
224
225 }
226
227 static int nbd_co_writev_1(BlockDriverState *bs, int64_t sector_num,
228                            int nb_sectors, QEMUIOVector *qiov,
229                            int offset)
230 {
231     NbdClientSession *client = nbd_get_client_session(bs);
232     struct nbd_request request = { .type = NBD_CMD_WRITE };
233     struct nbd_reply reply;
234     ssize_t ret;
235
236     if (!bdrv_enable_write_cache(bs) &&
237         (client->nbdflags & NBD_FLAG_SEND_FUA)) {
238         request.type |= NBD_CMD_FLAG_FUA;
239     }
240
241     request.from = sector_num * 512;
242     request.len = nb_sectors * 512;
243
244     nbd_coroutine_start(client, &request);
245     ret = nbd_co_send_request(bs, &request, qiov, offset);
246     if (ret < 0) {
247         reply.error = -ret;
248     } else {
249         nbd_co_receive_reply(client, &request, &reply, NULL, 0);
250     }
251     nbd_coroutine_end(client, &request);
252     return -reply.error;
253 }
254
255 /* qemu-nbd has a limit of slightly less than 1M per request.  Try to
256  * remain aligned to 4K. */
257 #define NBD_MAX_SECTORS 2040
258
259 int nbd_client_co_readv(BlockDriverState *bs, int64_t sector_num,
260                         int nb_sectors, QEMUIOVector *qiov)
261 {
262     int offset = 0;
263     int ret;
264     while (nb_sectors > NBD_MAX_SECTORS) {
265         ret = nbd_co_readv_1(bs, sector_num, NBD_MAX_SECTORS, qiov, offset);
266         if (ret < 0) {
267             return ret;
268         }
269         offset += NBD_MAX_SECTORS * 512;
270         sector_num += NBD_MAX_SECTORS;
271         nb_sectors -= NBD_MAX_SECTORS;
272     }
273     return nbd_co_readv_1(bs, sector_num, nb_sectors, qiov, offset);
274 }
275
276 int nbd_client_co_writev(BlockDriverState *bs, int64_t sector_num,
277                          int nb_sectors, QEMUIOVector *qiov)
278 {
279     int offset = 0;
280     int ret;
281     while (nb_sectors > NBD_MAX_SECTORS) {
282         ret = nbd_co_writev_1(bs, sector_num, NBD_MAX_SECTORS, qiov, offset);
283         if (ret < 0) {
284             return ret;
285         }
286         offset += NBD_MAX_SECTORS * 512;
287         sector_num += NBD_MAX_SECTORS;
288         nb_sectors -= NBD_MAX_SECTORS;
289     }
290     return nbd_co_writev_1(bs, sector_num, nb_sectors, qiov, offset);
291 }
292
293 int nbd_client_co_flush(BlockDriverState *bs)
294 {
295     NbdClientSession *client = nbd_get_client_session(bs);
296     struct nbd_request request = { .type = NBD_CMD_FLUSH };
297     struct nbd_reply reply;
298     ssize_t ret;
299
300     if (!(client->nbdflags & NBD_FLAG_SEND_FLUSH)) {
301         return 0;
302     }
303
304     if (client->nbdflags & NBD_FLAG_SEND_FUA) {
305         request.type |= NBD_CMD_FLAG_FUA;
306     }
307
308     request.from = 0;
309     request.len = 0;
310
311     nbd_coroutine_start(client, &request);
312     ret = nbd_co_send_request(bs, &request, NULL, 0);
313     if (ret < 0) {
314         reply.error = -ret;
315     } else {
316         nbd_co_receive_reply(client, &request, &reply, NULL, 0);
317     }
318     nbd_coroutine_end(client, &request);
319     return -reply.error;
320 }
321
322 int nbd_client_co_discard(BlockDriverState *bs, int64_t sector_num,
323                           int nb_sectors)
324 {
325     NbdClientSession *client = nbd_get_client_session(bs);
326     struct nbd_request request = { .type = NBD_CMD_TRIM };
327     struct nbd_reply reply;
328     ssize_t ret;
329
330     if (!(client->nbdflags & NBD_FLAG_SEND_TRIM)) {
331         return 0;
332     }
333     request.from = sector_num * 512;
334     request.len = nb_sectors * 512;
335
336     nbd_coroutine_start(client, &request);
337     ret = nbd_co_send_request(bs, &request, NULL, 0);
338     if (ret < 0) {
339         reply.error = -ret;
340     } else {
341         nbd_co_receive_reply(client, &request, &reply, NULL, 0);
342     }
343     nbd_coroutine_end(client, &request);
344     return -reply.error;
345
346 }
347
348 void nbd_client_detach_aio_context(BlockDriverState *bs)
349 {
350     aio_set_fd_handler(bdrv_get_aio_context(bs),
351                        nbd_get_client_session(bs)->sock, NULL, NULL, NULL);
352 }
353
354 void nbd_client_attach_aio_context(BlockDriverState *bs,
355                                    AioContext *new_context)
356 {
357     aio_set_fd_handler(new_context, nbd_get_client_session(bs)->sock,
358                        nbd_reply_ready, NULL, bs);
359 }
360
361 void nbd_client_close(BlockDriverState *bs)
362 {
363     NbdClientSession *client = nbd_get_client_session(bs);
364     struct nbd_request request = {
365         .type = NBD_CMD_DISC,
366         .from = 0,
367         .len = 0
368     };
369
370     if (client->sock == -1) {
371         return;
372     }
373
374     nbd_send_request(client->sock, &request);
375
376     nbd_teardown_connection(bs);
377 }
378
379 int nbd_client_init(BlockDriverState *bs, int sock, const char *export,
380                     Error **errp)
381 {
382     NbdClientSession *client = nbd_get_client_session(bs);
383     int ret;
384
385     /* NBD handshake */
386     logout("session init %s\n", export);
387     qemu_set_block(sock);
388     ret = nbd_receive_negotiate(sock, export,
389                                 &client->nbdflags, &client->size, errp);
390     if (ret < 0) {
391         logout("Failed to negotiate with the NBD server\n");
392         closesocket(sock);
393         return ret;
394     }
395
396     qemu_co_mutex_init(&client->send_mutex);
397     qemu_co_mutex_init(&client->free_sema);
398     client->sock = sock;
399
400     /* Now that we're connected, set the socket to be non-blocking and
401      * kick the reply mechanism.  */
402     qemu_set_nonblock(sock);
403     nbd_client_attach_aio_context(bs, bdrv_get_aio_context(bs));
404
405     logout("Established connection with NBD server\n");
406     return 0;
407 }