migration/rdma: Plug memory leaks in qemu_rdma_registration_stop()
[qemu/armbru.git] / io / channel-socket.c
blobe1b46670874f86391e644242a2e68530fb64cf1f
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/>.
20 #include "qemu/osdep.h"
21 #include "qemu-common.h"
22 #include "qapi/error.h"
23 #include "qapi/qapi-visit-sockets.h"
24 #include "qemu/module.h"
25 #include "io/channel-socket.h"
26 #include "io/channel-watch.h"
27 #include "trace.h"
28 #include "qapi/clone-visitor.h"
30 #define SOCKET_MAX_FDS 16
32 SocketAddress *
33 qio_channel_socket_get_local_address(QIOChannelSocket *ioc,
34 Error **errp)
36 return socket_sockaddr_to_address(&ioc->localAddr,
37 ioc->localAddrLen,
38 errp);
41 SocketAddress *
42 qio_channel_socket_get_remote_address(QIOChannelSocket *ioc,
43 Error **errp)
45 return socket_sockaddr_to_address(&ioc->remoteAddr,
46 ioc->remoteAddrLen,
47 errp);
50 QIOChannelSocket *
51 qio_channel_socket_new(void)
53 QIOChannelSocket *sioc;
54 QIOChannel *ioc;
56 sioc = QIO_CHANNEL_SOCKET(object_new(TYPE_QIO_CHANNEL_SOCKET));
57 sioc->fd = -1;
59 ioc = QIO_CHANNEL(sioc);
60 qio_channel_set_feature(ioc, QIO_CHANNEL_FEATURE_SHUTDOWN);
62 #ifdef WIN32
63 ioc->event = CreateEvent(NULL, FALSE, FALSE, NULL);
64 #endif
66 trace_qio_channel_socket_new(sioc);
68 return sioc;
72 static int
73 qio_channel_socket_set_fd(QIOChannelSocket *sioc,
74 int fd,
75 Error **errp)
77 if (sioc->fd != -1) {
78 error_setg(errp, "Socket is already open");
79 return -1;
82 sioc->fd = fd;
83 sioc->remoteAddrLen = sizeof(sioc->remoteAddr);
84 sioc->localAddrLen = sizeof(sioc->localAddr);
87 if (getpeername(fd, (struct sockaddr *)&sioc->remoteAddr,
88 &sioc->remoteAddrLen) < 0) {
89 if (errno == ENOTCONN) {
90 memset(&sioc->remoteAddr, 0, sizeof(sioc->remoteAddr));
91 sioc->remoteAddrLen = sizeof(sioc->remoteAddr);
92 } else {
93 error_setg_errno(errp, errno,
94 "Unable to query remote socket address");
95 goto error;
99 if (getsockname(fd, (struct sockaddr *)&sioc->localAddr,
100 &sioc->localAddrLen) < 0) {
101 error_setg_errno(errp, errno,
102 "Unable to query local socket address");
103 goto error;
106 #ifndef WIN32
107 if (sioc->localAddr.ss_family == AF_UNIX) {
108 QIOChannel *ioc = QIO_CHANNEL(sioc);
109 qio_channel_set_feature(ioc, QIO_CHANNEL_FEATURE_FD_PASS);
111 #endif /* WIN32 */
113 return 0;
115 error:
116 sioc->fd = -1; /* Let the caller close FD on failure */
117 return -1;
120 QIOChannelSocket *
121 qio_channel_socket_new_fd(int fd,
122 Error **errp)
124 QIOChannelSocket *ioc;
126 ioc = qio_channel_socket_new();
127 if (qio_channel_socket_set_fd(ioc, fd, errp) < 0) {
128 object_unref(OBJECT(ioc));
129 return NULL;
132 trace_qio_channel_socket_new_fd(ioc, fd);
134 return ioc;
138 int qio_channel_socket_connect_sync(QIOChannelSocket *ioc,
139 SocketAddress *addr,
140 Error **errp)
142 int fd;
144 trace_qio_channel_socket_connect_sync(ioc, addr);
145 fd = socket_connect(addr, errp);
146 if (fd < 0) {
147 trace_qio_channel_socket_connect_fail(ioc);
148 return -1;
151 trace_qio_channel_socket_connect_complete(ioc, fd);
152 if (qio_channel_socket_set_fd(ioc, fd, errp) < 0) {
153 close(fd);
154 return -1;
157 return 0;
161 static void qio_channel_socket_connect_worker(QIOTask *task,
162 gpointer opaque)
164 QIOChannelSocket *ioc = QIO_CHANNEL_SOCKET(qio_task_get_source(task));
165 SocketAddress *addr = opaque;
166 Error *err = NULL;
168 qio_channel_socket_connect_sync(ioc, addr, &err);
170 qio_task_set_error(task, err);
174 void qio_channel_socket_connect_async(QIOChannelSocket *ioc,
175 SocketAddress *addr,
176 QIOTaskFunc callback,
177 gpointer opaque,
178 GDestroyNotify destroy,
179 GMainContext *context)
181 QIOTask *task = qio_task_new(
182 OBJECT(ioc), callback, opaque, destroy);
183 SocketAddress *addrCopy;
185 addrCopy = QAPI_CLONE(SocketAddress, addr);
187 /* socket_connect() does a non-blocking connect(), but it
188 * still blocks in DNS lookups, so we must use a thread */
189 trace_qio_channel_socket_connect_async(ioc, addr);
190 qio_task_run_in_thread(task,
191 qio_channel_socket_connect_worker,
192 addrCopy,
193 (GDestroyNotify)qapi_free_SocketAddress,
194 context);
198 int qio_channel_socket_listen_sync(QIOChannelSocket *ioc,
199 SocketAddress *addr,
200 int num,
201 Error **errp)
203 int fd;
205 trace_qio_channel_socket_listen_sync(ioc, addr, num);
206 fd = socket_listen(addr, num, errp);
207 if (fd < 0) {
208 trace_qio_channel_socket_listen_fail(ioc);
209 return -1;
212 trace_qio_channel_socket_listen_complete(ioc, fd);
213 if (qio_channel_socket_set_fd(ioc, fd, errp) < 0) {
214 close(fd);
215 return -1;
217 qio_channel_set_feature(QIO_CHANNEL(ioc), QIO_CHANNEL_FEATURE_LISTEN);
219 return 0;
223 struct QIOChannelListenWorkerData {
224 SocketAddress *addr;
225 int num; /* amount of expected connections */
228 static void qio_channel_listen_worker_free(gpointer opaque)
230 struct QIOChannelListenWorkerData *data = opaque;
232 qapi_free_SocketAddress(data->addr);
233 g_free(data);
236 static void qio_channel_socket_listen_worker(QIOTask *task,
237 gpointer opaque)
239 QIOChannelSocket *ioc = QIO_CHANNEL_SOCKET(qio_task_get_source(task));
240 struct QIOChannelListenWorkerData *data = opaque;
241 Error *err = NULL;
243 qio_channel_socket_listen_sync(ioc, data->addr, data->num, &err);
245 qio_task_set_error(task, err);
249 void qio_channel_socket_listen_async(QIOChannelSocket *ioc,
250 SocketAddress *addr,
251 int num,
252 QIOTaskFunc callback,
253 gpointer opaque,
254 GDestroyNotify destroy,
255 GMainContext *context)
257 QIOTask *task = qio_task_new(
258 OBJECT(ioc), callback, opaque, destroy);
259 struct QIOChannelListenWorkerData *data;
261 data = g_new0(struct QIOChannelListenWorkerData, 1);
262 data->addr = QAPI_CLONE(SocketAddress, addr);
263 data->num = num;
265 /* socket_listen() blocks in DNS lookups, so we must use a thread */
266 trace_qio_channel_socket_listen_async(ioc, addr, num);
267 qio_task_run_in_thread(task,
268 qio_channel_socket_listen_worker,
269 data,
270 qio_channel_listen_worker_free,
271 context);
275 int qio_channel_socket_dgram_sync(QIOChannelSocket *ioc,
276 SocketAddress *localAddr,
277 SocketAddress *remoteAddr,
278 Error **errp)
280 int fd;
282 trace_qio_channel_socket_dgram_sync(ioc, localAddr, remoteAddr);
283 fd = socket_dgram(remoteAddr, localAddr, errp);
284 if (fd < 0) {
285 trace_qio_channel_socket_dgram_fail(ioc);
286 return -1;
289 trace_qio_channel_socket_dgram_complete(ioc, fd);
290 if (qio_channel_socket_set_fd(ioc, fd, errp) < 0) {
291 close(fd);
292 return -1;
295 return 0;
299 struct QIOChannelSocketDGramWorkerData {
300 SocketAddress *localAddr;
301 SocketAddress *remoteAddr;
305 static void qio_channel_socket_dgram_worker_free(gpointer opaque)
307 struct QIOChannelSocketDGramWorkerData *data = opaque;
308 qapi_free_SocketAddress(data->localAddr);
309 qapi_free_SocketAddress(data->remoteAddr);
310 g_free(data);
313 static void qio_channel_socket_dgram_worker(QIOTask *task,
314 gpointer opaque)
316 QIOChannelSocket *ioc = QIO_CHANNEL_SOCKET(qio_task_get_source(task));
317 struct QIOChannelSocketDGramWorkerData *data = opaque;
318 Error *err = NULL;
320 /* socket_dgram() blocks in DNS lookups, so we must use a thread */
321 qio_channel_socket_dgram_sync(ioc, data->localAddr,
322 data->remoteAddr, &err);
324 qio_task_set_error(task, err);
328 void qio_channel_socket_dgram_async(QIOChannelSocket *ioc,
329 SocketAddress *localAddr,
330 SocketAddress *remoteAddr,
331 QIOTaskFunc callback,
332 gpointer opaque,
333 GDestroyNotify destroy,
334 GMainContext *context)
336 QIOTask *task = qio_task_new(
337 OBJECT(ioc), callback, opaque, destroy);
338 struct QIOChannelSocketDGramWorkerData *data = g_new0(
339 struct QIOChannelSocketDGramWorkerData, 1);
341 data->localAddr = QAPI_CLONE(SocketAddress, localAddr);
342 data->remoteAddr = QAPI_CLONE(SocketAddress, remoteAddr);
344 trace_qio_channel_socket_dgram_async(ioc, localAddr, remoteAddr);
345 qio_task_run_in_thread(task,
346 qio_channel_socket_dgram_worker,
347 data,
348 qio_channel_socket_dgram_worker_free,
349 context);
353 QIOChannelSocket *
354 qio_channel_socket_accept(QIOChannelSocket *ioc,
355 Error **errp)
357 QIOChannelSocket *cioc;
359 cioc = qio_channel_socket_new();
360 cioc->remoteAddrLen = sizeof(ioc->remoteAddr);
361 cioc->localAddrLen = sizeof(ioc->localAddr);
363 retry:
364 trace_qio_channel_socket_accept(ioc);
365 cioc->fd = qemu_accept(ioc->fd, (struct sockaddr *)&cioc->remoteAddr,
366 &cioc->remoteAddrLen);
367 if (cioc->fd < 0) {
368 if (errno == EINTR) {
369 goto retry;
371 error_setg_errno(errp, errno, "Unable to accept connection");
372 trace_qio_channel_socket_accept_fail(ioc);
373 goto error;
376 if (getsockname(cioc->fd, (struct sockaddr *)&cioc->localAddr,
377 &cioc->localAddrLen) < 0) {
378 error_setg_errno(errp, errno,
379 "Unable to query local socket address");
380 goto error;
383 #ifndef WIN32
384 if (cioc->localAddr.ss_family == AF_UNIX) {
385 QIOChannel *ioc_local = QIO_CHANNEL(cioc);
386 qio_channel_set_feature(ioc_local, QIO_CHANNEL_FEATURE_FD_PASS);
388 #endif /* WIN32 */
390 trace_qio_channel_socket_accept_complete(ioc, cioc, cioc->fd);
391 return cioc;
393 error:
394 object_unref(OBJECT(cioc));
395 return NULL;
398 static void qio_channel_socket_init(Object *obj)
400 QIOChannelSocket *ioc = QIO_CHANNEL_SOCKET(obj);
401 ioc->fd = -1;
404 static void qio_channel_socket_finalize(Object *obj)
406 QIOChannelSocket *ioc = QIO_CHANNEL_SOCKET(obj);
408 if (ioc->fd != -1) {
409 QIOChannel *ioc_local = QIO_CHANNEL(ioc);
410 if (qio_channel_has_feature(ioc_local, QIO_CHANNEL_FEATURE_LISTEN)) {
411 Error *err = NULL;
413 socket_listen_cleanup(ioc->fd, &err);
414 if (err) {
415 error_report_err(err);
416 err = NULL;
419 #ifdef WIN32
420 WSAEventSelect(ioc->fd, NULL, 0);
421 #endif
422 closesocket(ioc->fd);
423 ioc->fd = -1;
428 #ifndef WIN32
429 static void qio_channel_socket_copy_fds(struct msghdr *msg,
430 int **fds, size_t *nfds)
432 struct cmsghdr *cmsg;
434 *nfds = 0;
435 *fds = NULL;
437 for (cmsg = CMSG_FIRSTHDR(msg); cmsg; cmsg = CMSG_NXTHDR(msg, cmsg)) {
438 int fd_size, i;
439 int gotfds;
441 if (cmsg->cmsg_len < CMSG_LEN(sizeof(int)) ||
442 cmsg->cmsg_level != SOL_SOCKET ||
443 cmsg->cmsg_type != SCM_RIGHTS) {
444 continue;
447 fd_size = cmsg->cmsg_len - CMSG_LEN(0);
449 if (!fd_size) {
450 continue;
453 gotfds = fd_size / sizeof(int);
454 *fds = g_renew(int, *fds, *nfds + gotfds);
455 memcpy(*fds + *nfds, CMSG_DATA(cmsg), fd_size);
457 for (i = 0; i < gotfds; i++) {
458 int fd = (*fds)[*nfds + i];
459 if (fd < 0) {
460 continue;
463 /* O_NONBLOCK is preserved across SCM_RIGHTS so reset it */
464 qemu_set_block(fd);
466 #ifndef MSG_CMSG_CLOEXEC
467 qemu_set_cloexec(fd);
468 #endif
470 *nfds += gotfds;
475 static ssize_t qio_channel_socket_readv(QIOChannel *ioc,
476 const struct iovec *iov,
477 size_t niov,
478 int **fds,
479 size_t *nfds,
480 Error **errp)
482 QIOChannelSocket *sioc = QIO_CHANNEL_SOCKET(ioc);
483 ssize_t ret;
484 struct msghdr msg = { NULL, };
485 char control[CMSG_SPACE(sizeof(int) * SOCKET_MAX_FDS)];
486 int sflags = 0;
488 memset(control, 0, CMSG_SPACE(sizeof(int) * SOCKET_MAX_FDS));
490 #ifdef MSG_CMSG_CLOEXEC
491 sflags |= MSG_CMSG_CLOEXEC;
492 #endif
494 msg.msg_iov = (struct iovec *)iov;
495 msg.msg_iovlen = niov;
496 if (fds && nfds) {
497 msg.msg_control = control;
498 msg.msg_controllen = sizeof(control);
501 retry:
502 ret = recvmsg(sioc->fd, &msg, sflags);
503 if (ret < 0) {
504 if (errno == EAGAIN) {
505 return QIO_CHANNEL_ERR_BLOCK;
507 if (errno == EINTR) {
508 goto retry;
511 error_setg_errno(errp, errno,
512 "Unable to read from socket");
513 return -1;
516 if (fds && nfds) {
517 qio_channel_socket_copy_fds(&msg, fds, nfds);
520 return ret;
523 static ssize_t qio_channel_socket_writev(QIOChannel *ioc,
524 const struct iovec *iov,
525 size_t niov,
526 int *fds,
527 size_t nfds,
528 Error **errp)
530 QIOChannelSocket *sioc = QIO_CHANNEL_SOCKET(ioc);
531 ssize_t ret;
532 struct msghdr msg = { NULL, };
533 char control[CMSG_SPACE(sizeof(int) * SOCKET_MAX_FDS)];
534 size_t fdsize = sizeof(int) * nfds;
535 struct cmsghdr *cmsg;
537 memset(control, 0, CMSG_SPACE(sizeof(int) * SOCKET_MAX_FDS));
539 msg.msg_iov = (struct iovec *)iov;
540 msg.msg_iovlen = niov;
542 if (nfds) {
543 if (nfds > SOCKET_MAX_FDS) {
544 error_setg_errno(errp, EINVAL,
545 "Only %d FDs can be sent, got %zu",
546 SOCKET_MAX_FDS, nfds);
547 return -1;
550 msg.msg_control = control;
551 msg.msg_controllen = CMSG_SPACE(sizeof(int) * nfds);
553 cmsg = CMSG_FIRSTHDR(&msg);
554 cmsg->cmsg_len = CMSG_LEN(fdsize);
555 cmsg->cmsg_level = SOL_SOCKET;
556 cmsg->cmsg_type = SCM_RIGHTS;
557 memcpy(CMSG_DATA(cmsg), fds, fdsize);
560 retry:
561 ret = sendmsg(sioc->fd, &msg, 0);
562 if (ret <= 0) {
563 if (errno == EAGAIN) {
564 return QIO_CHANNEL_ERR_BLOCK;
566 if (errno == EINTR) {
567 goto retry;
569 error_setg_errno(errp, errno,
570 "Unable to write to socket");
571 return -1;
573 return ret;
575 #else /* WIN32 */
576 static ssize_t qio_channel_socket_readv(QIOChannel *ioc,
577 const struct iovec *iov,
578 size_t niov,
579 int **fds,
580 size_t *nfds,
581 Error **errp)
583 QIOChannelSocket *sioc = QIO_CHANNEL_SOCKET(ioc);
584 ssize_t done = 0;
585 ssize_t i;
587 for (i = 0; i < niov; i++) {
588 ssize_t ret;
589 retry:
590 ret = recv(sioc->fd,
591 iov[i].iov_base,
592 iov[i].iov_len,
594 if (ret < 0) {
595 if (errno == EAGAIN) {
596 if (done) {
597 return done;
598 } else {
599 return QIO_CHANNEL_ERR_BLOCK;
601 } else if (errno == EINTR) {
602 goto retry;
603 } else {
604 error_setg_errno(errp, errno,
605 "Unable to read from socket");
606 return -1;
609 done += ret;
610 if (ret < iov[i].iov_len) {
611 return done;
615 return done;
618 static ssize_t qio_channel_socket_writev(QIOChannel *ioc,
619 const struct iovec *iov,
620 size_t niov,
621 int *fds,
622 size_t nfds,
623 Error **errp)
625 QIOChannelSocket *sioc = QIO_CHANNEL_SOCKET(ioc);
626 ssize_t done = 0;
627 ssize_t i;
629 for (i = 0; i < niov; i++) {
630 ssize_t ret;
631 retry:
632 ret = send(sioc->fd,
633 iov[i].iov_base,
634 iov[i].iov_len,
636 if (ret < 0) {
637 if (errno == EAGAIN) {
638 if (done) {
639 return done;
640 } else {
641 return QIO_CHANNEL_ERR_BLOCK;
643 } else if (errno == EINTR) {
644 goto retry;
645 } else {
646 error_setg_errno(errp, errno,
647 "Unable to write to socket");
648 return -1;
651 done += ret;
652 if (ret < iov[i].iov_len) {
653 return done;
657 return done;
659 #endif /* WIN32 */
661 static int
662 qio_channel_socket_set_blocking(QIOChannel *ioc,
663 bool enabled,
664 Error **errp)
666 QIOChannelSocket *sioc = QIO_CHANNEL_SOCKET(ioc);
668 if (enabled) {
669 qemu_set_block(sioc->fd);
670 } else {
671 qemu_set_nonblock(sioc->fd);
673 return 0;
677 static void
678 qio_channel_socket_set_delay(QIOChannel *ioc,
679 bool enabled)
681 QIOChannelSocket *sioc = QIO_CHANNEL_SOCKET(ioc);
682 int v = enabled ? 0 : 1;
684 qemu_setsockopt(sioc->fd,
685 IPPROTO_TCP, TCP_NODELAY,
686 &v, sizeof(v));
690 static void
691 qio_channel_socket_set_cork(QIOChannel *ioc,
692 bool enabled)
694 QIOChannelSocket *sioc = QIO_CHANNEL_SOCKET(ioc);
695 int v = enabled ? 1 : 0;
697 socket_set_cork(sioc->fd, v);
701 static int
702 qio_channel_socket_close(QIOChannel *ioc,
703 Error **errp)
705 QIOChannelSocket *sioc = QIO_CHANNEL_SOCKET(ioc);
706 int rc = 0;
707 Error *err = NULL;
709 if (sioc->fd != -1) {
710 #ifdef WIN32
711 WSAEventSelect(sioc->fd, NULL, 0);
712 #endif
713 if (qio_channel_has_feature(ioc, QIO_CHANNEL_FEATURE_LISTEN)) {
714 socket_listen_cleanup(sioc->fd, errp);
717 if (closesocket(sioc->fd) < 0) {
718 sioc->fd = -1;
719 error_setg_errno(&err, errno, "Unable to close socket");
720 error_propagate(errp, err);
721 return -1;
723 sioc->fd = -1;
725 return rc;
728 static int
729 qio_channel_socket_shutdown(QIOChannel *ioc,
730 QIOChannelShutdown how,
731 Error **errp)
733 QIOChannelSocket *sioc = QIO_CHANNEL_SOCKET(ioc);
734 int sockhow;
736 switch (how) {
737 case QIO_CHANNEL_SHUTDOWN_READ:
738 sockhow = SHUT_RD;
739 break;
740 case QIO_CHANNEL_SHUTDOWN_WRITE:
741 sockhow = SHUT_WR;
742 break;
743 case QIO_CHANNEL_SHUTDOWN_BOTH:
744 default:
745 sockhow = SHUT_RDWR;
746 break;
749 if (shutdown(sioc->fd, sockhow) < 0) {
750 error_setg_errno(errp, errno,
751 "Unable to shutdown socket");
752 return -1;
754 return 0;
757 static void qio_channel_socket_set_aio_fd_handler(QIOChannel *ioc,
758 AioContext *ctx,
759 IOHandler *io_read,
760 IOHandler *io_write,
761 void *opaque)
763 QIOChannelSocket *sioc = QIO_CHANNEL_SOCKET(ioc);
764 aio_set_fd_handler(ctx, sioc->fd, false, io_read, io_write, NULL, opaque);
767 static GSource *qio_channel_socket_create_watch(QIOChannel *ioc,
768 GIOCondition condition)
770 QIOChannelSocket *sioc = QIO_CHANNEL_SOCKET(ioc);
771 return qio_channel_create_socket_watch(ioc,
772 sioc->fd,
773 condition);
776 static void qio_channel_socket_class_init(ObjectClass *klass,
777 void *class_data G_GNUC_UNUSED)
779 QIOChannelClass *ioc_klass = QIO_CHANNEL_CLASS(klass);
781 ioc_klass->io_writev = qio_channel_socket_writev;
782 ioc_klass->io_readv = qio_channel_socket_readv;
783 ioc_klass->io_set_blocking = qio_channel_socket_set_blocking;
784 ioc_klass->io_close = qio_channel_socket_close;
785 ioc_klass->io_shutdown = qio_channel_socket_shutdown;
786 ioc_klass->io_set_cork = qio_channel_socket_set_cork;
787 ioc_klass->io_set_delay = qio_channel_socket_set_delay;
788 ioc_klass->io_create_watch = qio_channel_socket_create_watch;
789 ioc_klass->io_set_aio_fd_handler = qio_channel_socket_set_aio_fd_handler;
792 static const TypeInfo qio_channel_socket_info = {
793 .parent = TYPE_QIO_CHANNEL,
794 .name = TYPE_QIO_CHANNEL_SOCKET,
795 .instance_size = sizeof(QIOChannelSocket),
796 .instance_init = qio_channel_socket_init,
797 .instance_finalize = qio_channel_socket_finalize,
798 .class_init = qio_channel_socket_class_init,
801 static void qio_channel_socket_register_types(void)
803 type_register_static(&qio_channel_socket_info);
806 type_init(qio_channel_socket_register_types);