1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21#include "qemu/osdep.h"
22#include "qapi/error.h"
23#include "qapi/qapi-visit-sockets.h"
24#include "io/channel-socket.h"
25#include "io/channel-watch.h"
26#include "trace.h"
27#include "qapi/clone-visitor.h"
28
29#define SOCKET_MAX_FDS 16
30
31SocketAddress *
32qio_channel_socket_get_local_address(QIOChannelSocket *ioc,
33 Error **errp)
34{
35 return socket_sockaddr_to_address(&ioc->localAddr,
36 ioc->localAddrLen,
37 errp);
38}
39
40SocketAddress *
41qio_channel_socket_get_remote_address(QIOChannelSocket *ioc,
42 Error **errp)
43{
44 return socket_sockaddr_to_address(&ioc->remoteAddr,
45 ioc->remoteAddrLen,
46 errp);
47}
48
49QIOChannelSocket *
50qio_channel_socket_new(void)
51{
52 QIOChannelSocket *sioc;
53 QIOChannel *ioc;
54
55 sioc = QIO_CHANNEL_SOCKET(object_new(TYPE_QIO_CHANNEL_SOCKET));
56 sioc->fd = -1;
57
58 ioc = QIO_CHANNEL(sioc);
59 qio_channel_set_feature(ioc, QIO_CHANNEL_FEATURE_SHUTDOWN);
60
61#ifdef WIN32
62 ioc->event = CreateEvent(NULL, FALSE, FALSE, NULL);
63#endif
64
65 trace_qio_channel_socket_new(sioc);
66
67 return sioc;
68}
69
70
71static int
72qio_channel_socket_set_fd(QIOChannelSocket *sioc,
73 int fd,
74 Error **errp)
75{
76 if (sioc->fd != -1) {
77 error_setg(errp, "Socket is already open");
78 return -1;
79 }
80
81 sioc->fd = fd;
82 sioc->remoteAddrLen = sizeof(sioc->remoteAddr);
83 sioc->localAddrLen = sizeof(sioc->localAddr);
84
85
86 if (getpeername(fd, (struct sockaddr *)&sioc->remoteAddr,
87 &sioc->remoteAddrLen) < 0) {
88 if (errno == ENOTCONN) {
89 memset(&sioc->remoteAddr, 0, sizeof(sioc->remoteAddr));
90 sioc->remoteAddrLen = sizeof(sioc->remoteAddr);
91 } else {
92 error_setg_errno(errp, errno,
93 "Unable to query remote socket address");
94 goto error;
95 }
96 }
97
98 if (getsockname(fd, (struct sockaddr *)&sioc->localAddr,
99 &sioc->localAddrLen) < 0) {
100 error_setg_errno(errp, errno,
101 "Unable to query local socket address");
102 goto error;
103 }
104
105#ifndef WIN32
106 if (sioc->localAddr.ss_family == AF_UNIX) {
107 QIOChannel *ioc = QIO_CHANNEL(sioc);
108 qio_channel_set_feature(ioc, QIO_CHANNEL_FEATURE_FD_PASS);
109 }
110#endif
111
112 return 0;
113
114 error:
115 sioc->fd = -1;
116 return -1;
117}
118
119QIOChannelSocket *
120qio_channel_socket_new_fd(int fd,
121 Error **errp)
122{
123 QIOChannelSocket *ioc;
124
125 ioc = qio_channel_socket_new();
126 if (qio_channel_socket_set_fd(ioc, fd, errp) < 0) {
127 object_unref(OBJECT(ioc));
128 return NULL;
129 }
130
131 trace_qio_channel_socket_new_fd(ioc, fd);
132
133 return ioc;
134}
135
136
137int qio_channel_socket_connect_sync(QIOChannelSocket *ioc,
138 SocketAddress *addr,
139 Error **errp)
140{
141 int fd;
142
143 trace_qio_channel_socket_connect_sync(ioc, addr);
144 fd = socket_connect(addr, errp);
145 if (fd < 0) {
146 trace_qio_channel_socket_connect_fail(ioc);
147 return -1;
148 }
149
150 trace_qio_channel_socket_connect_complete(ioc, fd);
151 if (qio_channel_socket_set_fd(ioc, fd, errp) < 0) {
152 close(fd);
153 return -1;
154 }
155
156 return 0;
157}
158
159
160static void qio_channel_socket_connect_worker(QIOTask *task,
161 gpointer opaque)
162{
163 QIOChannelSocket *ioc = QIO_CHANNEL_SOCKET(qio_task_get_source(task));
164 SocketAddress *addr = opaque;
165 Error *err = NULL;
166
167 qio_channel_socket_connect_sync(ioc, addr, &err);
168
169 qio_task_set_error(task, err);
170}
171
172
173void qio_channel_socket_connect_async(QIOChannelSocket *ioc,
174 SocketAddress *addr,
175 QIOTaskFunc callback,
176 gpointer opaque,
177 GDestroyNotify destroy,
178 GMainContext *context)
179{
180 QIOTask *task = qio_task_new(
181 OBJECT(ioc), callback, opaque, destroy);
182 SocketAddress *addrCopy;
183
184 addrCopy = QAPI_CLONE(SocketAddress, addr);
185
186
187
188 trace_qio_channel_socket_connect_async(ioc, addr);
189 qio_task_run_in_thread(task,
190 qio_channel_socket_connect_worker,
191 addrCopy,
192 (GDestroyNotify)qapi_free_SocketAddress,
193 context);
194}
195
196
197int qio_channel_socket_listen_sync(QIOChannelSocket *ioc,
198 SocketAddress *addr,
199 Error **errp)
200{
201 int fd;
202
203 trace_qio_channel_socket_listen_sync(ioc, addr);
204 fd = socket_listen(addr, errp);
205 if (fd < 0) {
206 trace_qio_channel_socket_listen_fail(ioc);
207 return -1;
208 }
209
210 trace_qio_channel_socket_listen_complete(ioc, fd);
211 if (qio_channel_socket_set_fd(ioc, fd, errp) < 0) {
212 close(fd);
213 return -1;
214 }
215 qio_channel_set_feature(QIO_CHANNEL(ioc), QIO_CHANNEL_FEATURE_LISTEN);
216
217 return 0;
218}
219
220
221static void qio_channel_socket_listen_worker(QIOTask *task,
222 gpointer opaque)
223{
224 QIOChannelSocket *ioc = QIO_CHANNEL_SOCKET(qio_task_get_source(task));
225 SocketAddress *addr = opaque;
226 Error *err = NULL;
227
228 qio_channel_socket_listen_sync(ioc, addr, &err);
229
230 qio_task_set_error(task, err);
231}
232
233
234void qio_channel_socket_listen_async(QIOChannelSocket *ioc,
235 SocketAddress *addr,
236 QIOTaskFunc callback,
237 gpointer opaque,
238 GDestroyNotify destroy,
239 GMainContext *context)
240{
241 QIOTask *task = qio_task_new(
242 OBJECT(ioc), callback, opaque, destroy);
243 SocketAddress *addrCopy;
244
245 addrCopy = QAPI_CLONE(SocketAddress, addr);
246
247
248 trace_qio_channel_socket_listen_async(ioc, addr);
249 qio_task_run_in_thread(task,
250 qio_channel_socket_listen_worker,
251 addrCopy,
252 (GDestroyNotify)qapi_free_SocketAddress,
253 context);
254}
255
256
257int qio_channel_socket_dgram_sync(QIOChannelSocket *ioc,
258 SocketAddress *localAddr,
259 SocketAddress *remoteAddr,
260 Error **errp)
261{
262 int fd;
263
264 trace_qio_channel_socket_dgram_sync(ioc, localAddr, remoteAddr);
265 fd = socket_dgram(remoteAddr, localAddr, errp);
266 if (fd < 0) {
267 trace_qio_channel_socket_dgram_fail(ioc);
268 return -1;
269 }
270
271 trace_qio_channel_socket_dgram_complete(ioc, fd);
272 if (qio_channel_socket_set_fd(ioc, fd, errp) < 0) {
273 close(fd);
274 return -1;
275 }
276
277 return 0;
278}
279
280
281struct QIOChannelSocketDGramWorkerData {
282 SocketAddress *localAddr;
283 SocketAddress *remoteAddr;
284};
285
286
287static void qio_channel_socket_dgram_worker_free(gpointer opaque)
288{
289 struct QIOChannelSocketDGramWorkerData *data = opaque;
290 qapi_free_SocketAddress(data->localAddr);
291 qapi_free_SocketAddress(data->remoteAddr);
292 g_free(data);
293}
294
295static void qio_channel_socket_dgram_worker(QIOTask *task,
296 gpointer opaque)
297{
298 QIOChannelSocket *ioc = QIO_CHANNEL_SOCKET(qio_task_get_source(task));
299 struct QIOChannelSocketDGramWorkerData *data = opaque;
300 Error *err = NULL;
301
302
303 qio_channel_socket_dgram_sync(ioc, data->localAddr,
304 data->remoteAddr, &err);
305
306 qio_task_set_error(task, err);
307}
308
309
310void qio_channel_socket_dgram_async(QIOChannelSocket *ioc,
311 SocketAddress *localAddr,
312 SocketAddress *remoteAddr,
313 QIOTaskFunc callback,
314 gpointer opaque,
315 GDestroyNotify destroy,
316 GMainContext *context)
317{
318 QIOTask *task = qio_task_new(
319 OBJECT(ioc), callback, opaque, destroy);
320 struct QIOChannelSocketDGramWorkerData *data = g_new0(
321 struct QIOChannelSocketDGramWorkerData, 1);
322
323 data->localAddr = QAPI_CLONE(SocketAddress, localAddr);
324 data->remoteAddr = QAPI_CLONE(SocketAddress, remoteAddr);
325
326 trace_qio_channel_socket_dgram_async(ioc, localAddr, remoteAddr);
327 qio_task_run_in_thread(task,
328 qio_channel_socket_dgram_worker,
329 data,
330 qio_channel_socket_dgram_worker_free,
331 context);
332}
333
334
335QIOChannelSocket *
336qio_channel_socket_accept(QIOChannelSocket *ioc,
337 Error **errp)
338{
339 QIOChannelSocket *cioc;
340
341 cioc = qio_channel_socket_new();
342 cioc->remoteAddrLen = sizeof(ioc->remoteAddr);
343 cioc->localAddrLen = sizeof(ioc->localAddr);
344
345 retry:
346 trace_qio_channel_socket_accept(ioc);
347 cioc->fd = qemu_accept(ioc->fd, (struct sockaddr *)&cioc->remoteAddr,
348 &cioc->remoteAddrLen);
349 if (cioc->fd < 0) {
350 if (errno == EINTR) {
351 goto retry;
352 }
353 error_setg_errno(errp, errno, "Unable to accept connection");
354 trace_qio_channel_socket_accept_fail(ioc);
355 goto error;
356 }
357
358 if (getsockname(cioc->fd, (struct sockaddr *)&cioc->localAddr,
359 &cioc->localAddrLen) < 0) {
360 error_setg_errno(errp, errno,
361 "Unable to query local socket address");
362 goto error;
363 }
364
365#ifndef WIN32
366 if (cioc->localAddr.ss_family == AF_UNIX) {
367 QIOChannel *ioc_local = QIO_CHANNEL(cioc);
368 qio_channel_set_feature(ioc_local, QIO_CHANNEL_FEATURE_FD_PASS);
369 }
370#endif
371
372 trace_qio_channel_socket_accept_complete(ioc, cioc, cioc->fd);
373 return cioc;
374
375 error:
376 object_unref(OBJECT(cioc));
377 return NULL;
378}
379
380static void qio_channel_socket_init(Object *obj)
381{
382 QIOChannelSocket *ioc = QIO_CHANNEL_SOCKET(obj);
383 ioc->fd = -1;
384}
385
386static void qio_channel_socket_finalize(Object *obj)
387{
388 QIOChannelSocket *ioc = QIO_CHANNEL_SOCKET(obj);
389
390 if (ioc->fd != -1) {
391 QIOChannel *ioc_local = QIO_CHANNEL(ioc);
392 if (qio_channel_has_feature(ioc_local, QIO_CHANNEL_FEATURE_LISTEN)) {
393 Error *err = NULL;
394
395 socket_listen_cleanup(ioc->fd, &err);
396 if (err) {
397 error_report_err(err);
398 err = NULL;
399 }
400 }
401#ifdef WIN32
402 WSAEventSelect(ioc->fd, NULL, 0);
403#endif
404 closesocket(ioc->fd);
405 ioc->fd = -1;
406 }
407}
408
409
410#ifndef WIN32
411static void qio_channel_socket_copy_fds(struct msghdr *msg,
412 int **fds, size_t *nfds)
413{
414 struct cmsghdr *cmsg;
415
416 *nfds = 0;
417 *fds = NULL;
418
419 for (cmsg = CMSG_FIRSTHDR(msg); cmsg; cmsg = CMSG_NXTHDR(msg, cmsg)) {
420 int fd_size, i;
421 int gotfds;
422
423 if (cmsg->cmsg_len < CMSG_LEN(sizeof(int)) ||
424 cmsg->cmsg_level != SOL_SOCKET ||
425 cmsg->cmsg_type != SCM_RIGHTS) {
426 continue;
427 }
428
429 fd_size = cmsg->cmsg_len - CMSG_LEN(0);
430
431 if (!fd_size) {
432 continue;
433 }
434
435 gotfds = fd_size / sizeof(int);
436 *fds = g_renew(int, *fds, *nfds + gotfds);
437 memcpy(*fds + *nfds, CMSG_DATA(cmsg), fd_size);
438
439 for (i = 0; i < gotfds; i++) {
440 int fd = (*fds)[*nfds + i];
441 if (fd < 0) {
442 continue;
443 }
444
445
446 qemu_set_block(fd);
447
448#ifndef MSG_CMSG_CLOEXEC
449 qemu_set_cloexec(fd);
450#endif
451 }
452 *nfds += gotfds;
453 }
454}
455
456
457static ssize_t qio_channel_socket_readv(QIOChannel *ioc,
458 const struct iovec *iov,
459 size_t niov,
460 int **fds,
461 size_t *nfds,
462 Error **errp)
463{
464 QIOChannelSocket *sioc = QIO_CHANNEL_SOCKET(ioc);
465 ssize_t ret;
466 struct msghdr msg = { NULL, };
467 char control[CMSG_SPACE(sizeof(int) * SOCKET_MAX_FDS)];
468 int sflags = 0;
469
470 memset(control, 0, CMSG_SPACE(sizeof(int) * SOCKET_MAX_FDS));
471
472#ifdef MSG_CMSG_CLOEXEC
473 sflags |= MSG_CMSG_CLOEXEC;
474#endif
475
476 msg.msg_iov = (struct iovec *)iov;
477 msg.msg_iovlen = niov;
478 if (fds && nfds) {
479 msg.msg_control = control;
480 msg.msg_controllen = sizeof(control);
481 }
482
483 retry:
484 ret = recvmsg(sioc->fd, &msg, sflags);
485 if (ret < 0) {
486 if (errno == EAGAIN) {
487 return QIO_CHANNEL_ERR_BLOCK;
488 }
489 if (errno == EINTR) {
490 goto retry;
491 }
492
493 error_setg_errno(errp, errno,
494 "Unable to read from socket");
495 return -1;
496 }
497
498 if (fds && nfds) {
499 qio_channel_socket_copy_fds(&msg, fds, nfds);
500 }
501
502 return ret;
503}
504
505static ssize_t qio_channel_socket_writev(QIOChannel *ioc,
506 const struct iovec *iov,
507 size_t niov,
508 int *fds,
509 size_t nfds,
510 Error **errp)
511{
512 QIOChannelSocket *sioc = QIO_CHANNEL_SOCKET(ioc);
513 ssize_t ret;
514 struct msghdr msg = { NULL, };
515 char control[CMSG_SPACE(sizeof(int) * SOCKET_MAX_FDS)];
516 size_t fdsize = sizeof(int) * nfds;
517 struct cmsghdr *cmsg;
518
519 memset(control, 0, CMSG_SPACE(sizeof(int) * SOCKET_MAX_FDS));
520
521 msg.msg_iov = (struct iovec *)iov;
522 msg.msg_iovlen = niov;
523
524 if (nfds) {
525 if (nfds > SOCKET_MAX_FDS) {
526 error_setg_errno(errp, EINVAL,
527 "Only %d FDs can be sent, got %zu",
528 SOCKET_MAX_FDS, nfds);
529 return -1;
530 }
531
532 msg.msg_control = control;
533 msg.msg_controllen = CMSG_SPACE(sizeof(int) * nfds);
534
535 cmsg = CMSG_FIRSTHDR(&msg);
536 cmsg->cmsg_len = CMSG_LEN(fdsize);
537 cmsg->cmsg_level = SOL_SOCKET;
538 cmsg->cmsg_type = SCM_RIGHTS;
539 memcpy(CMSG_DATA(cmsg), fds, fdsize);
540 }
541
542 retry:
543 ret = sendmsg(sioc->fd, &msg, 0);
544 if (ret <= 0) {
545 if (errno == EAGAIN) {
546 return QIO_CHANNEL_ERR_BLOCK;
547 }
548 if (errno == EINTR) {
549 goto retry;
550 }
551 error_setg_errno(errp, errno,
552 "Unable to write to socket");
553 return -1;
554 }
555 return ret;
556}
557#else
558static ssize_t qio_channel_socket_readv(QIOChannel *ioc,
559 const struct iovec *iov,
560 size_t niov,
561 int **fds,
562 size_t *nfds,
563 Error **errp)
564{
565 QIOChannelSocket *sioc = QIO_CHANNEL_SOCKET(ioc);
566 ssize_t done = 0;
567 ssize_t i;
568
569 for (i = 0; i < niov; i++) {
570 ssize_t ret;
571 retry:
572 ret = recv(sioc->fd,
573 iov[i].iov_base,
574 iov[i].iov_len,
575 0);
576 if (ret < 0) {
577 if (errno == EAGAIN) {
578 if (done) {
579 return done;
580 } else {
581 return QIO_CHANNEL_ERR_BLOCK;
582 }
583 } else if (errno == EINTR) {
584 goto retry;
585 } else {
586 error_setg_errno(errp, errno,
587 "Unable to read from socket");
588 return -1;
589 }
590 }
591 done += ret;
592 if (ret < iov[i].iov_len) {
593 return done;
594 }
595 }
596
597 return done;
598}
599
600static ssize_t qio_channel_socket_writev(QIOChannel *ioc,
601 const struct iovec *iov,
602 size_t niov,
603 int *fds,
604 size_t nfds,
605 Error **errp)
606{
607 QIOChannelSocket *sioc = QIO_CHANNEL_SOCKET(ioc);
608 ssize_t done = 0;
609 ssize_t i;
610
611 for (i = 0; i < niov; i++) {
612 ssize_t ret;
613 retry:
614 ret = send(sioc->fd,
615 iov[i].iov_base,
616 iov[i].iov_len,
617 0);
618 if (ret < 0) {
619 if (errno == EAGAIN) {
620 if (done) {
621 return done;
622 } else {
623 return QIO_CHANNEL_ERR_BLOCK;
624 }
625 } else if (errno == EINTR) {
626 goto retry;
627 } else {
628 error_setg_errno(errp, errno,
629 "Unable to write to socket");
630 return -1;
631 }
632 }
633 done += ret;
634 if (ret < iov[i].iov_len) {
635 return done;
636 }
637 }
638
639 return done;
640}
641#endif
642
643static int
644qio_channel_socket_set_blocking(QIOChannel *ioc,
645 bool enabled,
646 Error **errp)
647{
648 QIOChannelSocket *sioc = QIO_CHANNEL_SOCKET(ioc);
649
650 if (enabled) {
651 qemu_set_block(sioc->fd);
652 } else {
653 qemu_set_nonblock(sioc->fd);
654 }
655 return 0;
656}
657
658
659static void
660qio_channel_socket_set_delay(QIOChannel *ioc,
661 bool enabled)
662{
663 QIOChannelSocket *sioc = QIO_CHANNEL_SOCKET(ioc);
664 int v = enabled ? 0 : 1;
665
666 qemu_setsockopt(sioc->fd,
667 IPPROTO_TCP, TCP_NODELAY,
668 &v, sizeof(v));
669}
670
671
672static void
673qio_channel_socket_set_cork(QIOChannel *ioc,
674 bool enabled)
675{
676 QIOChannelSocket *sioc = QIO_CHANNEL_SOCKET(ioc);
677 int v = enabled ? 1 : 0;
678
679 socket_set_cork(sioc->fd, v);
680}
681
682
683static int
684qio_channel_socket_close(QIOChannel *ioc,
685 Error **errp)
686{
687 QIOChannelSocket *sioc = QIO_CHANNEL_SOCKET(ioc);
688 int rc = 0;
689
690 if (sioc->fd != -1) {
691 SocketAddress *addr = socket_local_address(sioc->fd, errp);
692#ifdef WIN32
693 WSAEventSelect(sioc->fd, NULL, 0);
694#endif
695 if (closesocket(sioc->fd) < 0) {
696 sioc->fd = -1;
697 error_setg_errno(errp, errno,
698 "Unable to close socket");
699 return -1;
700 }
701 sioc->fd = -1;
702
703 if (addr && addr->type == SOCKET_ADDRESS_TYPE_UNIX
704 && addr->u.q_unix.path) {
705 if (unlink(addr->u.q_unix.path) < 0 && errno != ENOENT) {
706 error_setg_errno(errp, errno,
707 "Failed to unlink socket %s",
708 addr->u.q_unix.path);
709 rc = -1;
710 }
711 }
712
713 if (addr) {
714 qapi_free_SocketAddress(addr);
715 }
716 }
717 return rc;
718}
719
720static int
721qio_channel_socket_shutdown(QIOChannel *ioc,
722 QIOChannelShutdown how,
723 Error **errp)
724{
725 QIOChannelSocket *sioc = QIO_CHANNEL_SOCKET(ioc);
726 int sockhow;
727
728 switch (how) {
729 case QIO_CHANNEL_SHUTDOWN_READ:
730 sockhow = SHUT_RD;
731 break;
732 case QIO_CHANNEL_SHUTDOWN_WRITE:
733 sockhow = SHUT_WR;
734 break;
735 case QIO_CHANNEL_SHUTDOWN_BOTH:
736 default:
737 sockhow = SHUT_RDWR;
738 break;
739 }
740
741 if (shutdown(sioc->fd, sockhow) < 0) {
742 error_setg_errno(errp, errno,
743 "Unable to shutdown socket");
744 return -1;
745 }
746 return 0;
747}
748
749static void qio_channel_socket_set_aio_fd_handler(QIOChannel *ioc,
750 AioContext *ctx,
751 IOHandler *io_read,
752 IOHandler *io_write,
753 void *opaque)
754{
755 QIOChannelSocket *sioc = QIO_CHANNEL_SOCKET(ioc);
756 aio_set_fd_handler(ctx, sioc->fd, false, io_read, io_write, NULL, opaque);
757}
758
759static GSource *qio_channel_socket_create_watch(QIOChannel *ioc,
760 GIOCondition condition)
761{
762 QIOChannelSocket *sioc = QIO_CHANNEL_SOCKET(ioc);
763 return qio_channel_create_socket_watch(ioc,
764 sioc->fd,
765 condition);
766}
767
768static void qio_channel_socket_class_init(ObjectClass *klass,
769 void *class_data G_GNUC_UNUSED)
770{
771 QIOChannelClass *ioc_klass = QIO_CHANNEL_CLASS(klass);
772
773 ioc_klass->io_writev = qio_channel_socket_writev;
774 ioc_klass->io_readv = qio_channel_socket_readv;
775 ioc_klass->io_set_blocking = qio_channel_socket_set_blocking;
776 ioc_klass->io_close = qio_channel_socket_close;
777 ioc_klass->io_shutdown = qio_channel_socket_shutdown;
778 ioc_klass->io_set_cork = qio_channel_socket_set_cork;
779 ioc_klass->io_set_delay = qio_channel_socket_set_delay;
780 ioc_klass->io_create_watch = qio_channel_socket_create_watch;
781 ioc_klass->io_set_aio_fd_handler = qio_channel_socket_set_aio_fd_handler;
782}
783
784static const TypeInfo qio_channel_socket_info = {
785 .parent = TYPE_QIO_CHANNEL,
786 .name = TYPE_QIO_CHANNEL_SOCKET,
787 .instance_size = sizeof(QIOChannelSocket),
788 .instance_init = qio_channel_socket_init,
789 .instance_finalize = qio_channel_socket_finalize,
790 .class_init = qio_channel_socket_class_init,
791};
792
793static void qio_channel_socket_register_types(void)
794{
795 type_register_static(&qio_channel_socket_info);
796}
797
798type_init(qio_channel_socket_register_types);
799
