migration/rdma: Send error during cancelling
[qemu/ar7.git] / io / channel-socket.c
blob53386b7ba398b8aa50458d764e07dbe6d70112a2
1 /*
2 * QEMU I/O channels sockets driver
4 * Copyright (c) 2015 Red Hat, Inc.
6 * This library is free software; you can redistribute it and/or
7 * modify it under the terms of the GNU Lesser General Public
8 * License as published by the Free Software Foundation; either
9 * version 2 of the License, or (at your option) any later version.
11 * This library 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 GNU
14 * Lesser General Public License for more details.
16 * You should have received a copy of the GNU Lesser General Public
17 * License along with this library; if not, see <http://www.gnu.org/licenses/>.
21 #include "qemu/osdep.h"
22 #include "qapi/error.h"
23 #include "io/channel-socket.h"
24 #include "io/channel-watch.h"
25 #include "trace.h"
26 #include "qapi/clone-visitor.h"
28 #define SOCKET_MAX_FDS 16
30 SocketAddress *
31 qio_channel_socket_get_local_address(QIOChannelSocket *ioc,
32 Error **errp)
34 return socket_sockaddr_to_address(&ioc->localAddr,
35 ioc->localAddrLen,
36 errp);
39 SocketAddress *
40 qio_channel_socket_get_remote_address(QIOChannelSocket *ioc,
41 Error **errp)
43 return socket_sockaddr_to_address(&ioc->remoteAddr,
44 ioc->remoteAddrLen,
45 errp);
48 QIOChannelSocket *
49 qio_channel_socket_new(void)
51 QIOChannelSocket *sioc;
52 QIOChannel *ioc;
54 sioc = QIO_CHANNEL_SOCKET(object_new(TYPE_QIO_CHANNEL_SOCKET));
55 sioc->fd = -1;
57 ioc = QIO_CHANNEL(sioc);
58 qio_channel_set_feature(ioc, QIO_CHANNEL_FEATURE_SHUTDOWN);
60 #ifdef WIN32
61 ioc->event = CreateEvent(NULL, FALSE, FALSE, NULL);
62 #endif
64 trace_qio_channel_socket_new(sioc);
66 return sioc;
70 static int
71 qio_channel_socket_set_fd(QIOChannelSocket *sioc,
72 int fd,
73 Error **errp)
75 if (sioc->fd != -1) {
76 error_setg(errp, "Socket is already open");
77 return -1;
80 sioc->fd = fd;
81 sioc->remoteAddrLen = sizeof(sioc->remoteAddr);
82 sioc->localAddrLen = sizeof(sioc->localAddr);
85 if (getpeername(fd, (struct sockaddr *)&sioc->remoteAddr,
86 &sioc->remoteAddrLen) < 0) {
87 if (errno == ENOTCONN) {
88 memset(&sioc->remoteAddr, 0, sizeof(sioc->remoteAddr));
89 sioc->remoteAddrLen = sizeof(sioc->remoteAddr);
90 } else {
91 error_setg_errno(errp, errno,
92 "Unable to query remote socket address");
93 goto error;
97 if (getsockname(fd, (struct sockaddr *)&sioc->localAddr,
98 &sioc->localAddrLen) < 0) {
99 error_setg_errno(errp, errno,
100 "Unable to query local socket address");
101 goto error;
104 #ifndef WIN32
105 if (sioc->localAddr.ss_family == AF_UNIX) {
106 QIOChannel *ioc = QIO_CHANNEL(sioc);
107 qio_channel_set_feature(ioc, QIO_CHANNEL_FEATURE_FD_PASS);
109 #endif /* WIN32 */
111 return 0;
113 error:
114 sioc->fd = -1; /* Let the caller close FD on failure */
115 return -1;
118 QIOChannelSocket *
119 qio_channel_socket_new_fd(int fd,
120 Error **errp)
122 QIOChannelSocket *ioc;
124 ioc = qio_channel_socket_new();
125 if (qio_channel_socket_set_fd(ioc, fd, errp) < 0) {
126 object_unref(OBJECT(ioc));
127 return NULL;
130 trace_qio_channel_socket_new_fd(ioc, fd);
132 return ioc;
136 int qio_channel_socket_connect_sync(QIOChannelSocket *ioc,
137 SocketAddress *addr,
138 Error **errp)
140 int fd;
142 trace_qio_channel_socket_connect_sync(ioc, addr);
143 fd = socket_connect(addr, NULL, NULL, errp);
144 if (fd < 0) {
145 trace_qio_channel_socket_connect_fail(ioc);
146 return -1;
149 trace_qio_channel_socket_connect_complete(ioc, fd);
150 if (qio_channel_socket_set_fd(ioc, fd, errp) < 0) {
151 close(fd);
152 return -1;
155 return 0;
159 static void qio_channel_socket_connect_worker(QIOTask *task,
160 gpointer opaque)
162 QIOChannelSocket *ioc = QIO_CHANNEL_SOCKET(qio_task_get_source(task));
163 SocketAddress *addr = opaque;
164 Error *err = NULL;
166 qio_channel_socket_connect_sync(ioc, addr, &err);
168 qio_task_set_error(task, err);
172 void qio_channel_socket_connect_async(QIOChannelSocket *ioc,
173 SocketAddress *addr,
174 QIOTaskFunc callback,
175 gpointer opaque,
176 GDestroyNotify destroy)
178 QIOTask *task = qio_task_new(
179 OBJECT(ioc), callback, opaque, destroy);
180 SocketAddress *addrCopy;
182 addrCopy = QAPI_CLONE(SocketAddress, addr);
184 /* socket_connect() does a non-blocking connect(), but it
185 * still blocks in DNS lookups, so we must use a thread */
186 trace_qio_channel_socket_connect_async(ioc, addr);
187 qio_task_run_in_thread(task,
188 qio_channel_socket_connect_worker,
189 addrCopy,
190 (GDestroyNotify)qapi_free_SocketAddress);
194 int qio_channel_socket_listen_sync(QIOChannelSocket *ioc,
195 SocketAddress *addr,
196 Error **errp)
198 int fd;
200 trace_qio_channel_socket_listen_sync(ioc, addr);
201 fd = socket_listen(addr, errp);
202 if (fd < 0) {
203 trace_qio_channel_socket_listen_fail(ioc);
204 return -1;
207 trace_qio_channel_socket_listen_complete(ioc, fd);
208 if (qio_channel_socket_set_fd(ioc, fd, errp) < 0) {
209 close(fd);
210 return -1;
212 qio_channel_set_feature(QIO_CHANNEL(ioc), QIO_CHANNEL_FEATURE_LISTEN);
214 return 0;
218 static void qio_channel_socket_listen_worker(QIOTask *task,
219 gpointer opaque)
221 QIOChannelSocket *ioc = QIO_CHANNEL_SOCKET(qio_task_get_source(task));
222 SocketAddress *addr = opaque;
223 Error *err = NULL;
225 qio_channel_socket_listen_sync(ioc, addr, &err);
227 qio_task_set_error(task, err);
231 void qio_channel_socket_listen_async(QIOChannelSocket *ioc,
232 SocketAddress *addr,
233 QIOTaskFunc callback,
234 gpointer opaque,
235 GDestroyNotify destroy)
237 QIOTask *task = qio_task_new(
238 OBJECT(ioc), callback, opaque, destroy);
239 SocketAddress *addrCopy;
241 addrCopy = QAPI_CLONE(SocketAddress, addr);
243 /* socket_listen() blocks in DNS lookups, so we must use a thread */
244 trace_qio_channel_socket_listen_async(ioc, addr);
245 qio_task_run_in_thread(task,
246 qio_channel_socket_listen_worker,
247 addrCopy,
248 (GDestroyNotify)qapi_free_SocketAddress);
252 int qio_channel_socket_dgram_sync(QIOChannelSocket *ioc,
253 SocketAddress *localAddr,
254 SocketAddress *remoteAddr,
255 Error **errp)
257 int fd;
259 trace_qio_channel_socket_dgram_sync(ioc, localAddr, remoteAddr);
260 fd = socket_dgram(remoteAddr, localAddr, errp);
261 if (fd < 0) {
262 trace_qio_channel_socket_dgram_fail(ioc);
263 return -1;
266 trace_qio_channel_socket_dgram_complete(ioc, fd);
267 if (qio_channel_socket_set_fd(ioc, fd, errp) < 0) {
268 close(fd);
269 return -1;
272 return 0;
276 struct QIOChannelSocketDGramWorkerData {
277 SocketAddress *localAddr;
278 SocketAddress *remoteAddr;
282 static void qio_channel_socket_dgram_worker_free(gpointer opaque)
284 struct QIOChannelSocketDGramWorkerData *data = opaque;
285 qapi_free_SocketAddress(data->localAddr);
286 qapi_free_SocketAddress(data->remoteAddr);
287 g_free(data);
290 static void qio_channel_socket_dgram_worker(QIOTask *task,
291 gpointer opaque)
293 QIOChannelSocket *ioc = QIO_CHANNEL_SOCKET(qio_task_get_source(task));
294 struct QIOChannelSocketDGramWorkerData *data = opaque;
295 Error *err = NULL;
297 /* socket_dgram() blocks in DNS lookups, so we must use a thread */
298 qio_channel_socket_dgram_sync(ioc, data->localAddr,
299 data->remoteAddr, &err);
301 qio_task_set_error(task, err);
305 void qio_channel_socket_dgram_async(QIOChannelSocket *ioc,
306 SocketAddress *localAddr,
307 SocketAddress *remoteAddr,
308 QIOTaskFunc callback,
309 gpointer opaque,
310 GDestroyNotify destroy)
312 QIOTask *task = qio_task_new(
313 OBJECT(ioc), callback, opaque, destroy);
314 struct QIOChannelSocketDGramWorkerData *data = g_new0(
315 struct QIOChannelSocketDGramWorkerData, 1);
317 data->localAddr = QAPI_CLONE(SocketAddress, localAddr);
318 data->remoteAddr = QAPI_CLONE(SocketAddress, remoteAddr);
320 trace_qio_channel_socket_dgram_async(ioc, localAddr, remoteAddr);
321 qio_task_run_in_thread(task,
322 qio_channel_socket_dgram_worker,
323 data,
324 qio_channel_socket_dgram_worker_free);
328 QIOChannelSocket *
329 qio_channel_socket_accept(QIOChannelSocket *ioc,
330 Error **errp)
332 QIOChannelSocket *cioc;
334 cioc = qio_channel_socket_new();
335 cioc->remoteAddrLen = sizeof(ioc->remoteAddr);
336 cioc->localAddrLen = sizeof(ioc->localAddr);
338 retry:
339 trace_qio_channel_socket_accept(ioc);
340 cioc->fd = qemu_accept(ioc->fd, (struct sockaddr *)&cioc->remoteAddr,
341 &cioc->remoteAddrLen);
342 if (cioc->fd < 0) {
343 trace_qio_channel_socket_accept_fail(ioc);
344 if (errno == EINTR) {
345 goto retry;
347 goto error;
350 if (getsockname(cioc->fd, (struct sockaddr *)&cioc->localAddr,
351 &cioc->localAddrLen) < 0) {
352 error_setg_errno(errp, errno,
353 "Unable to query local socket address");
354 goto error;
357 #ifndef WIN32
358 if (cioc->localAddr.ss_family == AF_UNIX) {
359 QIOChannel *ioc_local = QIO_CHANNEL(cioc);
360 qio_channel_set_feature(ioc_local, QIO_CHANNEL_FEATURE_FD_PASS);
362 #endif /* WIN32 */
364 trace_qio_channel_socket_accept_complete(ioc, cioc, cioc->fd);
365 return cioc;
367 error:
368 object_unref(OBJECT(cioc));
369 return NULL;
372 static void qio_channel_socket_init(Object *obj)
374 QIOChannelSocket *ioc = QIO_CHANNEL_SOCKET(obj);
375 ioc->fd = -1;
378 static void qio_channel_socket_finalize(Object *obj)
380 QIOChannelSocket *ioc = QIO_CHANNEL_SOCKET(obj);
382 if (ioc->fd != -1) {
383 QIOChannel *ioc_local = QIO_CHANNEL(ioc);
384 if (qio_channel_has_feature(ioc_local, QIO_CHANNEL_FEATURE_LISTEN)) {
385 Error *err = NULL;
387 socket_listen_cleanup(ioc->fd, &err);
388 if (err) {
389 error_report_err(err);
390 err = NULL;
393 #ifdef WIN32
394 WSAEventSelect(ioc->fd, NULL, 0);
395 #endif
396 closesocket(ioc->fd);
397 ioc->fd = -1;
402 #ifndef WIN32
403 static void qio_channel_socket_copy_fds(struct msghdr *msg,
404 int **fds, size_t *nfds)
406 struct cmsghdr *cmsg;
408 *nfds = 0;
409 *fds = NULL;
411 for (cmsg = CMSG_FIRSTHDR(msg); cmsg; cmsg = CMSG_NXTHDR(msg, cmsg)) {
412 int fd_size, i;
413 int gotfds;
415 if (cmsg->cmsg_len < CMSG_LEN(sizeof(int)) ||
416 cmsg->cmsg_level != SOL_SOCKET ||
417 cmsg->cmsg_type != SCM_RIGHTS) {
418 continue;
421 fd_size = cmsg->cmsg_len - CMSG_LEN(0);
423 if (!fd_size) {
424 continue;
427 gotfds = fd_size / sizeof(int);
428 *fds = g_renew(int, *fds, *nfds + gotfds);
429 memcpy(*fds + *nfds, CMSG_DATA(cmsg), fd_size);
431 for (i = 0; i < gotfds; i++) {
432 int fd = (*fds)[*nfds + i];
433 if (fd < 0) {
434 continue;
437 /* O_NONBLOCK is preserved across SCM_RIGHTS so reset it */
438 qemu_set_block(fd);
440 #ifndef MSG_CMSG_CLOEXEC
441 qemu_set_cloexec(fd);
442 #endif
444 *nfds += gotfds;
449 static ssize_t qio_channel_socket_readv(QIOChannel *ioc,
450 const struct iovec *iov,
451 size_t niov,
452 int **fds,
453 size_t *nfds,
454 Error **errp)
456 QIOChannelSocket *sioc = QIO_CHANNEL_SOCKET(ioc);
457 ssize_t ret;
458 struct msghdr msg = { NULL, };
459 char control[CMSG_SPACE(sizeof(int) * SOCKET_MAX_FDS)];
460 int sflags = 0;
462 memset(control, 0, CMSG_SPACE(sizeof(int) * SOCKET_MAX_FDS));
464 #ifdef MSG_CMSG_CLOEXEC
465 sflags |= MSG_CMSG_CLOEXEC;
466 #endif
468 msg.msg_iov = (struct iovec *)iov;
469 msg.msg_iovlen = niov;
470 if (fds && nfds) {
471 msg.msg_control = control;
472 msg.msg_controllen = sizeof(control);
475 retry:
476 ret = recvmsg(sioc->fd, &msg, sflags);
477 if (ret < 0) {
478 if (errno == EAGAIN) {
479 return QIO_CHANNEL_ERR_BLOCK;
481 if (errno == EINTR) {
482 goto retry;
485 error_setg_errno(errp, errno,
486 "Unable to read from socket");
487 return -1;
490 if (fds && nfds) {
491 qio_channel_socket_copy_fds(&msg, fds, nfds);
494 return ret;
497 static ssize_t qio_channel_socket_writev(QIOChannel *ioc,
498 const struct iovec *iov,
499 size_t niov,
500 int *fds,
501 size_t nfds,
502 Error **errp)
504 QIOChannelSocket *sioc = QIO_CHANNEL_SOCKET(ioc);
505 ssize_t ret;
506 struct msghdr msg = { NULL, };
507 char control[CMSG_SPACE(sizeof(int) * SOCKET_MAX_FDS)];
508 size_t fdsize = sizeof(int) * nfds;
509 struct cmsghdr *cmsg;
511 memset(control, 0, CMSG_SPACE(sizeof(int) * SOCKET_MAX_FDS));
513 msg.msg_iov = (struct iovec *)iov;
514 msg.msg_iovlen = niov;
516 if (nfds) {
517 if (nfds > SOCKET_MAX_FDS) {
518 error_setg_errno(errp, EINVAL,
519 "Only %d FDs can be sent, got %zu",
520 SOCKET_MAX_FDS, nfds);
521 return -1;
524 msg.msg_control = control;
525 msg.msg_controllen = CMSG_SPACE(sizeof(int) * nfds);
527 cmsg = CMSG_FIRSTHDR(&msg);
528 cmsg->cmsg_len = CMSG_LEN(fdsize);
529 cmsg->cmsg_level = SOL_SOCKET;
530 cmsg->cmsg_type = SCM_RIGHTS;
531 memcpy(CMSG_DATA(cmsg), fds, fdsize);
534 retry:
535 ret = sendmsg(sioc->fd, &msg, 0);
536 if (ret <= 0) {
537 if (errno == EAGAIN) {
538 return QIO_CHANNEL_ERR_BLOCK;
540 if (errno == EINTR) {
541 goto retry;
543 error_setg_errno(errp, errno,
544 "Unable to write to socket");
545 return -1;
547 return ret;
549 #else /* WIN32 */
550 static ssize_t qio_channel_socket_readv(QIOChannel *ioc,
551 const struct iovec *iov,
552 size_t niov,
553 int **fds,
554 size_t *nfds,
555 Error **errp)
557 QIOChannelSocket *sioc = QIO_CHANNEL_SOCKET(ioc);
558 ssize_t done = 0;
559 ssize_t i;
561 for (i = 0; i < niov; i++) {
562 ssize_t ret;
563 retry:
564 ret = recv(sioc->fd,
565 iov[i].iov_base,
566 iov[i].iov_len,
568 if (ret < 0) {
569 if (errno == EAGAIN) {
570 if (done) {
571 return done;
572 } else {
573 return QIO_CHANNEL_ERR_BLOCK;
575 } else if (errno == EINTR) {
576 goto retry;
577 } else {
578 error_setg_errno(errp, errno,
579 "Unable to read from socket");
580 return -1;
583 done += ret;
584 if (ret < iov[i].iov_len) {
585 return done;
589 return done;
592 static ssize_t qio_channel_socket_writev(QIOChannel *ioc,
593 const struct iovec *iov,
594 size_t niov,
595 int *fds,
596 size_t nfds,
597 Error **errp)
599 QIOChannelSocket *sioc = QIO_CHANNEL_SOCKET(ioc);
600 ssize_t done = 0;
601 ssize_t i;
603 for (i = 0; i < niov; i++) {
604 ssize_t ret;
605 retry:
606 ret = send(sioc->fd,
607 iov[i].iov_base,
608 iov[i].iov_len,
610 if (ret < 0) {
611 if (errno == EAGAIN) {
612 if (done) {
613 return done;
614 } else {
615 return QIO_CHANNEL_ERR_BLOCK;
617 } else if (errno == EINTR) {
618 goto retry;
619 } else {
620 error_setg_errno(errp, errno,
621 "Unable to write to socket");
622 return -1;
625 done += ret;
626 if (ret < iov[i].iov_len) {
627 return done;
631 return done;
633 #endif /* WIN32 */
635 static int
636 qio_channel_socket_set_blocking(QIOChannel *ioc,
637 bool enabled,
638 Error **errp)
640 QIOChannelSocket *sioc = QIO_CHANNEL_SOCKET(ioc);
642 if (enabled) {
643 qemu_set_block(sioc->fd);
644 } else {
645 qemu_set_nonblock(sioc->fd);
647 return 0;
651 static void
652 qio_channel_socket_set_delay(QIOChannel *ioc,
653 bool enabled)
655 QIOChannelSocket *sioc = QIO_CHANNEL_SOCKET(ioc);
656 int v = enabled ? 0 : 1;
658 qemu_setsockopt(sioc->fd,
659 IPPROTO_TCP, TCP_NODELAY,
660 &v, sizeof(v));
664 static void
665 qio_channel_socket_set_cork(QIOChannel *ioc,
666 bool enabled)
668 QIOChannelSocket *sioc = QIO_CHANNEL_SOCKET(ioc);
669 int v = enabled ? 1 : 0;
671 socket_set_cork(sioc->fd, v);
675 static int
676 qio_channel_socket_close(QIOChannel *ioc,
677 Error **errp)
679 QIOChannelSocket *sioc = QIO_CHANNEL_SOCKET(ioc);
681 if (sioc->fd != -1) {
682 #ifdef WIN32
683 WSAEventSelect(sioc->fd, NULL, 0);
684 #endif
685 if (closesocket(sioc->fd) < 0) {
686 sioc->fd = -1;
687 error_setg_errno(errp, errno,
688 "Unable to close socket");
689 return -1;
691 sioc->fd = -1;
693 return 0;
696 static int
697 qio_channel_socket_shutdown(QIOChannel *ioc,
698 QIOChannelShutdown how,
699 Error **errp)
701 QIOChannelSocket *sioc = QIO_CHANNEL_SOCKET(ioc);
702 int sockhow;
704 switch (how) {
705 case QIO_CHANNEL_SHUTDOWN_READ:
706 sockhow = SHUT_RD;
707 break;
708 case QIO_CHANNEL_SHUTDOWN_WRITE:
709 sockhow = SHUT_WR;
710 break;
711 case QIO_CHANNEL_SHUTDOWN_BOTH:
712 default:
713 sockhow = SHUT_RDWR;
714 break;
717 if (shutdown(sioc->fd, sockhow) < 0) {
718 error_setg_errno(errp, errno,
719 "Unable to shutdown socket");
720 return -1;
722 return 0;
725 static void qio_channel_socket_set_aio_fd_handler(QIOChannel *ioc,
726 AioContext *ctx,
727 IOHandler *io_read,
728 IOHandler *io_write,
729 void *opaque)
731 QIOChannelSocket *sioc = QIO_CHANNEL_SOCKET(ioc);
732 aio_set_fd_handler(ctx, sioc->fd, false, io_read, io_write, NULL, opaque);
735 static GSource *qio_channel_socket_create_watch(QIOChannel *ioc,
736 GIOCondition condition)
738 QIOChannelSocket *sioc = QIO_CHANNEL_SOCKET(ioc);
739 return qio_channel_create_socket_watch(ioc,
740 sioc->fd,
741 condition);
744 static void qio_channel_socket_class_init(ObjectClass *klass,
745 void *class_data G_GNUC_UNUSED)
747 QIOChannelClass *ioc_klass = QIO_CHANNEL_CLASS(klass);
749 ioc_klass->io_writev = qio_channel_socket_writev;
750 ioc_klass->io_readv = qio_channel_socket_readv;
751 ioc_klass->io_set_blocking = qio_channel_socket_set_blocking;
752 ioc_klass->io_close = qio_channel_socket_close;
753 ioc_klass->io_shutdown = qio_channel_socket_shutdown;
754 ioc_klass->io_set_cork = qio_channel_socket_set_cork;
755 ioc_klass->io_set_delay = qio_channel_socket_set_delay;
756 ioc_klass->io_create_watch = qio_channel_socket_create_watch;
757 ioc_klass->io_set_aio_fd_handler = qio_channel_socket_set_aio_fd_handler;
760 static const TypeInfo qio_channel_socket_info = {
761 .parent = TYPE_QIO_CHANNEL,
762 .name = TYPE_QIO_CHANNEL_SOCKET,
763 .instance_size = sizeof(QIOChannelSocket),
764 .instance_init = qio_channel_socket_init,
765 .instance_finalize = qio_channel_socket_finalize,
766 .class_init = qio_channel_socket_class_init,
769 static void qio_channel_socket_register_types(void)
771 type_register_static(&qio_channel_socket_info);
774 type_init(qio_channel_socket_register_types);