spapr: Workaround for broken radix guests
[qemu.git] / 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 <Laurent.Vivier@bull.net>
7 *
8 * Some parts:
9 * Copyright (C) 2007 Anthony Liguori <anthony@codemonkey.ws>
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 }