iotests.py: Allow concurrent qemu instances
[qemu/ar7.git] / io / channel-socket.c
blob196a4f18f7702cda40c8fd730de9b14e922444cf
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 ioc->features |= (1 << 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 int val;
76 socklen_t len = sizeof(val);
78 if (sioc->fd != -1) {
79 error_setg(errp, "Socket is already open");
80 return -1;
83 sioc->fd = fd;
84 sioc->remoteAddrLen = sizeof(sioc->remoteAddr);
85 sioc->localAddrLen = sizeof(sioc->localAddr);
88 if (getpeername(fd, (struct sockaddr *)&sioc->remoteAddr,
89 &sioc->remoteAddrLen) < 0) {
90 if (errno == ENOTCONN) {
91 memset(&sioc->remoteAddr, 0, sizeof(sioc->remoteAddr));
92 sioc->remoteAddrLen = sizeof(sioc->remoteAddr);
93 } else {
94 error_setg_errno(errp, errno,
95 "Unable to query remote socket address");
96 goto error;
100 if (getsockname(fd, (struct sockaddr *)&sioc->localAddr,
101 &sioc->localAddrLen) < 0) {
102 error_setg_errno(errp, errno,
103 "Unable to query local socket address");
104 goto error;
107 #ifndef WIN32
108 if (sioc->localAddr.ss_family == AF_UNIX) {
109 QIOChannel *ioc = QIO_CHANNEL(sioc);
110 ioc->features |= (1 << QIO_CHANNEL_FEATURE_FD_PASS);
112 #endif /* WIN32 */
113 if (getsockopt(fd, SOL_SOCKET, SO_ACCEPTCONN, &val, &len) == 0 && val) {
114 QIOChannel *ioc = QIO_CHANNEL(sioc);
115 ioc->features |= (1 << QIO_CHANNEL_FEATURE_LISTEN);
118 return 0;
120 error:
121 sioc->fd = -1; /* Let the caller close FD on failure */
122 return -1;
125 QIOChannelSocket *
126 qio_channel_socket_new_fd(int fd,
127 Error **errp)
129 QIOChannelSocket *ioc;
131 ioc = qio_channel_socket_new();
132 if (qio_channel_socket_set_fd(ioc, fd, errp) < 0) {
133 object_unref(OBJECT(ioc));
134 return NULL;
137 trace_qio_channel_socket_new_fd(ioc, fd);
139 return ioc;
143 int qio_channel_socket_connect_sync(QIOChannelSocket *ioc,
144 SocketAddress *addr,
145 Error **errp)
147 int fd;
149 trace_qio_channel_socket_connect_sync(ioc, addr);
150 fd = socket_connect(addr, errp, NULL, NULL);
151 if (fd < 0) {
152 trace_qio_channel_socket_connect_fail(ioc);
153 return -1;
156 trace_qio_channel_socket_connect_complete(ioc, fd);
157 if (qio_channel_socket_set_fd(ioc, fd, errp) < 0) {
158 close(fd);
159 return -1;
162 return 0;
166 static int qio_channel_socket_connect_worker(QIOTask *task,
167 Error **errp,
168 gpointer opaque)
170 QIOChannelSocket *ioc = QIO_CHANNEL_SOCKET(qio_task_get_source(task));
171 SocketAddress *addr = opaque;
172 int ret;
174 ret = qio_channel_socket_connect_sync(ioc,
175 addr,
176 errp);
178 object_unref(OBJECT(ioc));
179 return ret;
183 void qio_channel_socket_connect_async(QIOChannelSocket *ioc,
184 SocketAddress *addr,
185 QIOTaskFunc callback,
186 gpointer opaque,
187 GDestroyNotify destroy)
189 QIOTask *task = qio_task_new(
190 OBJECT(ioc), callback, opaque, destroy);
191 SocketAddress *addrCopy;
193 addrCopy = QAPI_CLONE(SocketAddress, addr);
195 /* socket_connect() does a non-blocking connect(), but it
196 * still blocks in DNS lookups, so we must use a thread */
197 trace_qio_channel_socket_connect_async(ioc, addr);
198 qio_task_run_in_thread(task,
199 qio_channel_socket_connect_worker,
200 addrCopy,
201 (GDestroyNotify)qapi_free_SocketAddress);
205 int qio_channel_socket_listen_sync(QIOChannelSocket *ioc,
206 SocketAddress *addr,
207 Error **errp)
209 int fd;
211 trace_qio_channel_socket_listen_sync(ioc, addr);
212 fd = socket_listen(addr, errp);
213 if (fd < 0) {
214 trace_qio_channel_socket_listen_fail(ioc);
215 return -1;
218 trace_qio_channel_socket_listen_complete(ioc, fd);
219 if (qio_channel_socket_set_fd(ioc, fd, errp) < 0) {
220 close(fd);
221 return -1;
224 return 0;
228 static int qio_channel_socket_listen_worker(QIOTask *task,
229 Error **errp,
230 gpointer opaque)
232 QIOChannelSocket *ioc = QIO_CHANNEL_SOCKET(qio_task_get_source(task));
233 SocketAddress *addr = opaque;
234 int ret;
236 ret = qio_channel_socket_listen_sync(ioc,
237 addr,
238 errp);
240 object_unref(OBJECT(ioc));
241 return ret;
245 void qio_channel_socket_listen_async(QIOChannelSocket *ioc,
246 SocketAddress *addr,
247 QIOTaskFunc callback,
248 gpointer opaque,
249 GDestroyNotify destroy)
251 QIOTask *task = qio_task_new(
252 OBJECT(ioc), callback, opaque, destroy);
253 SocketAddress *addrCopy;
255 addrCopy = QAPI_CLONE(SocketAddress, addr);
257 /* socket_listen() blocks in DNS lookups, so we must use a thread */
258 trace_qio_channel_socket_listen_async(ioc, addr);
259 qio_task_run_in_thread(task,
260 qio_channel_socket_listen_worker,
261 addrCopy,
262 (GDestroyNotify)qapi_free_SocketAddress);
266 int qio_channel_socket_dgram_sync(QIOChannelSocket *ioc,
267 SocketAddress *localAddr,
268 SocketAddress *remoteAddr,
269 Error **errp)
271 int fd;
273 trace_qio_channel_socket_dgram_sync(ioc, localAddr, remoteAddr);
274 fd = socket_dgram(remoteAddr, localAddr, errp);
275 if (fd < 0) {
276 trace_qio_channel_socket_dgram_fail(ioc);
277 return -1;
280 trace_qio_channel_socket_dgram_complete(ioc, fd);
281 if (qio_channel_socket_set_fd(ioc, fd, errp) < 0) {
282 close(fd);
283 return -1;
286 return 0;
290 struct QIOChannelSocketDGramWorkerData {
291 SocketAddress *localAddr;
292 SocketAddress *remoteAddr;
296 static void qio_channel_socket_dgram_worker_free(gpointer opaque)
298 struct QIOChannelSocketDGramWorkerData *data = opaque;
299 qapi_free_SocketAddress(data->localAddr);
300 qapi_free_SocketAddress(data->remoteAddr);
301 g_free(data);
304 static int qio_channel_socket_dgram_worker(QIOTask *task,
305 Error **errp,
306 gpointer opaque)
308 QIOChannelSocket *ioc = QIO_CHANNEL_SOCKET(qio_task_get_source(task));
309 struct QIOChannelSocketDGramWorkerData *data = opaque;
310 int ret;
312 /* socket_dgram() blocks in DNS lookups, so we must use a thread */
313 ret = qio_channel_socket_dgram_sync(ioc,
314 data->localAddr,
315 data->remoteAddr,
316 errp);
318 object_unref(OBJECT(ioc));
319 return ret;
323 void qio_channel_socket_dgram_async(QIOChannelSocket *ioc,
324 SocketAddress *localAddr,
325 SocketAddress *remoteAddr,
326 QIOTaskFunc callback,
327 gpointer opaque,
328 GDestroyNotify destroy)
330 QIOTask *task = qio_task_new(
331 OBJECT(ioc), callback, opaque, destroy);
332 struct QIOChannelSocketDGramWorkerData *data = g_new0(
333 struct QIOChannelSocketDGramWorkerData, 1);
335 data->localAddr = QAPI_CLONE(SocketAddress, localAddr);
336 data->remoteAddr = QAPI_CLONE(SocketAddress, remoteAddr);
338 trace_qio_channel_socket_dgram_async(ioc, localAddr, remoteAddr);
339 qio_task_run_in_thread(task,
340 qio_channel_socket_dgram_worker,
341 data,
342 qio_channel_socket_dgram_worker_free);
346 QIOChannelSocket *
347 qio_channel_socket_accept(QIOChannelSocket *ioc,
348 Error **errp)
350 QIOChannelSocket *cioc;
352 cioc = QIO_CHANNEL_SOCKET(object_new(TYPE_QIO_CHANNEL_SOCKET));
353 cioc->fd = -1;
354 cioc->remoteAddrLen = sizeof(ioc->remoteAddr);
355 cioc->localAddrLen = sizeof(ioc->localAddr);
357 #ifdef WIN32
358 QIO_CHANNEL(cioc)->event = CreateEvent(NULL, FALSE, FALSE, NULL);
359 #endif
362 retry:
363 trace_qio_channel_socket_accept(ioc);
364 cioc->fd = qemu_accept(ioc->fd, (struct sockaddr *)&cioc->remoteAddr,
365 &cioc->remoteAddrLen);
366 if (cioc->fd < 0) {
367 trace_qio_channel_socket_accept_fail(ioc);
368 if (errno == EINTR) {
369 goto retry;
371 goto error;
374 if (getsockname(cioc->fd, (struct sockaddr *)&cioc->localAddr,
375 &cioc->localAddrLen) < 0) {
376 error_setg_errno(errp, errno,
377 "Unable to query local socket address");
378 goto error;
381 #ifndef WIN32
382 if (cioc->localAddr.ss_family == AF_UNIX) {
383 QIO_CHANNEL(cioc)->features |= (1 << QIO_CHANNEL_FEATURE_FD_PASS);
385 #endif /* WIN32 */
387 trace_qio_channel_socket_accept_complete(ioc, cioc, cioc->fd);
388 return cioc;
390 error:
391 object_unref(OBJECT(cioc));
392 return NULL;
395 static void qio_channel_socket_init(Object *obj)
397 QIOChannelSocket *ioc = QIO_CHANNEL_SOCKET(obj);
398 ioc->fd = -1;
401 static void qio_channel_socket_finalize(Object *obj)
403 QIOChannelSocket *ioc = QIO_CHANNEL_SOCKET(obj);
405 if (ioc->fd != -1) {
406 if (QIO_CHANNEL(ioc)->features & QIO_CHANNEL_FEATURE_LISTEN) {
407 Error *err = NULL;
409 socket_listen_cleanup(ioc->fd, &err);
410 if (err) {
411 error_report_err(err);
412 err = NULL;
415 #ifdef WIN32
416 WSAEventSelect(ioc->fd, NULL, 0);
417 #endif
418 closesocket(ioc->fd);
419 ioc->fd = -1;
424 #ifndef WIN32
425 static void qio_channel_socket_copy_fds(struct msghdr *msg,
426 int **fds, size_t *nfds)
428 struct cmsghdr *cmsg;
430 *nfds = 0;
431 *fds = NULL;
433 for (cmsg = CMSG_FIRSTHDR(msg); cmsg; cmsg = CMSG_NXTHDR(msg, cmsg)) {
434 int fd_size, i;
435 int gotfds;
437 if (cmsg->cmsg_len < CMSG_LEN(sizeof(int)) ||
438 cmsg->cmsg_level != SOL_SOCKET ||
439 cmsg->cmsg_type != SCM_RIGHTS) {
440 continue;
443 fd_size = cmsg->cmsg_len - CMSG_LEN(0);
445 if (!fd_size) {
446 continue;
449 gotfds = fd_size / sizeof(int);
450 *fds = g_renew(int, *fds, *nfds + gotfds);
451 memcpy(*fds + *nfds, CMSG_DATA(cmsg), fd_size);
453 for (i = 0; i < gotfds; i++) {
454 int fd = (*fds)[*nfds + i];
455 if (fd < 0) {
456 continue;
459 /* O_NONBLOCK is preserved across SCM_RIGHTS so reset it */
460 qemu_set_block(fd);
462 #ifndef MSG_CMSG_CLOEXEC
463 qemu_set_cloexec(fd);
464 #endif
466 *nfds += gotfds;
471 static ssize_t qio_channel_socket_readv(QIOChannel *ioc,
472 const struct iovec *iov,
473 size_t niov,
474 int **fds,
475 size_t *nfds,
476 Error **errp)
478 QIOChannelSocket *sioc = QIO_CHANNEL_SOCKET(ioc);
479 ssize_t ret;
480 struct msghdr msg = { NULL, };
481 char control[CMSG_SPACE(sizeof(int) * SOCKET_MAX_FDS)];
482 int sflags = 0;
484 memset(control, 0, CMSG_SPACE(sizeof(int) * SOCKET_MAX_FDS));
486 #ifdef MSG_CMSG_CLOEXEC
487 sflags |= MSG_CMSG_CLOEXEC;
488 #endif
490 msg.msg_iov = (struct iovec *)iov;
491 msg.msg_iovlen = niov;
492 if (fds && nfds) {
493 msg.msg_control = control;
494 msg.msg_controllen = sizeof(control);
497 retry:
498 ret = recvmsg(sioc->fd, &msg, sflags);
499 if (ret < 0) {
500 if (errno == EAGAIN) {
501 return QIO_CHANNEL_ERR_BLOCK;
503 if (errno == EINTR) {
504 goto retry;
507 error_setg_errno(errp, errno,
508 "Unable to read from socket");
509 return -1;
512 if (fds && nfds) {
513 qio_channel_socket_copy_fds(&msg, fds, nfds);
516 return ret;
519 static ssize_t qio_channel_socket_writev(QIOChannel *ioc,
520 const struct iovec *iov,
521 size_t niov,
522 int *fds,
523 size_t nfds,
524 Error **errp)
526 QIOChannelSocket *sioc = QIO_CHANNEL_SOCKET(ioc);
527 ssize_t ret;
528 struct msghdr msg = { NULL, };
529 char control[CMSG_SPACE(sizeof(int) * SOCKET_MAX_FDS)];
530 size_t fdsize = sizeof(int) * nfds;
531 struct cmsghdr *cmsg;
533 memset(control, 0, CMSG_SPACE(sizeof(int) * SOCKET_MAX_FDS));
535 msg.msg_iov = (struct iovec *)iov;
536 msg.msg_iovlen = niov;
538 if (nfds) {
539 if (nfds > SOCKET_MAX_FDS) {
540 error_setg_errno(errp, EINVAL,
541 "Only %d FDs can be sent, got %zu",
542 SOCKET_MAX_FDS, nfds);
543 return -1;
546 msg.msg_control = control;
547 msg.msg_controllen = CMSG_SPACE(sizeof(int) * nfds);
549 cmsg = CMSG_FIRSTHDR(&msg);
550 cmsg->cmsg_len = CMSG_LEN(fdsize);
551 cmsg->cmsg_level = SOL_SOCKET;
552 cmsg->cmsg_type = SCM_RIGHTS;
553 memcpy(CMSG_DATA(cmsg), fds, fdsize);
556 retry:
557 ret = sendmsg(sioc->fd, &msg, 0);
558 if (ret <= 0) {
559 if (errno == EAGAIN) {
560 return QIO_CHANNEL_ERR_BLOCK;
562 if (errno == EINTR) {
563 goto retry;
565 error_setg_errno(errp, errno,
566 "Unable to write to socket");
567 return -1;
569 return ret;
571 #else /* WIN32 */
572 static ssize_t qio_channel_socket_readv(QIOChannel *ioc,
573 const struct iovec *iov,
574 size_t niov,
575 int **fds,
576 size_t *nfds,
577 Error **errp)
579 QIOChannelSocket *sioc = QIO_CHANNEL_SOCKET(ioc);
580 ssize_t done = 0;
581 ssize_t i;
583 for (i = 0; i < niov; i++) {
584 ssize_t ret;
585 retry:
586 ret = recv(sioc->fd,
587 iov[i].iov_base,
588 iov[i].iov_len,
590 if (ret < 0) {
591 if (errno == EAGAIN) {
592 if (done) {
593 return done;
594 } else {
595 return QIO_CHANNEL_ERR_BLOCK;
597 } else if (errno == EINTR) {
598 goto retry;
599 } else {
600 error_setg_errno(errp, errno,
601 "Unable to read from socket");
602 return -1;
605 done += ret;
606 if (ret < iov[i].iov_len) {
607 return done;
611 return done;
614 static ssize_t qio_channel_socket_writev(QIOChannel *ioc,
615 const struct iovec *iov,
616 size_t niov,
617 int *fds,
618 size_t nfds,
619 Error **errp)
621 QIOChannelSocket *sioc = QIO_CHANNEL_SOCKET(ioc);
622 ssize_t done = 0;
623 ssize_t i;
625 for (i = 0; i < niov; i++) {
626 ssize_t ret;
627 retry:
628 ret = send(sioc->fd,
629 iov[i].iov_base,
630 iov[i].iov_len,
632 if (ret < 0) {
633 if (errno == EAGAIN) {
634 if (done) {
635 return done;
636 } else {
637 return QIO_CHANNEL_ERR_BLOCK;
639 } else if (errno == EINTR) {
640 goto retry;
641 } else {
642 error_setg_errno(errp, errno,
643 "Unable to write to socket");
644 return -1;
647 done += ret;
648 if (ret < iov[i].iov_len) {
649 return done;
653 return done;
655 #endif /* WIN32 */
657 static int
658 qio_channel_socket_set_blocking(QIOChannel *ioc,
659 bool enabled,
660 Error **errp)
662 QIOChannelSocket *sioc = QIO_CHANNEL_SOCKET(ioc);
664 if (enabled) {
665 qemu_set_block(sioc->fd);
666 } else {
667 qemu_set_nonblock(sioc->fd);
668 #ifdef WIN32
669 WSAEventSelect(sioc->fd, ioc->event,
670 FD_READ | FD_ACCEPT | FD_CLOSE |
671 FD_CONNECT | FD_WRITE | FD_OOB);
672 #endif
674 return 0;
678 static void
679 qio_channel_socket_set_delay(QIOChannel *ioc,
680 bool enabled)
682 QIOChannelSocket *sioc = QIO_CHANNEL_SOCKET(ioc);
683 int v = enabled ? 0 : 1;
685 qemu_setsockopt(sioc->fd,
686 IPPROTO_TCP, TCP_NODELAY,
687 &v, sizeof(v));
691 static void
692 qio_channel_socket_set_cork(QIOChannel *ioc,
693 bool enabled)
695 QIOChannelSocket *sioc = QIO_CHANNEL_SOCKET(ioc);
696 int v = enabled ? 1 : 0;
698 socket_set_cork(sioc->fd, v);
702 static int
703 qio_channel_socket_close(QIOChannel *ioc,
704 Error **errp)
706 QIOChannelSocket *sioc = QIO_CHANNEL_SOCKET(ioc);
708 if (sioc->fd != -1) {
709 #ifdef WIN32
710 WSAEventSelect(sioc->fd, NULL, 0);
711 #endif
712 if (closesocket(sioc->fd) < 0) {
713 sioc->fd = -1;
714 error_setg_errno(errp, errno,
715 "Unable to close socket");
716 return -1;
718 sioc->fd = -1;
720 return 0;
723 static int
724 qio_channel_socket_shutdown(QIOChannel *ioc,
725 QIOChannelShutdown how,
726 Error **errp)
728 QIOChannelSocket *sioc = QIO_CHANNEL_SOCKET(ioc);
729 int sockhow;
731 switch (how) {
732 case QIO_CHANNEL_SHUTDOWN_READ:
733 sockhow = SHUT_RD;
734 break;
735 case QIO_CHANNEL_SHUTDOWN_WRITE:
736 sockhow = SHUT_WR;
737 break;
738 case QIO_CHANNEL_SHUTDOWN_BOTH:
739 default:
740 sockhow = SHUT_RDWR;
741 break;
744 if (shutdown(sioc->fd, sockhow) < 0) {
745 error_setg_errno(errp, errno,
746 "Unable to shutdown socket");
747 return -1;
749 return 0;
752 static GSource *qio_channel_socket_create_watch(QIOChannel *ioc,
753 GIOCondition condition)
755 QIOChannelSocket *sioc = QIO_CHANNEL_SOCKET(ioc);
756 return qio_channel_create_socket_watch(ioc,
757 sioc->fd,
758 condition);
761 static void qio_channel_socket_class_init(ObjectClass *klass,
762 void *class_data G_GNUC_UNUSED)
764 QIOChannelClass *ioc_klass = QIO_CHANNEL_CLASS(klass);
766 ioc_klass->io_writev = qio_channel_socket_writev;
767 ioc_klass->io_readv = qio_channel_socket_readv;
768 ioc_klass->io_set_blocking = qio_channel_socket_set_blocking;
769 ioc_klass->io_close = qio_channel_socket_close;
770 ioc_klass->io_shutdown = qio_channel_socket_shutdown;
771 ioc_klass->io_set_cork = qio_channel_socket_set_cork;
772 ioc_klass->io_set_delay = qio_channel_socket_set_delay;
773 ioc_klass->io_create_watch = qio_channel_socket_create_watch;
776 static const TypeInfo qio_channel_socket_info = {
777 .parent = TYPE_QIO_CHANNEL,
778 .name = TYPE_QIO_CHANNEL_SOCKET,
779 .instance_size = sizeof(QIOChannelSocket),
780 .instance_init = qio_channel_socket_init,
781 .instance_finalize = qio_channel_socket_finalize,
782 .class_init = qio_channel_socket_class_init,
785 static void qio_channel_socket_register_types(void)
787 type_register_static(&qio_channel_socket_info);
790 type_init(qio_channel_socket_register_types);