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