4 * Copyright (c) 2015 Red Hat, Inc.
7 * Victor Kaplansky <victork@redhat.com>
9 * This work is licensed under the terms of the GNU GPL, version 2 or
10 * later. See the COPYING file in the top-level directory.
15 * - main should get parameters from the command line.
16 * - implement all request handlers. Still not implemented:
17 * vubr_get_queue_num_exec()
18 * vubr_send_rarp_exec()
19 * - test for broken requests and virtqueue.
20 * - implement features defined by Virtio 1.0 spec.
21 * - support mergeable buffers and indirect descriptors.
22 * - implement clean shutdown.
23 * - implement non-blocking writes to UDP backend.
24 * - implement polling strategy.
25 * - implement clean starting/stopping of vq processing
26 * - implement clean starting/stopping of used and buffers
30 #define _FILE_OFFSET_BITS 64
32 #include "qemu/atomic.h"
33 #include "qemu/osdep.h"
35 #include "standard-headers/linux/virtio_net.h"
36 #include "contrib/libvhost-user/libvhost-user.h"
38 #define VHOST_USER_BRIDGE_DEBUG 1
42 if (VHOST_USER_BRIDGE_DEBUG) { \
43 printf(__VA_ARGS__); \
47 typedef void (*CallbackFunc
)(int sock
, void *ctx
);
49 typedef struct Event
{
51 CallbackFunc callback
;
54 typedef struct Dispatcher
{
57 Event events
[FD_SETSIZE
];
60 typedef struct VubrDev
{
62 Dispatcher dispatcher
;
64 struct sockaddr_in backend_udp_dest
;
77 vubr_die(const char *s
)
84 dispatcher_init(Dispatcher
*dispr
)
86 FD_ZERO(&dispr
->fdset
);
92 dispatcher_add(Dispatcher
*dispr
, int sock
, void *ctx
, CallbackFunc cb
)
94 if (sock
>= FD_SETSIZE
) {
96 "Error: Failed to add new event. sock %d should be less than %d\n",
101 dispr
->events
[sock
].ctx
= ctx
;
102 dispr
->events
[sock
].callback
= cb
;
104 FD_SET(sock
, &dispr
->fdset
);
105 if (sock
> dispr
->max_sock
) {
106 dispr
->max_sock
= sock
;
108 DPRINT("Added sock %d for watching. max_sock: %d\n",
109 sock
, dispr
->max_sock
);
114 dispatcher_remove(Dispatcher
*dispr
, int sock
)
116 if (sock
>= FD_SETSIZE
) {
118 "Error: Failed to remove event. sock %d should be less than %d\n",
123 FD_CLR(sock
, &dispr
->fdset
);
124 DPRINT("Sock %d removed from dispatcher watch.\n", sock
);
130 dispatcher_wait(Dispatcher
*dispr
, uint32_t timeout
)
133 tv
.tv_sec
= timeout
/ 1000000;
134 tv
.tv_usec
= timeout
% 1000000;
136 fd_set fdset
= dispr
->fdset
;
138 /* wait until some of sockets become readable. */
139 int rc
= select(dispr
->max_sock
+ 1, &fdset
, 0, 0, &tv
);
150 /* Now call callback for every ready socket. */
153 for (sock
= 0; sock
< dispr
->max_sock
+ 1; sock
++) {
154 /* The callback on a socket can remove other sockets from the
155 * dispatcher, thus we have to check that the socket is
156 * still not removed from dispatcher's list
158 if (FD_ISSET(sock
, &fdset
) && FD_ISSET(sock
, &dispr
->fdset
)) {
159 Event
*e
= &dispr
->events
[sock
];
160 e
->callback(sock
, e
->ctx
);
168 vubr_handle_tx(VuDev
*dev
, int qidx
)
170 VuVirtq
*vq
= vu_get_queue(dev
, qidx
);
171 VubrDev
*vubr
= container_of(dev
, VubrDev
, vudev
);
172 int hdrlen
= vubr
->hdrlen
;
173 VuVirtqElement
*elem
= NULL
;
179 unsigned int out_num
;
180 struct iovec sg
[VIRTQUEUE_MAX_SIZE
], *out_sg
;
182 elem
= vu_queue_pop(dev
, vq
, sizeof(VuVirtqElement
));
187 out_num
= elem
->out_num
;
188 out_sg
= elem
->out_sg
;
190 fprintf(stderr
, "virtio-net header not in first element\n");
193 if (VHOST_USER_BRIDGE_DEBUG
) {
194 iov_hexdump(out_sg
, out_num
, stderr
, "TX:", 1024);
198 unsigned sg_num
= iov_copy(sg
, ARRAY_SIZE(sg
),
205 struct msghdr msg
= {
206 .msg_name
= (struct sockaddr
*) &vubr
->backend_udp_dest
,
207 .msg_namelen
= sizeof(struct sockaddr_in
),
209 .msg_iovlen
= out_num
,
212 ret
= sendmsg(vubr
->backend_udp_sock
, &msg
, 0);
213 } while (ret
== -1 && (errno
== EAGAIN
|| errno
== EINTR
));
216 vubr_die("sendmsg()");
219 vu_queue_push(dev
, vq
, elem
, 0);
220 vu_queue_notify(dev
, vq
);
230 /* this function reverse the effect of iov_discard_front() it must be
231 * called with 'front' being the original struct iovec and 'bytes'
232 * being the number of bytes you shaved off
235 iov_restore_front(struct iovec
*front
, struct iovec
*iov
, size_t bytes
)
239 for (cur
= front
; cur
!= iov
; cur
++) {
240 assert(bytes
>= cur
->iov_len
);
241 bytes
-= cur
->iov_len
;
244 cur
->iov_base
-= bytes
;
245 cur
->iov_len
+= bytes
;
249 iov_truncate(struct iovec
*iov
, unsigned iovc
, size_t bytes
)
253 for (i
= 0; i
< iovc
; i
++, iov
++) {
254 if (bytes
< iov
->iov_len
) {
255 iov
->iov_len
= bytes
;
259 bytes
-= iov
->iov_len
;
262 assert(!"couldn't truncate iov");
266 vubr_backend_recv_cb(int sock
, void *ctx
)
268 VubrDev
*vubr
= (VubrDev
*) ctx
;
269 VuDev
*dev
= &vubr
->vudev
;
270 VuVirtq
*vq
= vu_get_queue(dev
, 0);
271 VuVirtqElement
*elem
= NULL
;
272 struct iovec mhdr_sg
[VIRTQUEUE_MAX_SIZE
];
273 struct virtio_net_hdr_mrg_rxbuf mhdr
;
274 unsigned mhdr_cnt
= 0;
275 int hdrlen
= vubr
->hdrlen
;
277 struct virtio_net_hdr hdr
= {
279 .gso_type
= VIRTIO_NET_HDR_GSO_NONE
282 DPRINT("\n\n *** IN UDP RECEIVE CALLBACK ***\n\n");
283 DPRINT(" hdrlen = %d\n", hdrlen
);
285 if (!vu_queue_enabled(dev
, vq
) ||
286 !vu_queue_started(dev
, vq
) ||
287 !vu_queue_avail_bytes(dev
, vq
, hdrlen
, 0)) {
288 DPRINT("Got UDP packet, but no available descriptors on RX virtq.\n");
294 ssize_t ret
, total
= 0;
297 elem
= vu_queue_pop(dev
, vq
, sizeof(VuVirtqElement
));
302 if (elem
->in_num
< 1) {
303 fprintf(stderr
, "virtio-net contains no in buffers\n");
311 mhdr_cnt
= iov_copy(mhdr_sg
, ARRAY_SIZE(mhdr_sg
),
313 offsetof(typeof(mhdr
), num_buffers
),
314 sizeof(mhdr
.num_buffers
));
316 iov_from_buf(sg
, elem
->in_num
, 0, &hdr
, sizeof hdr
);
318 ret
= iov_discard_front(&sg
, &num
, hdrlen
);
319 assert(ret
== hdrlen
);
322 struct msghdr msg
= {
323 .msg_name
= (struct sockaddr
*) &vubr
->backend_udp_dest
,
324 .msg_namelen
= sizeof(struct sockaddr_in
),
326 .msg_iovlen
= elem
->in_num
,
327 .msg_flags
= MSG_DONTWAIT
,
330 ret
= recvmsg(vubr
->backend_udp_sock
, &msg
, 0);
331 } while (ret
== -1 && (errno
== EINTR
));
334 iov_restore_front(elem
->in_sg
, sg
, hdrlen
);
338 if (errno
== EWOULDBLOCK
) {
339 vu_queue_rewind(dev
, vq
, 1);
343 vubr_die("recvmsg()");
347 iov_truncate(elem
->in_sg
, elem
->in_num
, total
);
348 vu_queue_fill(dev
, vq
, elem
, total
, i
++);
353 break; /* could loop if DONTWAIT worked? */
357 mhdr
.num_buffers
= i
;
358 iov_from_buf(mhdr_sg
, mhdr_cnt
,
360 &mhdr
.num_buffers
, sizeof mhdr
.num_buffers
);
363 vu_queue_flush(dev
, vq
, i
);
364 vu_queue_notify(dev
, vq
);
370 vubr_receive_cb(int sock
, void *ctx
)
372 VubrDev
*vubr
= (VubrDev
*)ctx
;
374 if (!vu_dispatch(&vubr
->vudev
)) {
375 fprintf(stderr
, "Error while dispatching\n");
379 typedef struct WatchData
{
386 watch_cb(int sock
, void *ctx
)
388 struct WatchData
*wd
= ctx
;
390 wd
->cb(wd
->dev
, VU_WATCH_IN
, wd
->data
);
394 vubr_set_watch(VuDev
*dev
, int fd
, int condition
,
395 vu_watch_cb cb
, void *data
)
397 VubrDev
*vubr
= container_of(dev
, VubrDev
, vudev
);
398 static WatchData watches
[FD_SETSIZE
];
399 struct WatchData
*wd
= &watches
[fd
];
404 dispatcher_add(&vubr
->dispatcher
, fd
, wd
, watch_cb
);
408 vubr_remove_watch(VuDev
*dev
, int fd
)
410 VubrDev
*vubr
= container_of(dev
, VubrDev
, vudev
);
412 dispatcher_remove(&vubr
->dispatcher
, fd
);
416 vubr_send_rarp_exec(VuDev
*dev
, VhostUserMsg
*vmsg
)
418 DPRINT("Function %s() not implemented yet.\n", __func__
);
423 vubr_process_msg(VuDev
*dev
, VhostUserMsg
*vmsg
, int *do_reply
)
425 switch (vmsg
->request
) {
426 case VHOST_USER_SEND_RARP
:
427 *do_reply
= vubr_send_rarp_exec(dev
, vmsg
);
430 /* let the library handle the rest */
438 vubr_set_features(VuDev
*dev
, uint64_t features
)
440 VubrDev
*vubr
= container_of(dev
, VubrDev
, vudev
);
442 if ((features
& (1ULL << VIRTIO_F_VERSION_1
)) ||
443 (features
& (1ULL << VIRTIO_NET_F_MRG_RXBUF
))) {
451 vubr_get_features(VuDev
*dev
)
453 return 1ULL << VIRTIO_NET_F_GUEST_ANNOUNCE
|
454 1ULL << VIRTIO_NET_F_MRG_RXBUF
|
455 1ULL << VIRTIO_F_VERSION_1
;
459 vubr_queue_set_started(VuDev
*dev
, int qidx
, bool started
)
461 VubrDev
*vubr
= container_of(dev
, VubrDev
, vudev
);
462 VuVirtq
*vq
= vu_get_queue(dev
, qidx
);
464 if (started
&& vubr
->notifier
.fd
>= 0) {
465 vu_set_queue_host_notifier(dev
, vq
, vubr
->notifier
.fd
,
467 qidx
* getpagesize());
471 vu_set_queue_handler(dev
, vq
, started
? vubr_handle_tx
: NULL
);
476 vubr_panic(VuDev
*dev
, const char *msg
)
478 VubrDev
*vubr
= container_of(dev
, VubrDev
, vudev
);
480 fprintf(stderr
, "PANIC: %s\n", msg
);
482 dispatcher_remove(&vubr
->dispatcher
, dev
->sock
);
487 vubr_queue_is_processed_in_order(VuDev
*dev
, int qidx
)
492 static const VuDevIface vuiface
= {
493 .get_features
= vubr_get_features
,
494 .set_features
= vubr_set_features
,
495 .process_msg
= vubr_process_msg
,
496 .queue_set_started
= vubr_queue_set_started
,
497 .queue_is_processed_in_order
= vubr_queue_is_processed_in_order
,
501 vubr_accept_cb(int sock
, void *ctx
)
503 VubrDev
*dev
= (VubrDev
*)ctx
;
505 struct sockaddr_un un
;
506 socklen_t len
= sizeof(un
);
508 conn_fd
= accept(sock
, (struct sockaddr
*) &un
, &len
);
510 vubr_die("accept()");
512 DPRINT("Got connection from remote peer on sock %d\n", conn_fd
);
521 dispatcher_add(&dev
->dispatcher
, conn_fd
, ctx
, vubr_receive_cb
);
522 dispatcher_remove(&dev
->dispatcher
, sock
);
526 vubr_new(const char *path
, bool client
)
528 VubrDev
*dev
= (VubrDev
*) calloc(1, sizeof(VubrDev
));
529 struct sockaddr_un un
;
533 /* Get a UNIX socket. */
534 dev
->sock
= socket(AF_UNIX
, SOCK_STREAM
, 0);
535 if (dev
->sock
== -1) {
539 dev
->notifier
.fd
= -1;
541 un
.sun_family
= AF_UNIX
;
542 strcpy(un
.sun_path
, path
);
543 len
= sizeof(un
.sun_family
) + strlen(path
);
548 if (bind(dev
->sock
, (struct sockaddr
*) &un
, len
) == -1) {
552 if (listen(dev
->sock
, 1) == -1) {
557 DPRINT("Waiting for connections on UNIX socket %s ...\n", path
);
559 if (connect(dev
->sock
, (struct sockaddr
*)&un
, len
) == -1) {
568 cb
= vubr_receive_cb
;
571 dispatcher_init(&dev
->dispatcher
);
573 dispatcher_add(&dev
->dispatcher
, dev
->sock
, (void *)dev
, cb
);
578 static void *notifier_thread(void *arg
)
580 VuDev
*dev
= (VuDev
*)arg
;
581 VubrDev
*vubr
= container_of(dev
, VubrDev
, vudev
);
582 int pagesize
= getpagesize();
586 for (qidx
= 0; qidx
< VHOST_MAX_NR_VIRTQUEUE
; qidx
++) {
587 uint16_t *n
= vubr
->notifier
.addr
+ pagesize
* qidx
;
591 /* We won't miss notifications if we reset
592 * the memory first. */
595 DPRINT("Got a notification for queue%d via host notifier.\n",
599 vubr_handle_tx(dev
, qidx
);
610 vubr_host_notifier_setup(VubrDev
*dev
)
612 char template[] = "/tmp/vubr-XXXXXX";
618 length
= getpagesize() * VHOST_MAX_NR_VIRTQUEUE
;
620 fd
= mkstemp(template);
622 vubr_die("mkstemp()");
625 if (posix_fallocate(fd
, 0, length
) != 0) {
626 vubr_die("posix_fallocate()");
629 addr
= mmap(NULL
, length
, PROT_READ
| PROT_WRITE
, MAP_SHARED
, fd
, 0);
630 if (addr
== MAP_FAILED
) {
634 memset(addr
, 0xff, length
);
636 if (pthread_create(&thread
, NULL
, notifier_thread
, &dev
->vudev
) != 0) {
637 vubr_die("pthread_create()");
640 dev
->notifier
.fd
= fd
;
641 dev
->notifier
.addr
= addr
;
642 dev
->notifier
.thread
= thread
;
646 vubr_set_host(struct sockaddr_in
*saddr
, const char *host
)
648 if (isdigit(host
[0])) {
649 if (!inet_aton(host
, &saddr
->sin_addr
)) {
650 fprintf(stderr
, "inet_aton() failed.\n");
654 struct hostent
*he
= gethostbyname(host
);
657 fprintf(stderr
, "gethostbyname() failed.\n");
660 saddr
->sin_addr
= *(struct in_addr
*)he
->h_addr
;
665 vubr_backend_udp_setup(VubrDev
*dev
,
666 const char *local_host
,
667 const char *local_port
,
668 const char *remote_host
,
669 const char *remote_port
)
676 lport
= strtol(local_port
, (char **)&r
, 0);
677 if (r
== local_port
) {
678 fprintf(stderr
, "lport parsing failed.\n");
682 rport
= strtol(remote_port
, (char **)&r
, 0);
683 if (r
== remote_port
) {
684 fprintf(stderr
, "rport parsing failed.\n");
688 struct sockaddr_in si_local
= {
689 .sin_family
= AF_INET
,
690 .sin_port
= htons(lport
),
693 vubr_set_host(&si_local
, local_host
);
695 /* setup destination for sends */
696 dev
->backend_udp_dest
= (struct sockaddr_in
) {
697 .sin_family
= AF_INET
,
698 .sin_port
= htons(rport
),
700 vubr_set_host(&dev
->backend_udp_dest
, remote_host
);
702 sock
= socket(AF_INET
, SOCK_DGRAM
, IPPROTO_UDP
);
707 if (bind(sock
, (struct sockaddr
*)&si_local
, sizeof(si_local
)) == -1) {
711 dev
->backend_udp_sock
= sock
;
712 dispatcher_add(&dev
->dispatcher
, sock
, dev
, vubr_backend_recv_cb
);
713 DPRINT("Waiting for data from udp backend on %s:%d...\n",
718 vubr_run(VubrDev
*dev
)
722 dispatcher_wait(&dev
->dispatcher
, 200000);
723 /* Here one can try polling strategy. */
728 vubr_parse_host_port(const char **host
, const char **port
, const char *buf
)
730 char *p
= strchr(buf
, ':');
737 *port
= strdup(p
+ 1);
741 #define DEFAULT_UD_SOCKET "/tmp/vubr.sock"
742 #define DEFAULT_LHOST "127.0.0.1"
743 #define DEFAULT_LPORT "4444"
744 #define DEFAULT_RHOST "127.0.0.1"
745 #define DEFAULT_RPORT "5555"
747 static const char *ud_socket_path
= DEFAULT_UD_SOCKET
;
748 static const char *lhost
= DEFAULT_LHOST
;
749 static const char *lport
= DEFAULT_LPORT
;
750 static const char *rhost
= DEFAULT_RHOST
;
751 static const char *rport
= DEFAULT_RPORT
;
754 main(int argc
, char *argv
[])
759 bool host_notifier
= false;
761 while ((opt
= getopt(argc
, argv
, "l:r:u:cH")) != -1) {
765 if (vubr_parse_host_port(&lhost
, &lport
, optarg
) < 0) {
770 if (vubr_parse_host_port(&rhost
, &rport
, optarg
) < 0) {
775 ud_socket_path
= strdup(optarg
);
781 host_notifier
= true;
788 DPRINT("ud socket: %s (%s)\n", ud_socket_path
,
789 client
? "client" : "server");
790 DPRINT("local: %s:%s\n", lhost
, lport
);
791 DPRINT("remote: %s:%s\n", rhost
, rport
);
793 dev
= vubr_new(ud_socket_path
, client
);
799 vubr_host_notifier_setup(dev
);
802 vubr_backend_udp_setup(dev
, lhost
, lport
, rhost
, rport
);
805 vu_deinit(&dev
->vudev
);
810 fprintf(stderr
, "Usage: %s ", argv
[0]);
811 fprintf(stderr
, "[-c] [-H] [-u ud_socket_path] [-l lhost:lport] [-r rhost:rport]\n");
812 fprintf(stderr
, "\t-u path to unix doman socket. default: %s\n",
814 fprintf(stderr
, "\t-l local host and port. default: %s:%s\n",
815 DEFAULT_LHOST
, DEFAULT_LPORT
);
816 fprintf(stderr
, "\t-r remote host and port. default: %s:%s\n",
817 DEFAULT_RHOST
, DEFAULT_RPORT
);
818 fprintf(stderr
, "\t-c client mode\n");
819 fprintf(stderr
, "\t-H use host notifier\n");