xref: /openbmc/qemu/nbd/server.c (revision b4682a63)
1 /*
2  *  Copyright (C) 2016-2018 Red Hat, Inc.
3  *  Copyright (C) 2005  Anthony Liguori <anthony@codemonkey.ws>
4  *
5  *  Network Block Device Server Side
6  *
7  *  This program is free software; you can redistribute it and/or modify
8  *  it under the terms of the GNU General Public License as published by
9  *  the Free Software Foundation; under version 2 of the License.
10  *
11  *  This program is distributed in the hope that it will be useful,
12  *  but WITHOUT ANY WARRANTY; without even the implied warranty of
13  *  MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
14  *  GNU General Public License for more details.
15  *
16  *  You should have received a copy of the GNU General Public License
17  *  along with this program; if not, see <http://www.gnu.org/licenses/>.
18  */
19 
20 #include "qemu/osdep.h"
21 #include "qapi/error.h"
22 #include "trace.h"
23 #include "nbd-internal.h"
24 
25 #define NBD_META_ID_BASE_ALLOCATION 0
26 #define NBD_META_ID_DIRTY_BITMAP 1
27 
28 /* NBD_MAX_BITMAP_EXTENTS: 1 mb of extents data. An empirical
29  * constant. If an increase is needed, note that the NBD protocol
30  * recommends no larger than 32 mb, so that the client won't consider
31  * the reply as a denial of service attack. */
32 #define NBD_MAX_BITMAP_EXTENTS (0x100000 / 8)
33 
34 static int system_errno_to_nbd_errno(int err)
35 {
36     switch (err) {
37     case 0:
38         return NBD_SUCCESS;
39     case EPERM:
40     case EROFS:
41         return NBD_EPERM;
42     case EIO:
43         return NBD_EIO;
44     case ENOMEM:
45         return NBD_ENOMEM;
46 #ifdef EDQUOT
47     case EDQUOT:
48 #endif
49     case EFBIG:
50     case ENOSPC:
51         return NBD_ENOSPC;
52     case EOVERFLOW:
53         return NBD_EOVERFLOW;
54     case ESHUTDOWN:
55         return NBD_ESHUTDOWN;
56     case EINVAL:
57     default:
58         return NBD_EINVAL;
59     }
60 }
61 
62 /* Definitions for opaque data types */
63 
64 typedef struct NBDRequestData NBDRequestData;
65 
66 struct NBDRequestData {
67     QSIMPLEQ_ENTRY(NBDRequestData) entry;
68     NBDClient *client;
69     uint8_t *data;
70     bool complete;
71 };
72 
73 struct NBDExport {
74     int refcount;
75     void (*close)(NBDExport *exp);
76 
77     BlockBackend *blk;
78     char *name;
79     char *description;
80     uint64_t dev_offset;
81     uint64_t size;
82     uint16_t nbdflags;
83     QTAILQ_HEAD(, NBDClient) clients;
84     QTAILQ_ENTRY(NBDExport) next;
85 
86     AioContext *ctx;
87 
88     BlockBackend *eject_notifier_blk;
89     Notifier eject_notifier;
90 
91     BdrvDirtyBitmap *export_bitmap;
92     char *export_bitmap_context;
93 };
94 
95 static QTAILQ_HEAD(, NBDExport) exports = QTAILQ_HEAD_INITIALIZER(exports);
96 
97 /* NBDExportMetaContexts represents a list of contexts to be exported,
98  * as selected by NBD_OPT_SET_META_CONTEXT. Also used for
99  * NBD_OPT_LIST_META_CONTEXT. */
100 typedef struct NBDExportMetaContexts {
101     NBDExport *exp;
102     bool valid; /* means that negotiation of the option finished without
103                    errors */
104     bool base_allocation; /* export base:allocation context (block status) */
105     bool bitmap; /* export qemu:dirty-bitmap:<export bitmap name> */
106 } NBDExportMetaContexts;
107 
108 struct NBDClient {
109     int refcount;
110     void (*close_fn)(NBDClient *client, bool negotiated);
111 
112     NBDExport *exp;
113     QCryptoTLSCreds *tlscreds;
114     char *tlsauthz;
115     QIOChannelSocket *sioc; /* The underlying data channel */
116     QIOChannel *ioc; /* The current I/O channel which may differ (eg TLS) */
117 
118     Coroutine *recv_coroutine;
119 
120     CoMutex send_lock;
121     Coroutine *send_coroutine;
122 
123     QTAILQ_ENTRY(NBDClient) next;
124     int nb_requests;
125     bool closing;
126 
127     bool structured_reply;
128     NBDExportMetaContexts export_meta;
129 
130     uint32_t opt; /* Current option being negotiated */
131     uint32_t optlen; /* remaining length of data in ioc for the option being
132                         negotiated now */
133 };
134 
135 static void nbd_client_receive_next_request(NBDClient *client);
136 
137 /* Basic flow for negotiation
138 
139    Server         Client
140    Negotiate
141 
142    or
143 
144    Server         Client
145    Negotiate #1
146                   Option
147    Negotiate #2
148 
149    ----
150 
151    followed by
152 
153    Server         Client
154                   Request
155    Response
156                   Request
157    Response
158                   ...
159    ...
160                   Request (type == 2)
161 
162 */
163 
164 static inline void set_be_option_rep(NBDOptionReply *rep, uint32_t option,
165                                      uint32_t type, uint32_t length)
166 {
167     stq_be_p(&rep->magic, NBD_REP_MAGIC);
168     stl_be_p(&rep->option, option);
169     stl_be_p(&rep->type, type);
170     stl_be_p(&rep->length, length);
171 }
172 
173 /* Send a reply header, including length, but no payload.
174  * Return -errno on error, 0 on success. */
175 static int nbd_negotiate_send_rep_len(NBDClient *client, uint32_t type,
176                                       uint32_t len, Error **errp)
177 {
178     NBDOptionReply rep;
179 
180     trace_nbd_negotiate_send_rep_len(client->opt, nbd_opt_lookup(client->opt),
181                                      type, nbd_rep_lookup(type), len);
182 
183     assert(len < NBD_MAX_BUFFER_SIZE);
184 
185     set_be_option_rep(&rep, client->opt, type, len);
186     return nbd_write(client->ioc, &rep, sizeof(rep), errp);
187 }
188 
189 /* Send a reply header with default 0 length.
190  * Return -errno on error, 0 on success. */
191 static int nbd_negotiate_send_rep(NBDClient *client, uint32_t type,
192                                   Error **errp)
193 {
194     return nbd_negotiate_send_rep_len(client, type, 0, errp);
195 }
196 
197 /* Send an error reply.
198  * Return -errno on error, 0 on success. */
199 static int GCC_FMT_ATTR(4, 0)
200 nbd_negotiate_send_rep_verr(NBDClient *client, uint32_t type,
201                             Error **errp, const char *fmt, va_list va)
202 {
203     char *msg;
204     int ret;
205     size_t len;
206 
207     msg = g_strdup_vprintf(fmt, va);
208     len = strlen(msg);
209     assert(len < 4096);
210     trace_nbd_negotiate_send_rep_err(msg);
211     ret = nbd_negotiate_send_rep_len(client, type, len, errp);
212     if (ret < 0) {
213         goto out;
214     }
215     if (nbd_write(client->ioc, msg, len, errp) < 0) {
216         error_prepend(errp, "write failed (error message): ");
217         ret = -EIO;
218     } else {
219         ret = 0;
220     }
221 
222 out:
223     g_free(msg);
224     return ret;
225 }
226 
227 /* Send an error reply.
228  * Return -errno on error, 0 on success. */
229 static int GCC_FMT_ATTR(4, 5)
230 nbd_negotiate_send_rep_err(NBDClient *client, uint32_t type,
231                            Error **errp, const char *fmt, ...)
232 {
233     va_list va;
234     int ret;
235 
236     va_start(va, fmt);
237     ret = nbd_negotiate_send_rep_verr(client, type, errp, fmt, va);
238     va_end(va);
239     return ret;
240 }
241 
242 /* Drop remainder of the current option, and send a reply with the
243  * given error type and message. Return -errno on read or write
244  * failure; or 0 if connection is still live. */
245 static int GCC_FMT_ATTR(4, 0)
246 nbd_opt_vdrop(NBDClient *client, uint32_t type, Error **errp,
247               const char *fmt, va_list va)
248 {
249     int ret = nbd_drop(client->ioc, client->optlen, errp);
250 
251     client->optlen = 0;
252     if (!ret) {
253         ret = nbd_negotiate_send_rep_verr(client, type, errp, fmt, va);
254     }
255     return ret;
256 }
257 
258 static int GCC_FMT_ATTR(4, 5)
259 nbd_opt_drop(NBDClient *client, uint32_t type, Error **errp,
260              const char *fmt, ...)
261 {
262     int ret;
263     va_list va;
264 
265     va_start(va, fmt);
266     ret = nbd_opt_vdrop(client, type, errp, fmt, va);
267     va_end(va);
268 
269     return ret;
270 }
271 
272 static int GCC_FMT_ATTR(3, 4)
273 nbd_opt_invalid(NBDClient *client, Error **errp, const char *fmt, ...)
274 {
275     int ret;
276     va_list va;
277 
278     va_start(va, fmt);
279     ret = nbd_opt_vdrop(client, NBD_REP_ERR_INVALID, errp, fmt, va);
280     va_end(va);
281 
282     return ret;
283 }
284 
285 /* Read size bytes from the unparsed payload of the current option.
286  * Return -errno on I/O error, 0 if option was completely handled by
287  * sending a reply about inconsistent lengths, or 1 on success. */
288 static int nbd_opt_read(NBDClient *client, void *buffer, size_t size,
289                         Error **errp)
290 {
291     if (size > client->optlen) {
292         return nbd_opt_invalid(client, errp,
293                                "Inconsistent lengths in option %s",
294                                nbd_opt_lookup(client->opt));
295     }
296     client->optlen -= size;
297     return qio_channel_read_all(client->ioc, buffer, size, errp) < 0 ? -EIO : 1;
298 }
299 
300 /* Drop size bytes from the unparsed payload of the current option.
301  * Return -errno on I/O error, 0 if option was completely handled by
302  * sending a reply about inconsistent lengths, or 1 on success. */
303 static int nbd_opt_skip(NBDClient *client, size_t size, Error **errp)
304 {
305     if (size > client->optlen) {
306         return nbd_opt_invalid(client, errp,
307                                "Inconsistent lengths in option %s",
308                                nbd_opt_lookup(client->opt));
309     }
310     client->optlen -= size;
311     return nbd_drop(client->ioc, size, errp) < 0 ? -EIO : 1;
312 }
313 
314 /* nbd_opt_read_name
315  *
316  * Read a string with the format:
317  *   uint32_t len     (<= NBD_MAX_NAME_SIZE)
318  *   len bytes string (not 0-terminated)
319  *
320  * @name should be enough to store NBD_MAX_NAME_SIZE+1.
321  * If @length is non-null, it will be set to the actual string length.
322  *
323  * Return -errno on I/O error, 0 if option was completely handled by
324  * sending a reply about inconsistent lengths, or 1 on success.
325  */
326 static int nbd_opt_read_name(NBDClient *client, char *name, uint32_t *length,
327                              Error **errp)
328 {
329     int ret;
330     uint32_t len;
331 
332     ret = nbd_opt_read(client, &len, sizeof(len), errp);
333     if (ret <= 0) {
334         return ret;
335     }
336     len = cpu_to_be32(len);
337 
338     if (len > NBD_MAX_NAME_SIZE) {
339         return nbd_opt_invalid(client, errp,
340                                "Invalid name length: %" PRIu32, len);
341     }
342 
343     ret = nbd_opt_read(client, name, len, errp);
344     if (ret <= 0) {
345         return ret;
346     }
347     name[len] = '\0';
348 
349     if (length) {
350         *length = len;
351     }
352 
353     return 1;
354 }
355 
356 /* Send a single NBD_REP_SERVER reply to NBD_OPT_LIST, including payload.
357  * Return -errno on error, 0 on success. */
358 static int nbd_negotiate_send_rep_list(NBDClient *client, NBDExport *exp,
359                                        Error **errp)
360 {
361     size_t name_len, desc_len;
362     uint32_t len;
363     const char *name = exp->name ? exp->name : "";
364     const char *desc = exp->description ? exp->description : "";
365     QIOChannel *ioc = client->ioc;
366     int ret;
367 
368     trace_nbd_negotiate_send_rep_list(name, desc);
369     name_len = strlen(name);
370     desc_len = strlen(desc);
371     len = name_len + desc_len + sizeof(len);
372     ret = nbd_negotiate_send_rep_len(client, NBD_REP_SERVER, len, errp);
373     if (ret < 0) {
374         return ret;
375     }
376 
377     len = cpu_to_be32(name_len);
378     if (nbd_write(ioc, &len, sizeof(len), errp) < 0) {
379         error_prepend(errp, "write failed (name length): ");
380         return -EINVAL;
381     }
382 
383     if (nbd_write(ioc, name, name_len, errp) < 0) {
384         error_prepend(errp, "write failed (name buffer): ");
385         return -EINVAL;
386     }
387 
388     if (nbd_write(ioc, desc, desc_len, errp) < 0) {
389         error_prepend(errp, "write failed (description buffer): ");
390         return -EINVAL;
391     }
392 
393     return 0;
394 }
395 
396 /* Process the NBD_OPT_LIST command, with a potential series of replies.
397  * Return -errno on error, 0 on success. */
398 static int nbd_negotiate_handle_list(NBDClient *client, Error **errp)
399 {
400     NBDExport *exp;
401     assert(client->opt == NBD_OPT_LIST);
402 
403     /* For each export, send a NBD_REP_SERVER reply. */
404     QTAILQ_FOREACH(exp, &exports, next) {
405         if (nbd_negotiate_send_rep_list(client, exp, errp)) {
406             return -EINVAL;
407         }
408     }
409     /* Finish with a NBD_REP_ACK. */
410     return nbd_negotiate_send_rep(client, NBD_REP_ACK, errp);
411 }
412 
413 static void nbd_check_meta_export(NBDClient *client)
414 {
415     client->export_meta.valid &= client->exp == client->export_meta.exp;
416 }
417 
418 /* Send a reply to NBD_OPT_EXPORT_NAME.
419  * Return -errno on error, 0 on success. */
420 static int nbd_negotiate_handle_export_name(NBDClient *client,
421                                             uint16_t myflags, bool no_zeroes,
422                                             Error **errp)
423 {
424     char name[NBD_MAX_NAME_SIZE + 1];
425     char buf[NBD_REPLY_EXPORT_NAME_SIZE] = "";
426     size_t len;
427     int ret;
428 
429     /* Client sends:
430         [20 ..  xx]   export name (length bytes)
431        Server replies:
432         [ 0 ..   7]   size
433         [ 8 ..   9]   export flags
434         [10 .. 133]   reserved     (0) [unless no_zeroes]
435      */
436     trace_nbd_negotiate_handle_export_name();
437     if (client->optlen >= sizeof(name)) {
438         error_setg(errp, "Bad length received");
439         return -EINVAL;
440     }
441     if (nbd_read(client->ioc, name, client->optlen, "export name", errp) < 0) {
442         return -EIO;
443     }
444     name[client->optlen] = '\0';
445     client->optlen = 0;
446 
447     trace_nbd_negotiate_handle_export_name_request(name);
448 
449     client->exp = nbd_export_find(name);
450     if (!client->exp) {
451         error_setg(errp, "export not found");
452         return -EINVAL;
453     }
454 
455     trace_nbd_negotiate_new_style_size_flags(client->exp->size,
456                                              client->exp->nbdflags | myflags);
457     stq_be_p(buf, client->exp->size);
458     stw_be_p(buf + 8, client->exp->nbdflags | myflags);
459     len = no_zeroes ? 10 : sizeof(buf);
460     ret = nbd_write(client->ioc, buf, len, errp);
461     if (ret < 0) {
462         error_prepend(errp, "write failed: ");
463         return ret;
464     }
465 
466     QTAILQ_INSERT_TAIL(&client->exp->clients, client, next);
467     nbd_export_get(client->exp);
468     nbd_check_meta_export(client);
469 
470     return 0;
471 }
472 
473 /* Send a single NBD_REP_INFO, with a buffer @buf of @length bytes.
474  * The buffer does NOT include the info type prefix.
475  * Return -errno on error, 0 if ready to send more. */
476 static int nbd_negotiate_send_info(NBDClient *client,
477                                    uint16_t info, uint32_t length, void *buf,
478                                    Error **errp)
479 {
480     int rc;
481 
482     trace_nbd_negotiate_send_info(info, nbd_info_lookup(info), length);
483     rc = nbd_negotiate_send_rep_len(client, NBD_REP_INFO,
484                                     sizeof(info) + length, errp);
485     if (rc < 0) {
486         return rc;
487     }
488     info = cpu_to_be16(info);
489     if (nbd_write(client->ioc, &info, sizeof(info), errp) < 0) {
490         return -EIO;
491     }
492     if (nbd_write(client->ioc, buf, length, errp) < 0) {
493         return -EIO;
494     }
495     return 0;
496 }
497 
498 /* nbd_reject_length: Handle any unexpected payload.
499  * @fatal requests that we quit talking to the client, even if we are able
500  * to successfully send an error reply.
501  * Return:
502  * -errno  transmission error occurred or @fatal was requested, errp is set
503  * 0       error message successfully sent to client, errp is not set
504  */
505 static int nbd_reject_length(NBDClient *client, bool fatal, Error **errp)
506 {
507     int ret;
508 
509     assert(client->optlen);
510     ret = nbd_opt_invalid(client, errp, "option '%s' has unexpected length",
511                           nbd_opt_lookup(client->opt));
512     if (fatal && !ret) {
513         error_setg(errp, "option '%s' has unexpected length",
514                    nbd_opt_lookup(client->opt));
515         return -EINVAL;
516     }
517     return ret;
518 }
519 
520 /* Handle NBD_OPT_INFO and NBD_OPT_GO.
521  * Return -errno on error, 0 if ready for next option, and 1 to move
522  * into transmission phase.  */
523 static int nbd_negotiate_handle_info(NBDClient *client, uint16_t myflags,
524                                      Error **errp)
525 {
526     int rc;
527     char name[NBD_MAX_NAME_SIZE + 1];
528     NBDExport *exp;
529     uint16_t requests;
530     uint16_t request;
531     uint32_t namelen;
532     bool sendname = false;
533     bool blocksize = false;
534     uint32_t sizes[3];
535     char buf[sizeof(uint64_t) + sizeof(uint16_t)];
536 
537     /* Client sends:
538         4 bytes: L, name length (can be 0)
539         L bytes: export name
540         2 bytes: N, number of requests (can be 0)
541         N * 2 bytes: N requests
542     */
543     rc = nbd_opt_read_name(client, name, &namelen, errp);
544     if (rc <= 0) {
545         return rc;
546     }
547     trace_nbd_negotiate_handle_export_name_request(name);
548 
549     rc = nbd_opt_read(client, &requests, sizeof(requests), errp);
550     if (rc <= 0) {
551         return rc;
552     }
553     requests = be16_to_cpu(requests);
554     trace_nbd_negotiate_handle_info_requests(requests);
555     while (requests--) {
556         rc = nbd_opt_read(client, &request, sizeof(request), errp);
557         if (rc <= 0) {
558             return rc;
559         }
560         request = be16_to_cpu(request);
561         trace_nbd_negotiate_handle_info_request(request,
562                                                 nbd_info_lookup(request));
563         /* We care about NBD_INFO_NAME and NBD_INFO_BLOCK_SIZE;
564          * everything else is either a request we don't know or
565          * something we send regardless of request */
566         switch (request) {
567         case NBD_INFO_NAME:
568             sendname = true;
569             break;
570         case NBD_INFO_BLOCK_SIZE:
571             blocksize = true;
572             break;
573         }
574     }
575     if (client->optlen) {
576         return nbd_reject_length(client, false, errp);
577     }
578 
579     exp = nbd_export_find(name);
580     if (!exp) {
581         return nbd_negotiate_send_rep_err(client, NBD_REP_ERR_UNKNOWN,
582                                           errp, "export '%s' not present",
583                                           name);
584     }
585 
586     /* Don't bother sending NBD_INFO_NAME unless client requested it */
587     if (sendname) {
588         rc = nbd_negotiate_send_info(client, NBD_INFO_NAME, namelen, name,
589                                      errp);
590         if (rc < 0) {
591             return rc;
592         }
593     }
594 
595     /* Send NBD_INFO_DESCRIPTION only if available, regardless of
596      * client request */
597     if (exp->description) {
598         size_t len = strlen(exp->description);
599 
600         rc = nbd_negotiate_send_info(client, NBD_INFO_DESCRIPTION,
601                                      len, exp->description, errp);
602         if (rc < 0) {
603             return rc;
604         }
605     }
606 
607     /* Send NBD_INFO_BLOCK_SIZE always, but tweak the minimum size
608      * according to whether the client requested it, and according to
609      * whether this is OPT_INFO or OPT_GO. */
610     /* minimum - 1 for back-compat, or actual if client will obey it. */
611     if (client->opt == NBD_OPT_INFO || blocksize) {
612         sizes[0] = blk_get_request_alignment(exp->blk);
613     } else {
614         sizes[0] = 1;
615     }
616     assert(sizes[0] <= NBD_MAX_BUFFER_SIZE);
617     /* preferred - Hard-code to 4096 for now.
618      * TODO: is blk_bs(blk)->bl.opt_transfer appropriate? */
619     sizes[1] = MAX(4096, sizes[0]);
620     /* maximum - At most 32M, but smaller as appropriate. */
621     sizes[2] = MIN(blk_get_max_transfer(exp->blk), NBD_MAX_BUFFER_SIZE);
622     trace_nbd_negotiate_handle_info_block_size(sizes[0], sizes[1], sizes[2]);
623     sizes[0] = cpu_to_be32(sizes[0]);
624     sizes[1] = cpu_to_be32(sizes[1]);
625     sizes[2] = cpu_to_be32(sizes[2]);
626     rc = nbd_negotiate_send_info(client, NBD_INFO_BLOCK_SIZE,
627                                  sizeof(sizes), sizes, errp);
628     if (rc < 0) {
629         return rc;
630     }
631 
632     /* Send NBD_INFO_EXPORT always */
633     trace_nbd_negotiate_new_style_size_flags(exp->size,
634                                              exp->nbdflags | myflags);
635     stq_be_p(buf, exp->size);
636     stw_be_p(buf + 8, exp->nbdflags | myflags);
637     rc = nbd_negotiate_send_info(client, NBD_INFO_EXPORT,
638                                  sizeof(buf), buf, errp);
639     if (rc < 0) {
640         return rc;
641     }
642 
643     /* If the client is just asking for NBD_OPT_INFO, but forgot to
644      * request block sizes, return an error.
645      * TODO: consult blk_bs(blk)->request_align, and only error if it
646      * is not 1? */
647     if (client->opt == NBD_OPT_INFO && !blocksize) {
648         return nbd_negotiate_send_rep_err(client,
649                                           NBD_REP_ERR_BLOCK_SIZE_REQD,
650                                           errp,
651                                           "request NBD_INFO_BLOCK_SIZE to "
652                                           "use this export");
653     }
654 
655     /* Final reply */
656     rc = nbd_negotiate_send_rep(client, NBD_REP_ACK, errp);
657     if (rc < 0) {
658         return rc;
659     }
660 
661     if (client->opt == NBD_OPT_GO) {
662         client->exp = exp;
663         QTAILQ_INSERT_TAIL(&client->exp->clients, client, next);
664         nbd_export_get(client->exp);
665         nbd_check_meta_export(client);
666         rc = 1;
667     }
668     return rc;
669 }
670 
671 
672 /* Handle NBD_OPT_STARTTLS. Return NULL to drop connection, or else the
673  * new channel for all further (now-encrypted) communication. */
674 static QIOChannel *nbd_negotiate_handle_starttls(NBDClient *client,
675                                                  Error **errp)
676 {
677     QIOChannel *ioc;
678     QIOChannelTLS *tioc;
679     struct NBDTLSHandshakeData data = { 0 };
680 
681     assert(client->opt == NBD_OPT_STARTTLS);
682 
683     trace_nbd_negotiate_handle_starttls();
684     ioc = client->ioc;
685 
686     if (nbd_negotiate_send_rep(client, NBD_REP_ACK, errp) < 0) {
687         return NULL;
688     }
689 
690     tioc = qio_channel_tls_new_server(ioc,
691                                       client->tlscreds,
692                                       client->tlsauthz,
693                                       errp);
694     if (!tioc) {
695         return NULL;
696     }
697 
698     qio_channel_set_name(QIO_CHANNEL(tioc), "nbd-server-tls");
699     trace_nbd_negotiate_handle_starttls_handshake();
700     data.loop = g_main_loop_new(g_main_context_default(), FALSE);
701     qio_channel_tls_handshake(tioc,
702                               nbd_tls_handshake,
703                               &data,
704                               NULL,
705                               NULL);
706 
707     if (!data.complete) {
708         g_main_loop_run(data.loop);
709     }
710     g_main_loop_unref(data.loop);
711     if (data.error) {
712         object_unref(OBJECT(tioc));
713         error_propagate(errp, data.error);
714         return NULL;
715     }
716 
717     return QIO_CHANNEL(tioc);
718 }
719 
720 /* nbd_negotiate_send_meta_context
721  *
722  * Send one chunk of reply to NBD_OPT_{LIST,SET}_META_CONTEXT
723  *
724  * For NBD_OPT_LIST_META_CONTEXT @context_id is ignored, 0 is used instead.
725  */
726 static int nbd_negotiate_send_meta_context(NBDClient *client,
727                                            const char *context,
728                                            uint32_t context_id,
729                                            Error **errp)
730 {
731     NBDOptionReplyMetaContext opt;
732     struct iovec iov[] = {
733         {.iov_base = &opt, .iov_len = sizeof(opt)},
734         {.iov_base = (void *)context, .iov_len = strlen(context)}
735     };
736 
737     if (client->opt == NBD_OPT_LIST_META_CONTEXT) {
738         context_id = 0;
739     }
740 
741     trace_nbd_negotiate_meta_query_reply(context, context_id);
742     set_be_option_rep(&opt.h, client->opt, NBD_REP_META_CONTEXT,
743                       sizeof(opt) - sizeof(opt.h) + iov[1].iov_len);
744     stl_be_p(&opt.context_id, context_id);
745 
746     return qio_channel_writev_all(client->ioc, iov, 2, errp) < 0 ? -EIO : 0;
747 }
748 
749 /* Read strlen(@pattern) bytes, and set @match to true if they match @pattern.
750  * @match is never set to false.
751  *
752  * Return -errno on I/O error, 0 if option was completely handled by
753  * sending a reply about inconsistent lengths, or 1 on success.
754  *
755  * Note: return code = 1 doesn't mean that we've read exactly @pattern.
756  * It only means that there are no errors.
757  */
758 static int nbd_meta_pattern(NBDClient *client, const char *pattern, bool *match,
759                             Error **errp)
760 {
761     int ret;
762     char *query;
763     size_t len = strlen(pattern);
764 
765     assert(len);
766 
767     query = g_malloc(len);
768     ret = nbd_opt_read(client, query, len, errp);
769     if (ret <= 0) {
770         g_free(query);
771         return ret;
772     }
773 
774     if (strncmp(query, pattern, len) == 0) {
775         trace_nbd_negotiate_meta_query_parse(pattern);
776         *match = true;
777     } else {
778         trace_nbd_negotiate_meta_query_skip("pattern not matched");
779     }
780     g_free(query);
781 
782     return 1;
783 }
784 
785 /*
786  * Read @len bytes, and set @match to true if they match @pattern, or if @len
787  * is 0 and the client is performing _LIST_. @match is never set to false.
788  *
789  * Return -errno on I/O error, 0 if option was completely handled by
790  * sending a reply about inconsistent lengths, or 1 on success.
791  *
792  * Note: return code = 1 doesn't mean that we've read exactly @pattern.
793  * It only means that there are no errors.
794  */
795 static int nbd_meta_empty_or_pattern(NBDClient *client, const char *pattern,
796                                      uint32_t len, bool *match, Error **errp)
797 {
798     if (len == 0) {
799         if (client->opt == NBD_OPT_LIST_META_CONTEXT) {
800             *match = true;
801         }
802         trace_nbd_negotiate_meta_query_parse("empty");
803         return 1;
804     }
805 
806     if (len != strlen(pattern)) {
807         trace_nbd_negotiate_meta_query_skip("different lengths");
808         return nbd_opt_skip(client, len, errp);
809     }
810 
811     return nbd_meta_pattern(client, pattern, match, errp);
812 }
813 
814 /* nbd_meta_base_query
815  *
816  * Handle queries to 'base' namespace. For now, only the base:allocation
817  * context is available.  'len' is the amount of text remaining to be read from
818  * the current name, after the 'base:' portion has been stripped.
819  *
820  * Return -errno on I/O error, 0 if option was completely handled by
821  * sending a reply about inconsistent lengths, or 1 on success.
822  */
823 static int nbd_meta_base_query(NBDClient *client, NBDExportMetaContexts *meta,
824                                uint32_t len, Error **errp)
825 {
826     return nbd_meta_empty_or_pattern(client, "allocation", len,
827                                      &meta->base_allocation, errp);
828 }
829 
830 /* nbd_meta_bitmap_query
831  *
832  * Handle query to 'qemu:' namespace.
833  * @len is the amount of text remaining to be read from the current name, after
834  * the 'qemu:' portion has been stripped.
835  *
836  * Return -errno on I/O error, 0 if option was completely handled by
837  * sending a reply about inconsistent lengths, or 1 on success. */
838 static int nbd_meta_qemu_query(NBDClient *client, NBDExportMetaContexts *meta,
839                                uint32_t len, Error **errp)
840 {
841     bool dirty_bitmap = false;
842     size_t dirty_bitmap_len = strlen("dirty-bitmap:");
843     int ret;
844 
845     if (!meta->exp->export_bitmap) {
846         trace_nbd_negotiate_meta_query_skip("no dirty-bitmap exported");
847         return nbd_opt_skip(client, len, errp);
848     }
849 
850     if (len == 0) {
851         if (client->opt == NBD_OPT_LIST_META_CONTEXT) {
852             meta->bitmap = true;
853         }
854         trace_nbd_negotiate_meta_query_parse("empty");
855         return 1;
856     }
857 
858     if (len < dirty_bitmap_len) {
859         trace_nbd_negotiate_meta_query_skip("not dirty-bitmap:");
860         return nbd_opt_skip(client, len, errp);
861     }
862 
863     len -= dirty_bitmap_len;
864     ret = nbd_meta_pattern(client, "dirty-bitmap:", &dirty_bitmap, errp);
865     if (ret <= 0) {
866         return ret;
867     }
868     if (!dirty_bitmap) {
869         trace_nbd_negotiate_meta_query_skip("not dirty-bitmap:");
870         return nbd_opt_skip(client, len, errp);
871     }
872 
873     trace_nbd_negotiate_meta_query_parse("dirty-bitmap:");
874 
875     return nbd_meta_empty_or_pattern(
876             client, meta->exp->export_bitmap_context +
877             strlen("qemu:dirty_bitmap:"), len, &meta->bitmap, errp);
878 }
879 
880 /* nbd_negotiate_meta_query
881  *
882  * Parse namespace name and call corresponding function to parse body of the
883  * query.
884  *
885  * The only supported namespace now is 'base'.
886  *
887  * The function aims not wasting time and memory to read long unknown namespace
888  * names.
889  *
890  * Return -errno on I/O error, 0 if option was completely handled by
891  * sending a reply about inconsistent lengths, or 1 on success. */
892 static int nbd_negotiate_meta_query(NBDClient *client,
893                                     NBDExportMetaContexts *meta, Error **errp)
894 {
895     /*
896      * Both 'qemu' and 'base' namespaces have length = 5 including a
897      * colon. If another length namespace is later introduced, this
898      * should certainly be refactored.
899      */
900     int ret;
901     size_t ns_len = 5;
902     char ns[5];
903     uint32_t len;
904 
905     ret = nbd_opt_read(client, &len, sizeof(len), errp);
906     if (ret <= 0) {
907         return ret;
908     }
909     len = cpu_to_be32(len);
910 
911     if (len < ns_len) {
912         trace_nbd_negotiate_meta_query_skip("length too short");
913         return nbd_opt_skip(client, len, errp);
914     }
915 
916     len -= ns_len;
917     ret = nbd_opt_read(client, ns, ns_len, errp);
918     if (ret <= 0) {
919         return ret;
920     }
921 
922     if (!strncmp(ns, "base:", ns_len)) {
923         trace_nbd_negotiate_meta_query_parse("base:");
924         return nbd_meta_base_query(client, meta, len, errp);
925     } else if (!strncmp(ns, "qemu:", ns_len)) {
926         trace_nbd_negotiate_meta_query_parse("qemu:");
927         return nbd_meta_qemu_query(client, meta, len, errp);
928     }
929 
930     trace_nbd_negotiate_meta_query_skip("unknown namespace");
931     return nbd_opt_skip(client, len, errp);
932 }
933 
934 /* nbd_negotiate_meta_queries
935  * Handle NBD_OPT_LIST_META_CONTEXT and NBD_OPT_SET_META_CONTEXT
936  *
937  * Return -errno on I/O error, or 0 if option was completely handled. */
938 static int nbd_negotiate_meta_queries(NBDClient *client,
939                                       NBDExportMetaContexts *meta, Error **errp)
940 {
941     int ret;
942     char export_name[NBD_MAX_NAME_SIZE + 1];
943     NBDExportMetaContexts local_meta;
944     uint32_t nb_queries;
945     int i;
946 
947     if (!client->structured_reply) {
948         return nbd_opt_invalid(client, errp,
949                                "request option '%s' when structured reply "
950                                "is not negotiated",
951                                nbd_opt_lookup(client->opt));
952     }
953 
954     if (client->opt == NBD_OPT_LIST_META_CONTEXT) {
955         /* Only change the caller's meta on SET. */
956         meta = &local_meta;
957     }
958 
959     memset(meta, 0, sizeof(*meta));
960 
961     ret = nbd_opt_read_name(client, export_name, NULL, errp);
962     if (ret <= 0) {
963         return ret;
964     }
965 
966     meta->exp = nbd_export_find(export_name);
967     if (meta->exp == NULL) {
968         return nbd_opt_drop(client, NBD_REP_ERR_UNKNOWN, errp,
969                             "export '%s' not present", export_name);
970     }
971 
972     ret = nbd_opt_read(client, &nb_queries, sizeof(nb_queries), errp);
973     if (ret <= 0) {
974         return ret;
975     }
976     nb_queries = cpu_to_be32(nb_queries);
977     trace_nbd_negotiate_meta_context(nbd_opt_lookup(client->opt),
978                                      export_name, nb_queries);
979 
980     if (client->opt == NBD_OPT_LIST_META_CONTEXT && !nb_queries) {
981         /* enable all known contexts */
982         meta->base_allocation = true;
983         meta->bitmap = !!meta->exp->export_bitmap;
984     } else {
985         for (i = 0; i < nb_queries; ++i) {
986             ret = nbd_negotiate_meta_query(client, meta, errp);
987             if (ret <= 0) {
988                 return ret;
989             }
990         }
991     }
992 
993     if (meta->base_allocation) {
994         ret = nbd_negotiate_send_meta_context(client, "base:allocation",
995                                               NBD_META_ID_BASE_ALLOCATION,
996                                               errp);
997         if (ret < 0) {
998             return ret;
999         }
1000     }
1001 
1002     if (meta->bitmap) {
1003         ret = nbd_negotiate_send_meta_context(client,
1004                                               meta->exp->export_bitmap_context,
1005                                               NBD_META_ID_DIRTY_BITMAP,
1006                                               errp);
1007         if (ret < 0) {
1008             return ret;
1009         }
1010     }
1011 
1012     ret = nbd_negotiate_send_rep(client, NBD_REP_ACK, errp);
1013     if (ret == 0) {
1014         meta->valid = true;
1015     }
1016 
1017     return ret;
1018 }
1019 
1020 /* nbd_negotiate_options
1021  * Process all NBD_OPT_* client option commands, during fixed newstyle
1022  * negotiation.
1023  * Return:
1024  * -errno  on error, errp is set
1025  * 0       on successful negotiation, errp is not set
1026  * 1       if client sent NBD_OPT_ABORT, i.e. on valid disconnect,
1027  *         errp is not set
1028  */
1029 static int nbd_negotiate_options(NBDClient *client, uint16_t myflags,
1030                                  Error **errp)
1031 {
1032     uint32_t flags;
1033     bool fixedNewstyle = false;
1034     bool no_zeroes = false;
1035 
1036     /* Client sends:
1037         [ 0 ..   3]   client flags
1038 
1039        Then we loop until NBD_OPT_EXPORT_NAME or NBD_OPT_GO:
1040         [ 0 ..   7]   NBD_OPTS_MAGIC
1041         [ 8 ..  11]   NBD option
1042         [12 ..  15]   Data length
1043         ...           Rest of request
1044 
1045         [ 0 ..   7]   NBD_OPTS_MAGIC
1046         [ 8 ..  11]   Second NBD option
1047         [12 ..  15]   Data length
1048         ...           Rest of request
1049     */
1050 
1051     if (nbd_read32(client->ioc, &flags, "flags", errp) < 0) {
1052         return -EIO;
1053     }
1054     trace_nbd_negotiate_options_flags(flags);
1055     if (flags & NBD_FLAG_C_FIXED_NEWSTYLE) {
1056         fixedNewstyle = true;
1057         flags &= ~NBD_FLAG_C_FIXED_NEWSTYLE;
1058     }
1059     if (flags & NBD_FLAG_C_NO_ZEROES) {
1060         no_zeroes = true;
1061         flags &= ~NBD_FLAG_C_NO_ZEROES;
1062     }
1063     if (flags != 0) {
1064         error_setg(errp, "Unknown client flags 0x%" PRIx32 " received", flags);
1065         return -EINVAL;
1066     }
1067 
1068     while (1) {
1069         int ret;
1070         uint32_t option, length;
1071         uint64_t magic;
1072 
1073         if (nbd_read64(client->ioc, &magic, "opts magic", errp) < 0) {
1074             return -EINVAL;
1075         }
1076         trace_nbd_negotiate_options_check_magic(magic);
1077         if (magic != NBD_OPTS_MAGIC) {
1078             error_setg(errp, "Bad magic received");
1079             return -EINVAL;
1080         }
1081 
1082         if (nbd_read32(client->ioc, &option, "option", errp) < 0) {
1083             return -EINVAL;
1084         }
1085         client->opt = option;
1086 
1087         if (nbd_read32(client->ioc, &length, "option length", errp) < 0) {
1088             return -EINVAL;
1089         }
1090         assert(!client->optlen);
1091         client->optlen = length;
1092 
1093         if (length > NBD_MAX_BUFFER_SIZE) {
1094             error_setg(errp, "len (%" PRIu32" ) is larger than max len (%u)",
1095                        length, NBD_MAX_BUFFER_SIZE);
1096             return -EINVAL;
1097         }
1098 
1099         trace_nbd_negotiate_options_check_option(option,
1100                                                  nbd_opt_lookup(option));
1101         if (client->tlscreds &&
1102             client->ioc == (QIOChannel *)client->sioc) {
1103             QIOChannel *tioc;
1104             if (!fixedNewstyle) {
1105                 error_setg(errp, "Unsupported option 0x%" PRIx32, option);
1106                 return -EINVAL;
1107             }
1108             switch (option) {
1109             case NBD_OPT_STARTTLS:
1110                 if (length) {
1111                     /* Unconditionally drop the connection if the client
1112                      * can't start a TLS negotiation correctly */
1113                     return nbd_reject_length(client, true, errp);
1114                 }
1115                 tioc = nbd_negotiate_handle_starttls(client, errp);
1116                 if (!tioc) {
1117                     return -EIO;
1118                 }
1119                 ret = 0;
1120                 object_unref(OBJECT(client->ioc));
1121                 client->ioc = QIO_CHANNEL(tioc);
1122                 break;
1123 
1124             case NBD_OPT_EXPORT_NAME:
1125                 /* No way to return an error to client, so drop connection */
1126                 error_setg(errp, "Option 0x%x not permitted before TLS",
1127                            option);
1128                 return -EINVAL;
1129 
1130             default:
1131                 /* Let the client keep trying, unless they asked to
1132                  * quit. Always try to give an error back to the
1133                  * client; but when replying to OPT_ABORT, be aware
1134                  * that the client may hang up before receiving the
1135                  * error, in which case we are fine ignoring the
1136                  * resulting EPIPE. */
1137                 ret = nbd_opt_drop(client, NBD_REP_ERR_TLS_REQD,
1138                                    option == NBD_OPT_ABORT ? NULL : errp,
1139                                    "Option 0x%" PRIx32
1140                                    " not permitted before TLS", option);
1141                 if (option == NBD_OPT_ABORT) {
1142                     return 1;
1143                 }
1144                 break;
1145             }
1146         } else if (fixedNewstyle) {
1147             switch (option) {
1148             case NBD_OPT_LIST:
1149                 if (length) {
1150                     ret = nbd_reject_length(client, false, errp);
1151                 } else {
1152                     ret = nbd_negotiate_handle_list(client, errp);
1153                 }
1154                 break;
1155 
1156             case NBD_OPT_ABORT:
1157                 /* NBD spec says we must try to reply before
1158                  * disconnecting, but that we must also tolerate
1159                  * guests that don't wait for our reply. */
1160                 nbd_negotiate_send_rep(client, NBD_REP_ACK, NULL);
1161                 return 1;
1162 
1163             case NBD_OPT_EXPORT_NAME:
1164                 return nbd_negotiate_handle_export_name(client,
1165                                                         myflags, no_zeroes,
1166                                                         errp);
1167 
1168             case NBD_OPT_INFO:
1169             case NBD_OPT_GO:
1170                 ret = nbd_negotiate_handle_info(client, myflags, errp);
1171                 if (ret == 1) {
1172                     assert(option == NBD_OPT_GO);
1173                     return 0;
1174                 }
1175                 break;
1176 
1177             case NBD_OPT_STARTTLS:
1178                 if (length) {
1179                     ret = nbd_reject_length(client, false, errp);
1180                 } else if (client->tlscreds) {
1181                     ret = nbd_negotiate_send_rep_err(client,
1182                                                      NBD_REP_ERR_INVALID, errp,
1183                                                      "TLS already enabled");
1184                 } else {
1185                     ret = nbd_negotiate_send_rep_err(client,
1186                                                      NBD_REP_ERR_POLICY, errp,
1187                                                      "TLS not configured");
1188                 }
1189                 break;
1190 
1191             case NBD_OPT_STRUCTURED_REPLY:
1192                 if (length) {
1193                     ret = nbd_reject_length(client, false, errp);
1194                 } else if (client->structured_reply) {
1195                     ret = nbd_negotiate_send_rep_err(
1196                         client, NBD_REP_ERR_INVALID, errp,
1197                         "structured reply already negotiated");
1198                 } else {
1199                     ret = nbd_negotiate_send_rep(client, NBD_REP_ACK, errp);
1200                     client->structured_reply = true;
1201                     myflags |= NBD_FLAG_SEND_DF;
1202                 }
1203                 break;
1204 
1205             case NBD_OPT_LIST_META_CONTEXT:
1206             case NBD_OPT_SET_META_CONTEXT:
1207                 ret = nbd_negotiate_meta_queries(client, &client->export_meta,
1208                                                  errp);
1209                 break;
1210 
1211             default:
1212                 ret = nbd_opt_drop(client, NBD_REP_ERR_UNSUP, errp,
1213                                    "Unsupported option %" PRIu32 " (%s)",
1214                                    option, nbd_opt_lookup(option));
1215                 break;
1216             }
1217         } else {
1218             /*
1219              * If broken new-style we should drop the connection
1220              * for anything except NBD_OPT_EXPORT_NAME
1221              */
1222             switch (option) {
1223             case NBD_OPT_EXPORT_NAME:
1224                 return nbd_negotiate_handle_export_name(client,
1225                                                         myflags, no_zeroes,
1226                                                         errp);
1227 
1228             default:
1229                 error_setg(errp, "Unsupported option %" PRIu32 " (%s)",
1230                            option, nbd_opt_lookup(option));
1231                 return -EINVAL;
1232             }
1233         }
1234         if (ret < 0) {
1235             return ret;
1236         }
1237     }
1238 }
1239 
1240 /* nbd_negotiate
1241  * Return:
1242  * -errno  on error, errp is set
1243  * 0       on successful negotiation, errp is not set
1244  * 1       if client sent NBD_OPT_ABORT, i.e. on valid disconnect,
1245  *         errp is not set
1246  */
1247 static coroutine_fn int nbd_negotiate(NBDClient *client, Error **errp)
1248 {
1249     char buf[NBD_OLDSTYLE_NEGOTIATE_SIZE] = "";
1250     int ret;
1251     const uint16_t myflags = (NBD_FLAG_HAS_FLAGS | NBD_FLAG_SEND_TRIM |
1252                               NBD_FLAG_SEND_FLUSH | NBD_FLAG_SEND_FUA |
1253                               NBD_FLAG_SEND_WRITE_ZEROES | NBD_FLAG_SEND_CACHE);
1254 
1255     /* Old style negotiation header, no room for options
1256         [ 0 ..   7]   passwd       ("NBDMAGIC")
1257         [ 8 ..  15]   magic        (NBD_CLIENT_MAGIC)
1258         [16 ..  23]   size
1259         [24 ..  27]   export flags (zero-extended)
1260         [28 .. 151]   reserved     (0)
1261 
1262        New style negotiation header, client can send options
1263         [ 0 ..   7]   passwd       ("NBDMAGIC")
1264         [ 8 ..  15]   magic        (NBD_OPTS_MAGIC)
1265         [16 ..  17]   server flags (0)
1266         ....options sent, ending in NBD_OPT_EXPORT_NAME or NBD_OPT_GO....
1267      */
1268 
1269     qio_channel_set_blocking(client->ioc, false, NULL);
1270 
1271     trace_nbd_negotiate_begin();
1272     memcpy(buf, "NBDMAGIC", 8);
1273 
1274     stq_be_p(buf + 8, NBD_OPTS_MAGIC);
1275     stw_be_p(buf + 16, NBD_FLAG_FIXED_NEWSTYLE | NBD_FLAG_NO_ZEROES);
1276 
1277     if (nbd_write(client->ioc, buf, 18, errp) < 0) {
1278         error_prepend(errp, "write failed: ");
1279         return -EINVAL;
1280     }
1281     ret = nbd_negotiate_options(client, myflags, errp);
1282     if (ret != 0) {
1283         if (ret < 0) {
1284             error_prepend(errp, "option negotiation failed: ");
1285         }
1286         return ret;
1287     }
1288 
1289     assert(!client->optlen);
1290     trace_nbd_negotiate_success();
1291 
1292     return 0;
1293 }
1294 
1295 static int nbd_receive_request(QIOChannel *ioc, NBDRequest *request,
1296                                Error **errp)
1297 {
1298     uint8_t buf[NBD_REQUEST_SIZE];
1299     uint32_t magic;
1300     int ret;
1301 
1302     ret = nbd_read(ioc, buf, sizeof(buf), "request", errp);
1303     if (ret < 0) {
1304         return ret;
1305     }
1306 
1307     /* Request
1308        [ 0 ..  3]   magic   (NBD_REQUEST_MAGIC)
1309        [ 4 ..  5]   flags   (NBD_CMD_FLAG_FUA, ...)
1310        [ 6 ..  7]   type    (NBD_CMD_READ, ...)
1311        [ 8 .. 15]   handle
1312        [16 .. 23]   from
1313        [24 .. 27]   len
1314      */
1315 
1316     magic = ldl_be_p(buf);
1317     request->flags  = lduw_be_p(buf + 4);
1318     request->type   = lduw_be_p(buf + 6);
1319     request->handle = ldq_be_p(buf + 8);
1320     request->from   = ldq_be_p(buf + 16);
1321     request->len    = ldl_be_p(buf + 24);
1322 
1323     trace_nbd_receive_request(magic, request->flags, request->type,
1324                               request->from, request->len);
1325 
1326     if (magic != NBD_REQUEST_MAGIC) {
1327         error_setg(errp, "invalid magic (got 0x%" PRIx32 ")", magic);
1328         return -EINVAL;
1329     }
1330     return 0;
1331 }
1332 
1333 #define MAX_NBD_REQUESTS 16
1334 
1335 void nbd_client_get(NBDClient *client)
1336 {
1337     client->refcount++;
1338 }
1339 
1340 void nbd_client_put(NBDClient *client)
1341 {
1342     if (--client->refcount == 0) {
1343         /* The last reference should be dropped by client->close,
1344          * which is called by client_close.
1345          */
1346         assert(client->closing);
1347 
1348         qio_channel_detach_aio_context(client->ioc);
1349         object_unref(OBJECT(client->sioc));
1350         object_unref(OBJECT(client->ioc));
1351         if (client->tlscreds) {
1352             object_unref(OBJECT(client->tlscreds));
1353         }
1354         g_free(client->tlsauthz);
1355         if (client->exp) {
1356             QTAILQ_REMOVE(&client->exp->clients, client, next);
1357             nbd_export_put(client->exp);
1358         }
1359         g_free(client);
1360     }
1361 }
1362 
1363 static void client_close(NBDClient *client, bool negotiated)
1364 {
1365     if (client->closing) {
1366         return;
1367     }
1368 
1369     client->closing = true;
1370 
1371     /* Force requests to finish.  They will drop their own references,
1372      * then we'll close the socket and free the NBDClient.
1373      */
1374     qio_channel_shutdown(client->ioc, QIO_CHANNEL_SHUTDOWN_BOTH,
1375                          NULL);
1376 
1377     /* Also tell the client, so that they release their reference.  */
1378     if (client->close_fn) {
1379         client->close_fn(client, negotiated);
1380     }
1381 }
1382 
1383 static NBDRequestData *nbd_request_get(NBDClient *client)
1384 {
1385     NBDRequestData *req;
1386 
1387     assert(client->nb_requests <= MAX_NBD_REQUESTS - 1);
1388     client->nb_requests++;
1389 
1390     req = g_new0(NBDRequestData, 1);
1391     nbd_client_get(client);
1392     req->client = client;
1393     return req;
1394 }
1395 
1396 static void nbd_request_put(NBDRequestData *req)
1397 {
1398     NBDClient *client = req->client;
1399 
1400     if (req->data) {
1401         qemu_vfree(req->data);
1402     }
1403     g_free(req);
1404 
1405     client->nb_requests--;
1406     nbd_client_receive_next_request(client);
1407 
1408     nbd_client_put(client);
1409 }
1410 
1411 static void blk_aio_attached(AioContext *ctx, void *opaque)
1412 {
1413     NBDExport *exp = opaque;
1414     NBDClient *client;
1415 
1416     trace_nbd_blk_aio_attached(exp->name, ctx);
1417 
1418     exp->ctx = ctx;
1419 
1420     QTAILQ_FOREACH(client, &exp->clients, next) {
1421         qio_channel_attach_aio_context(client->ioc, ctx);
1422         if (client->recv_coroutine) {
1423             aio_co_schedule(ctx, client->recv_coroutine);
1424         }
1425         if (client->send_coroutine) {
1426             aio_co_schedule(ctx, client->send_coroutine);
1427         }
1428     }
1429 }
1430 
1431 static void blk_aio_detach(void *opaque)
1432 {
1433     NBDExport *exp = opaque;
1434     NBDClient *client;
1435 
1436     trace_nbd_blk_aio_detach(exp->name, exp->ctx);
1437 
1438     QTAILQ_FOREACH(client, &exp->clients, next) {
1439         qio_channel_detach_aio_context(client->ioc);
1440     }
1441 
1442     exp->ctx = NULL;
1443 }
1444 
1445 static void nbd_eject_notifier(Notifier *n, void *data)
1446 {
1447     NBDExport *exp = container_of(n, NBDExport, eject_notifier);
1448     nbd_export_close(exp);
1449 }
1450 
1451 NBDExport *nbd_export_new(BlockDriverState *bs, uint64_t dev_offset,
1452                           uint64_t size, const char *name, const char *desc,
1453                           const char *bitmap, uint16_t nbdflags,
1454                           void (*close)(NBDExport *), bool writethrough,
1455                           BlockBackend *on_eject_blk, Error **errp)
1456 {
1457     AioContext *ctx;
1458     BlockBackend *blk;
1459     NBDExport *exp = g_new0(NBDExport, 1);
1460     uint64_t perm;
1461     int ret;
1462 
1463     /*
1464      * NBD exports are used for non-shared storage migration.  Make sure
1465      * that BDRV_O_INACTIVE is cleared and the image is ready for write
1466      * access since the export could be available before migration handover.
1467      */
1468     assert(name);
1469     ctx = bdrv_get_aio_context(bs);
1470     aio_context_acquire(ctx);
1471     bdrv_invalidate_cache(bs, NULL);
1472     aio_context_release(ctx);
1473 
1474     /* Don't allow resize while the NBD server is running, otherwise we don't
1475      * care what happens with the node. */
1476     perm = BLK_PERM_CONSISTENT_READ;
1477     if ((nbdflags & NBD_FLAG_READ_ONLY) == 0) {
1478         perm |= BLK_PERM_WRITE;
1479     }
1480     blk = blk_new(perm, BLK_PERM_CONSISTENT_READ | BLK_PERM_WRITE_UNCHANGED |
1481                         BLK_PERM_WRITE | BLK_PERM_GRAPH_MOD);
1482     ret = blk_insert_bs(blk, bs, errp);
1483     if (ret < 0) {
1484         goto fail;
1485     }
1486     blk_set_enable_write_cache(blk, !writethrough);
1487 
1488     exp->refcount = 1;
1489     QTAILQ_INIT(&exp->clients);
1490     exp->blk = blk;
1491     assert(dev_offset <= INT64_MAX);
1492     exp->dev_offset = dev_offset;
1493     exp->name = g_strdup(name);
1494     exp->description = g_strdup(desc);
1495     exp->nbdflags = nbdflags;
1496     assert(size <= INT64_MAX - dev_offset);
1497     exp->size = QEMU_ALIGN_DOWN(size, BDRV_SECTOR_SIZE);
1498 
1499     if (bitmap) {
1500         BdrvDirtyBitmap *bm = NULL;
1501 
1502         while (true) {
1503             bm = bdrv_find_dirty_bitmap(bs, bitmap);
1504             if (bm != NULL || bs->backing == NULL) {
1505                 break;
1506             }
1507 
1508             bs = bs->backing->bs;
1509         }
1510 
1511         if (bm == NULL) {
1512             error_setg(errp, "Bitmap '%s' is not found", bitmap);
1513             goto fail;
1514         }
1515 
1516         if (bdrv_dirty_bitmap_check(bm, BDRV_BITMAP_ALLOW_RO, errp)) {
1517             goto fail;
1518         }
1519 
1520         if ((nbdflags & NBD_FLAG_READ_ONLY) && bdrv_is_writable(bs) &&
1521             bdrv_dirty_bitmap_enabled(bm)) {
1522             error_setg(errp,
1523                        "Enabled bitmap '%s' incompatible with readonly export",
1524                        bitmap);
1525             goto fail;
1526         }
1527 
1528         bdrv_dirty_bitmap_set_busy(bm, true);
1529         exp->export_bitmap = bm;
1530         exp->export_bitmap_context = g_strdup_printf("qemu:dirty-bitmap:%s",
1531                                                      bitmap);
1532     }
1533 
1534     exp->close = close;
1535     exp->ctx = blk_get_aio_context(blk);
1536     blk_add_aio_context_notifier(blk, blk_aio_attached, blk_aio_detach, exp);
1537 
1538     if (on_eject_blk) {
1539         blk_ref(on_eject_blk);
1540         exp->eject_notifier_blk = on_eject_blk;
1541         exp->eject_notifier.notify = nbd_eject_notifier;
1542         blk_add_remove_bs_notifier(on_eject_blk, &exp->eject_notifier);
1543     }
1544     QTAILQ_INSERT_TAIL(&exports, exp, next);
1545     nbd_export_get(exp);
1546     return exp;
1547 
1548 fail:
1549     blk_unref(blk);
1550     g_free(exp->name);
1551     g_free(exp->description);
1552     g_free(exp);
1553     return NULL;
1554 }
1555 
1556 NBDExport *nbd_export_find(const char *name)
1557 {
1558     NBDExport *exp;
1559     QTAILQ_FOREACH(exp, &exports, next) {
1560         if (strcmp(name, exp->name) == 0) {
1561             return exp;
1562         }
1563     }
1564 
1565     return NULL;
1566 }
1567 
1568 void nbd_export_close(NBDExport *exp)
1569 {
1570     NBDClient *client, *next;
1571 
1572     nbd_export_get(exp);
1573     /*
1574      * TODO: Should we expand QMP NbdServerRemoveNode enum to allow a
1575      * close mode that stops advertising the export to new clients but
1576      * still permits existing clients to run to completion? Because of
1577      * that possibility, nbd_export_close() can be called more than
1578      * once on an export.
1579      */
1580     QTAILQ_FOREACH_SAFE(client, &exp->clients, next, next) {
1581         client_close(client, true);
1582     }
1583     if (exp->name) {
1584         nbd_export_put(exp);
1585         g_free(exp->name);
1586         exp->name = NULL;
1587         QTAILQ_REMOVE(&exports, exp, next);
1588     }
1589     g_free(exp->description);
1590     exp->description = NULL;
1591     nbd_export_put(exp);
1592 }
1593 
1594 void nbd_export_remove(NBDExport *exp, NbdServerRemoveMode mode, Error **errp)
1595 {
1596     if (mode == NBD_SERVER_REMOVE_MODE_HARD || QTAILQ_EMPTY(&exp->clients)) {
1597         nbd_export_close(exp);
1598         return;
1599     }
1600 
1601     assert(mode == NBD_SERVER_REMOVE_MODE_SAFE);
1602 
1603     error_setg(errp, "export '%s' still in use", exp->name);
1604     error_append_hint(errp, "Use mode='hard' to force client disconnect\n");
1605 }
1606 
1607 void nbd_export_get(NBDExport *exp)
1608 {
1609     assert(exp->refcount > 0);
1610     exp->refcount++;
1611 }
1612 
1613 void nbd_export_put(NBDExport *exp)
1614 {
1615     assert(exp->refcount > 0);
1616     if (exp->refcount == 1) {
1617         nbd_export_close(exp);
1618     }
1619 
1620     /* nbd_export_close() may theoretically reduce refcount to 0. It may happen
1621      * if someone calls nbd_export_put() on named export not through
1622      * nbd_export_set_name() when refcount is 1. So, let's assert that
1623      * it is > 0.
1624      */
1625     assert(exp->refcount > 0);
1626     if (--exp->refcount == 0) {
1627         assert(exp->name == NULL);
1628         assert(exp->description == NULL);
1629 
1630         if (exp->close) {
1631             exp->close(exp);
1632         }
1633 
1634         if (exp->blk) {
1635             if (exp->eject_notifier_blk) {
1636                 notifier_remove(&exp->eject_notifier);
1637                 blk_unref(exp->eject_notifier_blk);
1638             }
1639             blk_remove_aio_context_notifier(exp->blk, blk_aio_attached,
1640                                             blk_aio_detach, exp);
1641             blk_unref(exp->blk);
1642             exp->blk = NULL;
1643         }
1644 
1645         if (exp->export_bitmap) {
1646             bdrv_dirty_bitmap_set_busy(exp->export_bitmap, false);
1647             g_free(exp->export_bitmap_context);
1648         }
1649 
1650         g_free(exp);
1651     }
1652 }
1653 
1654 BlockBackend *nbd_export_get_blockdev(NBDExport *exp)
1655 {
1656     return exp->blk;
1657 }
1658 
1659 void nbd_export_close_all(void)
1660 {
1661     NBDExport *exp, *next;
1662 
1663     QTAILQ_FOREACH_SAFE(exp, &exports, next, next) {
1664         nbd_export_close(exp);
1665     }
1666 }
1667 
1668 static int coroutine_fn nbd_co_send_iov(NBDClient *client, struct iovec *iov,
1669                                         unsigned niov, Error **errp)
1670 {
1671     int ret;
1672 
1673     g_assert(qemu_in_coroutine());
1674     qemu_co_mutex_lock(&client->send_lock);
1675     client->send_coroutine = qemu_coroutine_self();
1676 
1677     ret = qio_channel_writev_all(client->ioc, iov, niov, errp) < 0 ? -EIO : 0;
1678 
1679     client->send_coroutine = NULL;
1680     qemu_co_mutex_unlock(&client->send_lock);
1681 
1682     return ret;
1683 }
1684 
1685 static inline void set_be_simple_reply(NBDSimpleReply *reply, uint64_t error,
1686                                        uint64_t handle)
1687 {
1688     stl_be_p(&reply->magic, NBD_SIMPLE_REPLY_MAGIC);
1689     stl_be_p(&reply->error, error);
1690     stq_be_p(&reply->handle, handle);
1691 }
1692 
1693 static int nbd_co_send_simple_reply(NBDClient *client,
1694                                     uint64_t handle,
1695                                     uint32_t error,
1696                                     void *data,
1697                                     size_t len,
1698                                     Error **errp)
1699 {
1700     NBDSimpleReply reply;
1701     int nbd_err = system_errno_to_nbd_errno(error);
1702     struct iovec iov[] = {
1703         {.iov_base = &reply, .iov_len = sizeof(reply)},
1704         {.iov_base = data, .iov_len = len}
1705     };
1706 
1707     trace_nbd_co_send_simple_reply(handle, nbd_err, nbd_err_lookup(nbd_err),
1708                                    len);
1709     set_be_simple_reply(&reply, nbd_err, handle);
1710 
1711     return nbd_co_send_iov(client, iov, len ? 2 : 1, errp);
1712 }
1713 
1714 static inline void set_be_chunk(NBDStructuredReplyChunk *chunk, uint16_t flags,
1715                                 uint16_t type, uint64_t handle, uint32_t length)
1716 {
1717     stl_be_p(&chunk->magic, NBD_STRUCTURED_REPLY_MAGIC);
1718     stw_be_p(&chunk->flags, flags);
1719     stw_be_p(&chunk->type, type);
1720     stq_be_p(&chunk->handle, handle);
1721     stl_be_p(&chunk->length, length);
1722 }
1723 
1724 static int coroutine_fn nbd_co_send_structured_done(NBDClient *client,
1725                                                     uint64_t handle,
1726                                                     Error **errp)
1727 {
1728     NBDStructuredReplyChunk chunk;
1729     struct iovec iov[] = {
1730         {.iov_base = &chunk, .iov_len = sizeof(chunk)},
1731     };
1732 
1733     trace_nbd_co_send_structured_done(handle);
1734     set_be_chunk(&chunk, NBD_REPLY_FLAG_DONE, NBD_REPLY_TYPE_NONE, handle, 0);
1735 
1736     return nbd_co_send_iov(client, iov, 1, errp);
1737 }
1738 
1739 static int coroutine_fn nbd_co_send_structured_read(NBDClient *client,
1740                                                     uint64_t handle,
1741                                                     uint64_t offset,
1742                                                     void *data,
1743                                                     size_t size,
1744                                                     bool final,
1745                                                     Error **errp)
1746 {
1747     NBDStructuredReadData chunk;
1748     struct iovec iov[] = {
1749         {.iov_base = &chunk, .iov_len = sizeof(chunk)},
1750         {.iov_base = data, .iov_len = size}
1751     };
1752 
1753     assert(size);
1754     trace_nbd_co_send_structured_read(handle, offset, data, size);
1755     set_be_chunk(&chunk.h, final ? NBD_REPLY_FLAG_DONE : 0,
1756                  NBD_REPLY_TYPE_OFFSET_DATA, handle,
1757                  sizeof(chunk) - sizeof(chunk.h) + size);
1758     stq_be_p(&chunk.offset, offset);
1759 
1760     return nbd_co_send_iov(client, iov, 2, errp);
1761 }
1762 
1763 static int coroutine_fn nbd_co_send_structured_error(NBDClient *client,
1764                                                      uint64_t handle,
1765                                                      uint32_t error,
1766                                                      const char *msg,
1767                                                      Error **errp)
1768 {
1769     NBDStructuredError chunk;
1770     int nbd_err = system_errno_to_nbd_errno(error);
1771     struct iovec iov[] = {
1772         {.iov_base = &chunk, .iov_len = sizeof(chunk)},
1773         {.iov_base = (char *)msg, .iov_len = msg ? strlen(msg) : 0},
1774     };
1775 
1776     assert(nbd_err);
1777     trace_nbd_co_send_structured_error(handle, nbd_err,
1778                                        nbd_err_lookup(nbd_err), msg ? msg : "");
1779     set_be_chunk(&chunk.h, NBD_REPLY_FLAG_DONE, NBD_REPLY_TYPE_ERROR, handle,
1780                  sizeof(chunk) - sizeof(chunk.h) + iov[1].iov_len);
1781     stl_be_p(&chunk.error, nbd_err);
1782     stw_be_p(&chunk.message_length, iov[1].iov_len);
1783 
1784     return nbd_co_send_iov(client, iov, 1 + !!iov[1].iov_len, errp);
1785 }
1786 
1787 /* Do a sparse read and send the structured reply to the client.
1788  * Returns -errno if sending fails. bdrv_block_status_above() failure is
1789  * reported to the client, at which point this function succeeds.
1790  */
1791 static int coroutine_fn nbd_co_send_sparse_read(NBDClient *client,
1792                                                 uint64_t handle,
1793                                                 uint64_t offset,
1794                                                 uint8_t *data,
1795                                                 size_t size,
1796                                                 Error **errp)
1797 {
1798     int ret = 0;
1799     NBDExport *exp = client->exp;
1800     size_t progress = 0;
1801 
1802     while (progress < size) {
1803         int64_t pnum;
1804         int status = bdrv_block_status_above(blk_bs(exp->blk), NULL,
1805                                              offset + progress,
1806                                              size - progress, &pnum, NULL,
1807                                              NULL);
1808         bool final;
1809 
1810         if (status < 0) {
1811             char *msg = g_strdup_printf("unable to check for holes: %s",
1812                                         strerror(-status));
1813 
1814             ret = nbd_co_send_structured_error(client, handle, -status, msg,
1815                                                errp);
1816             g_free(msg);
1817             return ret;
1818         }
1819         assert(pnum && pnum <= size - progress);
1820         final = progress + pnum == size;
1821         if (status & BDRV_BLOCK_ZERO) {
1822             NBDStructuredReadHole chunk;
1823             struct iovec iov[] = {
1824                 {.iov_base = &chunk, .iov_len = sizeof(chunk)},
1825             };
1826 
1827             trace_nbd_co_send_structured_read_hole(handle, offset + progress,
1828                                                    pnum);
1829             set_be_chunk(&chunk.h, final ? NBD_REPLY_FLAG_DONE : 0,
1830                          NBD_REPLY_TYPE_OFFSET_HOLE,
1831                          handle, sizeof(chunk) - sizeof(chunk.h));
1832             stq_be_p(&chunk.offset, offset + progress);
1833             stl_be_p(&chunk.length, pnum);
1834             ret = nbd_co_send_iov(client, iov, 1, errp);
1835         } else {
1836             ret = blk_pread(exp->blk, offset + progress + exp->dev_offset,
1837                             data + progress, pnum);
1838             if (ret < 0) {
1839                 error_setg_errno(errp, -ret, "reading from file failed");
1840                 break;
1841             }
1842             ret = nbd_co_send_structured_read(client, handle, offset + progress,
1843                                               data + progress, pnum, final,
1844                                               errp);
1845         }
1846 
1847         if (ret < 0) {
1848             break;
1849         }
1850         progress += pnum;
1851     }
1852     return ret;
1853 }
1854 
1855 /*
1856  * Populate @extents from block status. Update @bytes to be the actual
1857  * length encoded (which may be smaller than the original), and update
1858  * @nb_extents to the number of extents used.
1859  *
1860  * Returns zero on success and -errno on bdrv_block_status_above failure.
1861  */
1862 static int blockstatus_to_extents(BlockDriverState *bs, uint64_t offset,
1863                                   uint64_t *bytes, NBDExtent *extents,
1864                                   unsigned int *nb_extents)
1865 {
1866     uint64_t remaining_bytes = *bytes;
1867     NBDExtent *extent = extents, *extents_end = extents + *nb_extents;
1868     bool first_extent = true;
1869 
1870     assert(*nb_extents);
1871     while (remaining_bytes) {
1872         uint32_t flags;
1873         int64_t num;
1874         int ret = bdrv_block_status_above(bs, NULL, offset, remaining_bytes,
1875                                           &num, NULL, NULL);
1876 
1877         if (ret < 0) {
1878             return ret;
1879         }
1880 
1881         flags = (ret & BDRV_BLOCK_ALLOCATED ? 0 : NBD_STATE_HOLE) |
1882                 (ret & BDRV_BLOCK_ZERO      ? NBD_STATE_ZERO : 0);
1883         offset += num;
1884         remaining_bytes -= num;
1885 
1886         if (first_extent) {
1887             extent->flags = flags;
1888             extent->length = num;
1889             first_extent = false;
1890             continue;
1891         }
1892 
1893         if (flags == extent->flags) {
1894             /* extend current extent */
1895             extent->length += num;
1896         } else {
1897             if (extent + 1 == extents_end) {
1898                 break;
1899             }
1900 
1901             /* start new extent */
1902             extent++;
1903             extent->flags = flags;
1904             extent->length = num;
1905         }
1906     }
1907 
1908     extents_end = extent + 1;
1909 
1910     for (extent = extents; extent < extents_end; extent++) {
1911         extent->flags = cpu_to_be32(extent->flags);
1912         extent->length = cpu_to_be32(extent->length);
1913     }
1914 
1915     *bytes -= remaining_bytes;
1916     *nb_extents = extents_end - extents;
1917 
1918     return 0;
1919 }
1920 
1921 /* nbd_co_send_extents
1922  *
1923  * @length is only for tracing purposes (and may be smaller or larger
1924  * than the client's original request). @last controls whether
1925  * NBD_REPLY_FLAG_DONE is sent. @extents should already be in
1926  * big-endian format.
1927  */
1928 static int nbd_co_send_extents(NBDClient *client, uint64_t handle,
1929                                NBDExtent *extents, unsigned int nb_extents,
1930                                uint64_t length, bool last,
1931                                uint32_t context_id, Error **errp)
1932 {
1933     NBDStructuredMeta chunk;
1934 
1935     struct iovec iov[] = {
1936         {.iov_base = &chunk, .iov_len = sizeof(chunk)},
1937         {.iov_base = extents, .iov_len = nb_extents * sizeof(extents[0])}
1938     };
1939 
1940     trace_nbd_co_send_extents(handle, nb_extents, context_id, length, last);
1941     set_be_chunk(&chunk.h, last ? NBD_REPLY_FLAG_DONE : 0,
1942                  NBD_REPLY_TYPE_BLOCK_STATUS,
1943                  handle, sizeof(chunk) - sizeof(chunk.h) + iov[1].iov_len);
1944     stl_be_p(&chunk.context_id, context_id);
1945 
1946     return nbd_co_send_iov(client, iov, 2, errp);
1947 }
1948 
1949 /* Get block status from the exported device and send it to the client */
1950 static int nbd_co_send_block_status(NBDClient *client, uint64_t handle,
1951                                     BlockDriverState *bs, uint64_t offset,
1952                                     uint32_t length, bool dont_fragment,
1953                                     bool last, uint32_t context_id,
1954                                     Error **errp)
1955 {
1956     int ret;
1957     unsigned int nb_extents = dont_fragment ? 1 : NBD_MAX_BITMAP_EXTENTS;
1958     NBDExtent *extents = g_new(NBDExtent, nb_extents);
1959     uint64_t final_length = length;
1960 
1961     ret = blockstatus_to_extents(bs, offset, &final_length, extents,
1962                                  &nb_extents);
1963     if (ret < 0) {
1964         g_free(extents);
1965         return nbd_co_send_structured_error(
1966                 client, handle, -ret, "can't get block status", errp);
1967     }
1968 
1969     ret = nbd_co_send_extents(client, handle, extents, nb_extents,
1970                               final_length, last, context_id, errp);
1971 
1972     g_free(extents);
1973 
1974     return ret;
1975 }
1976 
1977 /*
1978  * Populate @extents from a dirty bitmap. Unless @dont_fragment, the
1979  * final extent may exceed the original @length. Store in @length the
1980  * byte length encoded (which may be smaller or larger than the
1981  * original), and return the number of extents used.
1982  */
1983 static unsigned int bitmap_to_extents(BdrvDirtyBitmap *bitmap, uint64_t offset,
1984                                       uint64_t *length, NBDExtent *extents,
1985                                       unsigned int nb_extents,
1986                                       bool dont_fragment)
1987 {
1988     uint64_t begin = offset, end = offset;
1989     uint64_t overall_end = offset + *length;
1990     unsigned int i = 0;
1991     BdrvDirtyBitmapIter *it;
1992     bool dirty;
1993 
1994     bdrv_dirty_bitmap_lock(bitmap);
1995 
1996     it = bdrv_dirty_iter_new(bitmap);
1997     dirty = bdrv_get_dirty_locked(NULL, bitmap, offset);
1998 
1999     assert(begin < overall_end && nb_extents);
2000     while (begin < overall_end && i < nb_extents) {
2001         bool next_dirty = !dirty;
2002 
2003         if (dirty) {
2004             end = bdrv_dirty_bitmap_next_zero(bitmap, begin, UINT64_MAX);
2005         } else {
2006             bdrv_set_dirty_iter(it, begin);
2007             end = bdrv_dirty_iter_next(it);
2008         }
2009         if (end == -1 || end - begin > UINT32_MAX) {
2010             /* Cap to an aligned value < 4G beyond begin. */
2011             end = MIN(bdrv_dirty_bitmap_size(bitmap),
2012                       begin + UINT32_MAX + 1 -
2013                       bdrv_dirty_bitmap_granularity(bitmap));
2014             next_dirty = dirty;
2015         }
2016         if (dont_fragment && end > overall_end) {
2017             end = overall_end;
2018         }
2019 
2020         extents[i].length = cpu_to_be32(end - begin);
2021         extents[i].flags = cpu_to_be32(dirty ? NBD_STATE_DIRTY : 0);
2022         i++;
2023         begin = end;
2024         dirty = next_dirty;
2025     }
2026 
2027     bdrv_dirty_iter_free(it);
2028 
2029     bdrv_dirty_bitmap_unlock(bitmap);
2030 
2031     assert(offset < end);
2032     *length = end - offset;
2033     return i;
2034 }
2035 
2036 static int nbd_co_send_bitmap(NBDClient *client, uint64_t handle,
2037                               BdrvDirtyBitmap *bitmap, uint64_t offset,
2038                               uint32_t length, bool dont_fragment, bool last,
2039                               uint32_t context_id, Error **errp)
2040 {
2041     int ret;
2042     unsigned int nb_extents = dont_fragment ? 1 : NBD_MAX_BITMAP_EXTENTS;
2043     NBDExtent *extents = g_new(NBDExtent, nb_extents);
2044     uint64_t final_length = length;
2045 
2046     nb_extents = bitmap_to_extents(bitmap, offset, &final_length, extents,
2047                                    nb_extents, dont_fragment);
2048 
2049     ret = nbd_co_send_extents(client, handle, extents, nb_extents,
2050                               final_length, last, context_id, errp);
2051 
2052     g_free(extents);
2053 
2054     return ret;
2055 }
2056 
2057 /* nbd_co_receive_request
2058  * Collect a client request. Return 0 if request looks valid, -EIO to drop
2059  * connection right away, and any other negative value to report an error to
2060  * the client (although the caller may still need to disconnect after reporting
2061  * the error).
2062  */
2063 static int nbd_co_receive_request(NBDRequestData *req, NBDRequest *request,
2064                                   Error **errp)
2065 {
2066     NBDClient *client = req->client;
2067     int valid_flags;
2068 
2069     g_assert(qemu_in_coroutine());
2070     assert(client->recv_coroutine == qemu_coroutine_self());
2071     if (nbd_receive_request(client->ioc, request, errp) < 0) {
2072         return -EIO;
2073     }
2074 
2075     trace_nbd_co_receive_request_decode_type(request->handle, request->type,
2076                                              nbd_cmd_lookup(request->type));
2077 
2078     if (request->type != NBD_CMD_WRITE) {
2079         /* No payload, we are ready to read the next request.  */
2080         req->complete = true;
2081     }
2082 
2083     if (request->type == NBD_CMD_DISC) {
2084         /* Special case: we're going to disconnect without a reply,
2085          * whether or not flags, from, or len are bogus */
2086         return -EIO;
2087     }
2088 
2089     if (request->type == NBD_CMD_READ || request->type == NBD_CMD_WRITE ||
2090         request->type == NBD_CMD_CACHE)
2091     {
2092         if (request->len > NBD_MAX_BUFFER_SIZE) {
2093             error_setg(errp, "len (%" PRIu32" ) is larger than max len (%u)",
2094                        request->len, NBD_MAX_BUFFER_SIZE);
2095             return -EINVAL;
2096         }
2097 
2098         req->data = blk_try_blockalign(client->exp->blk, request->len);
2099         if (req->data == NULL) {
2100             error_setg(errp, "No memory");
2101             return -ENOMEM;
2102         }
2103     }
2104     if (request->type == NBD_CMD_WRITE) {
2105         if (nbd_read(client->ioc, req->data, request->len, "CMD_WRITE data",
2106                      errp) < 0)
2107         {
2108             return -EIO;
2109         }
2110         req->complete = true;
2111 
2112         trace_nbd_co_receive_request_payload_received(request->handle,
2113                                                       request->len);
2114     }
2115 
2116     /* Sanity checks. */
2117     if (client->exp->nbdflags & NBD_FLAG_READ_ONLY &&
2118         (request->type == NBD_CMD_WRITE ||
2119          request->type == NBD_CMD_WRITE_ZEROES ||
2120          request->type == NBD_CMD_TRIM)) {
2121         error_setg(errp, "Export is read-only");
2122         return -EROFS;
2123     }
2124     if (request->from > client->exp->size ||
2125         request->len > client->exp->size - request->from) {
2126         error_setg(errp, "operation past EOF; From: %" PRIu64 ", Len: %" PRIu32
2127                    ", Size: %" PRIu64, request->from, request->len,
2128                    client->exp->size);
2129         return (request->type == NBD_CMD_WRITE ||
2130                 request->type == NBD_CMD_WRITE_ZEROES) ? -ENOSPC : -EINVAL;
2131     }
2132     valid_flags = NBD_CMD_FLAG_FUA;
2133     if (request->type == NBD_CMD_READ && client->structured_reply) {
2134         valid_flags |= NBD_CMD_FLAG_DF;
2135     } else if (request->type == NBD_CMD_WRITE_ZEROES) {
2136         valid_flags |= NBD_CMD_FLAG_NO_HOLE;
2137     } else if (request->type == NBD_CMD_BLOCK_STATUS) {
2138         valid_flags |= NBD_CMD_FLAG_REQ_ONE;
2139     }
2140     if (request->flags & ~valid_flags) {
2141         error_setg(errp, "unsupported flags for command %s (got 0x%x)",
2142                    nbd_cmd_lookup(request->type), request->flags);
2143         return -EINVAL;
2144     }
2145 
2146     return 0;
2147 }
2148 
2149 /* Send simple reply without a payload, or a structured error
2150  * @error_msg is ignored if @ret >= 0
2151  * Returns 0 if connection is still live, -errno on failure to talk to client
2152  */
2153 static coroutine_fn int nbd_send_generic_reply(NBDClient *client,
2154                                                uint64_t handle,
2155                                                int ret,
2156                                                const char *error_msg,
2157                                                Error **errp)
2158 {
2159     if (client->structured_reply && ret < 0) {
2160         return nbd_co_send_structured_error(client, handle, -ret, error_msg,
2161                                             errp);
2162     } else {
2163         return nbd_co_send_simple_reply(client, handle, ret < 0 ? -ret : 0,
2164                                         NULL, 0, errp);
2165     }
2166 }
2167 
2168 /* Handle NBD_CMD_READ request.
2169  * Return -errno if sending fails. Other errors are reported directly to the
2170  * client as an error reply. */
2171 static coroutine_fn int nbd_do_cmd_read(NBDClient *client, NBDRequest *request,
2172                                         uint8_t *data, Error **errp)
2173 {
2174     int ret;
2175     NBDExport *exp = client->exp;
2176 
2177     assert(request->type == NBD_CMD_READ || request->type == NBD_CMD_CACHE);
2178 
2179     /* XXX: NBD Protocol only documents use of FUA with WRITE */
2180     if (request->flags & NBD_CMD_FLAG_FUA) {
2181         ret = blk_co_flush(exp->blk);
2182         if (ret < 0) {
2183             return nbd_send_generic_reply(client, request->handle, ret,
2184                                           "flush failed", errp);
2185         }
2186     }
2187 
2188     if (client->structured_reply && !(request->flags & NBD_CMD_FLAG_DF) &&
2189         request->len && request->type != NBD_CMD_CACHE)
2190     {
2191         return nbd_co_send_sparse_read(client, request->handle, request->from,
2192                                        data, request->len, errp);
2193     }
2194 
2195     ret = blk_pread(exp->blk, request->from + exp->dev_offset, data,
2196                     request->len);
2197     if (ret < 0 || request->type == NBD_CMD_CACHE) {
2198         return nbd_send_generic_reply(client, request->handle, ret,
2199                                       "reading from file failed", errp);
2200     }
2201 
2202     if (client->structured_reply) {
2203         if (request->len) {
2204             return nbd_co_send_structured_read(client, request->handle,
2205                                                request->from, data,
2206                                                request->len, true, errp);
2207         } else {
2208             return nbd_co_send_structured_done(client, request->handle, errp);
2209         }
2210     } else {
2211         return nbd_co_send_simple_reply(client, request->handle, 0,
2212                                         data, request->len, errp);
2213     }
2214 }
2215 
2216 /* Handle NBD request.
2217  * Return -errno if sending fails. Other errors are reported directly to the
2218  * client as an error reply. */
2219 static coroutine_fn int nbd_handle_request(NBDClient *client,
2220                                            NBDRequest *request,
2221                                            uint8_t *data, Error **errp)
2222 {
2223     int ret;
2224     int flags;
2225     NBDExport *exp = client->exp;
2226     char *msg;
2227 
2228     switch (request->type) {
2229     case NBD_CMD_READ:
2230     case NBD_CMD_CACHE:
2231         return nbd_do_cmd_read(client, request, data, errp);
2232 
2233     case NBD_CMD_WRITE:
2234         flags = 0;
2235         if (request->flags & NBD_CMD_FLAG_FUA) {
2236             flags |= BDRV_REQ_FUA;
2237         }
2238         ret = blk_pwrite(exp->blk, request->from + exp->dev_offset,
2239                          data, request->len, flags);
2240         return nbd_send_generic_reply(client, request->handle, ret,
2241                                       "writing to file failed", errp);
2242 
2243     case NBD_CMD_WRITE_ZEROES:
2244         flags = 0;
2245         if (request->flags & NBD_CMD_FLAG_FUA) {
2246             flags |= BDRV_REQ_FUA;
2247         }
2248         if (!(request->flags & NBD_CMD_FLAG_NO_HOLE)) {
2249             flags |= BDRV_REQ_MAY_UNMAP;
2250         }
2251         ret = blk_pwrite_zeroes(exp->blk, request->from + exp->dev_offset,
2252                                 request->len, flags);
2253         return nbd_send_generic_reply(client, request->handle, ret,
2254                                       "writing to file failed", errp);
2255 
2256     case NBD_CMD_DISC:
2257         /* unreachable, thanks to special case in nbd_co_receive_request() */
2258         abort();
2259 
2260     case NBD_CMD_FLUSH:
2261         ret = blk_co_flush(exp->blk);
2262         return nbd_send_generic_reply(client, request->handle, ret,
2263                                       "flush failed", errp);
2264 
2265     case NBD_CMD_TRIM:
2266         ret = blk_co_pdiscard(exp->blk, request->from + exp->dev_offset,
2267                               request->len);
2268         if (ret == 0 && request->flags & NBD_CMD_FLAG_FUA) {
2269             ret = blk_co_flush(exp->blk);
2270         }
2271         return nbd_send_generic_reply(client, request->handle, ret,
2272                                       "discard failed", errp);
2273 
2274     case NBD_CMD_BLOCK_STATUS:
2275         if (!request->len) {
2276             return nbd_send_generic_reply(client, request->handle, -EINVAL,
2277                                           "need non-zero length", errp);
2278         }
2279         if (client->export_meta.valid &&
2280             (client->export_meta.base_allocation ||
2281              client->export_meta.bitmap))
2282         {
2283             bool dont_fragment = request->flags & NBD_CMD_FLAG_REQ_ONE;
2284 
2285             if (client->export_meta.base_allocation) {
2286                 ret = nbd_co_send_block_status(client, request->handle,
2287                                                blk_bs(exp->blk), request->from,
2288                                                request->len, dont_fragment,
2289                                                !client->export_meta.bitmap,
2290                                                NBD_META_ID_BASE_ALLOCATION,
2291                                                errp);
2292                 if (ret < 0) {
2293                     return ret;
2294                 }
2295             }
2296 
2297             if (client->export_meta.bitmap) {
2298                 ret = nbd_co_send_bitmap(client, request->handle,
2299                                          client->exp->export_bitmap,
2300                                          request->from, request->len,
2301                                          dont_fragment,
2302                                          true, NBD_META_ID_DIRTY_BITMAP, errp);
2303                 if (ret < 0) {
2304                     return ret;
2305                 }
2306             }
2307 
2308             return ret;
2309         } else {
2310             return nbd_send_generic_reply(client, request->handle, -EINVAL,
2311                                           "CMD_BLOCK_STATUS not negotiated",
2312                                           errp);
2313         }
2314 
2315     default:
2316         msg = g_strdup_printf("invalid request type (%" PRIu32 ") received",
2317                               request->type);
2318         ret = nbd_send_generic_reply(client, request->handle, -EINVAL, msg,
2319                                      errp);
2320         g_free(msg);
2321         return ret;
2322     }
2323 }
2324 
2325 /* Owns a reference to the NBDClient passed as opaque.  */
2326 static coroutine_fn void nbd_trip(void *opaque)
2327 {
2328     NBDClient *client = opaque;
2329     NBDRequestData *req;
2330     NBDRequest request = { 0 };    /* GCC thinks it can be used uninitialized */
2331     int ret;
2332     Error *local_err = NULL;
2333 
2334     trace_nbd_trip();
2335     if (client->closing) {
2336         nbd_client_put(client);
2337         return;
2338     }
2339 
2340     req = nbd_request_get(client);
2341     ret = nbd_co_receive_request(req, &request, &local_err);
2342     client->recv_coroutine = NULL;
2343 
2344     if (client->closing) {
2345         /*
2346          * The client may be closed when we are blocked in
2347          * nbd_co_receive_request()
2348          */
2349         goto done;
2350     }
2351 
2352     nbd_client_receive_next_request(client);
2353     if (ret == -EIO) {
2354         goto disconnect;
2355     }
2356 
2357     if (ret < 0) {
2358         /* It wans't -EIO, so, according to nbd_co_receive_request()
2359          * semantics, we should return the error to the client. */
2360         Error *export_err = local_err;
2361 
2362         local_err = NULL;
2363         ret = nbd_send_generic_reply(client, request.handle, -EINVAL,
2364                                      error_get_pretty(export_err), &local_err);
2365         error_free(export_err);
2366     } else {
2367         ret = nbd_handle_request(client, &request, req->data, &local_err);
2368     }
2369     if (ret < 0) {
2370         error_prepend(&local_err, "Failed to send reply: ");
2371         goto disconnect;
2372     }
2373 
2374     /* We must disconnect after NBD_CMD_WRITE if we did not
2375      * read the payload.
2376      */
2377     if (!req->complete) {
2378         error_setg(&local_err, "Request handling failed in intermediate state");
2379         goto disconnect;
2380     }
2381 
2382 done:
2383     nbd_request_put(req);
2384     nbd_client_put(client);
2385     return;
2386 
2387 disconnect:
2388     if (local_err) {
2389         error_reportf_err(local_err, "Disconnect client, due to: ");
2390     }
2391     nbd_request_put(req);
2392     client_close(client, true);
2393     nbd_client_put(client);
2394 }
2395 
2396 static void nbd_client_receive_next_request(NBDClient *client)
2397 {
2398     if (!client->recv_coroutine && client->nb_requests < MAX_NBD_REQUESTS) {
2399         nbd_client_get(client);
2400         client->recv_coroutine = qemu_coroutine_create(nbd_trip, client);
2401         aio_co_schedule(client->exp->ctx, client->recv_coroutine);
2402     }
2403 }
2404 
2405 static coroutine_fn void nbd_co_client_start(void *opaque)
2406 {
2407     NBDClient *client = opaque;
2408     Error *local_err = NULL;
2409 
2410     qemu_co_mutex_init(&client->send_lock);
2411 
2412     if (nbd_negotiate(client, &local_err)) {
2413         if (local_err) {
2414             error_report_err(local_err);
2415         }
2416         client_close(client, false);
2417         return;
2418     }
2419 
2420     nbd_client_receive_next_request(client);
2421 }
2422 
2423 /*
2424  * Create a new client listener using the given channel @sioc.
2425  * Begin servicing it in a coroutine.  When the connection closes, call
2426  * @close_fn with an indication of whether the client completed negotiation.
2427  */
2428 void nbd_client_new(QIOChannelSocket *sioc,
2429                     QCryptoTLSCreds *tlscreds,
2430                     const char *tlsauthz,
2431                     void (*close_fn)(NBDClient *, bool))
2432 {
2433     NBDClient *client;
2434     Coroutine *co;
2435 
2436     client = g_new0(NBDClient, 1);
2437     client->refcount = 1;
2438     client->tlscreds = tlscreds;
2439     if (tlscreds) {
2440         object_ref(OBJECT(client->tlscreds));
2441     }
2442     client->tlsauthz = g_strdup(tlsauthz);
2443     client->sioc = sioc;
2444     object_ref(OBJECT(client->sioc));
2445     client->ioc = QIO_CHANNEL(sioc);
2446     object_ref(OBJECT(client->ioc));
2447     client->close_fn = close_fn;
2448 
2449     co = qemu_coroutine_create(nbd_co_client_start, client);
2450     qemu_coroutine_enter(co);
2451 }
2452