+}
+
+static int nbd_config(BDRVNBDState *s, QDict *options)
+{
+ Error *local_err = NULL;
+
+ if (qdict_haskey(options, "path")) {
+ if (qdict_haskey(options, "host")) {
+ qerror_report(ERROR_CLASS_GENERIC_ERROR, "path and host may not "
+ "be used at the same time.");
+ return -EINVAL;
+ }
+ s->is_unix = true;
+ } else if (qdict_haskey(options, "host")) {
+ s->is_unix = false;
+ } else {
+ return -EINVAL;
+ }
+
+ s->socket_opts = qemu_opts_create_nofail(&socket_optslist);
+
+ qemu_opts_absorb_qdict(s->socket_opts, options, &local_err);
+ if (error_is_set(&local_err)) {
+ qerror_report_err(local_err);
+ error_free(local_err);
+ return -EINVAL;
+ }
+
+ if (!qemu_opt_get(s->socket_opts, "port")) {
+ qemu_opt_set_number(s->socket_opts, "port", NBD_DEFAULT_PORT);
+ }
+
+ s->export_name = g_strdup(qdict_get_try_str(options, "export"));
+ if (s->export_name) {
+ qdict_del(options, "export");
+ }
+
+ return 0;
+}
+
+
+static void nbd_coroutine_start(BDRVNBDState *s, struct nbd_request *request)
+{
+ int i;
+
+ /* Poor man semaphore. The free_sema is locked when no other request
+ * can be accepted, and unlocked after receiving one reply. */
+ if (s->in_flight >= MAX_NBD_REQUESTS - 1) {
+ qemu_co_mutex_lock(&s->free_sema);
+ assert(s->in_flight < MAX_NBD_REQUESTS);
+ }
+ s->in_flight++;
+
+ for (i = 0; i < MAX_NBD_REQUESTS; i++) {
+ if (s->recv_coroutine[i] == NULL) {
+ s->recv_coroutine[i] = qemu_coroutine_self();
+ break;
+ }
+ }
+
+ assert(i < MAX_NBD_REQUESTS);
+ request->handle = INDEX_TO_HANDLE(s, i);
+}
+
+static int nbd_have_request(void *opaque)
+{
+ BDRVNBDState *s = opaque;
+
+ return s->in_flight > 0;
+}
+
+static void nbd_reply_ready(void *opaque)
+{
+ BDRVNBDState *s = opaque;
+ uint64_t i;
+ int ret;
+
+ if (s->reply.handle == 0) {
+ /* No reply already in flight. Fetch a header. It is possible
+ * that another thread has done the same thing in parallel, so
+ * the socket is not readable anymore.
+ */
+ ret = nbd_receive_reply(s->sock, &s->reply);
+ if (ret == -EAGAIN) {
+ return;
+ }
+ if (ret < 0) {
+ s->reply.handle = 0;
+ goto fail;
+ }
+ }
+
+ /* There's no need for a mutex on the receive side, because the
+ * handler acts as a synchronization point and ensures that only
+ * one coroutine is called until the reply finishes. */
+ i = HANDLE_TO_INDEX(s, s->reply.handle);
+ if (i >= MAX_NBD_REQUESTS) {
+ goto fail;
+ }
+
+ if (s->recv_coroutine[i]) {
+ qemu_coroutine_enter(s->recv_coroutine[i], NULL);
+ return;
+ }
+
+fail:
+ for (i = 0; i < MAX_NBD_REQUESTS; i++) {
+ if (s->recv_coroutine[i]) {
+ qemu_coroutine_enter(s->recv_coroutine[i], NULL);
+ }
+ }
+}
+
+static void nbd_restart_write(void *opaque)
+{
+ BDRVNBDState *s = opaque;
+ qemu_coroutine_enter(s->send_coroutine, NULL);
+}
+
+static int nbd_co_send_request(BDRVNBDState *s, struct nbd_request *request,
+ QEMUIOVector *qiov, int offset)
+{
+ int rc, ret;
+
+ qemu_co_mutex_lock(&s->send_mutex);
+ s->send_coroutine = qemu_coroutine_self();
+ qemu_aio_set_fd_handler(s->sock, nbd_reply_ready, nbd_restart_write,
+ nbd_have_request, s);
+ if (qiov) {
+ if (!s->is_unix) {
+ socket_set_cork(s->sock, 1);
+ }
+ rc = nbd_send_request(s->sock, request);
+ if (rc >= 0) {
+ ret = qemu_co_sendv(s->sock, qiov->iov, qiov->niov,
+ offset, request->len);
+ if (ret != request->len) {
+ rc = -EIO;
+ }
+ }
+ if (!s->is_unix) {
+ socket_set_cork(s->sock, 0);
+ }
+ } else {
+ rc = nbd_send_request(s->sock, request);
+ }
+ qemu_aio_set_fd_handler(s->sock, nbd_reply_ready, NULL,
+ nbd_have_request, s);
+ s->send_coroutine = NULL;
+ qemu_co_mutex_unlock(&s->send_mutex);
+ return rc;
+}
+
+static void nbd_co_receive_reply(BDRVNBDState *s, struct nbd_request *request,
+ struct nbd_reply *reply,
+ QEMUIOVector *qiov, int offset)
+{
+ int ret;
+
+ /* Wait until we're woken up by the read handler. TODO: perhaps
+ * peek at the next reply and avoid yielding if it's ours? */
+ qemu_coroutine_yield();
+ *reply = s->reply;
+ if (reply->handle != request->handle) {
+ reply->error = EIO;
+ } else {
+ if (qiov && reply->error == 0) {
+ ret = qemu_co_recvv(s->sock, qiov->iov, qiov->niov,
+ offset, request->len);
+ if (ret != request->len) {
+ reply->error = EIO;
+ }
+ }
+
+ /* Tell the read handler to read another header. */
+ s->reply.handle = 0;
+ }
+}
+
+static void nbd_coroutine_end(BDRVNBDState *s, struct nbd_request *request)
+{
+ int i = HANDLE_TO_INDEX(s, request->handle);
+ s->recv_coroutine[i] = NULL;
+ if (s->in_flight-- == MAX_NBD_REQUESTS) {
+ qemu_co_mutex_unlock(&s->free_sema);