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