quorum: Inline quorum_aio_cb()
[qemu.git] / tests / vhost-user-bridge.c
1 /*
2 * Vhost User Bridge
3 *
4 * Copyright (c) 2015 Red Hat, Inc.
5 *
6 * Authors:
7 * Victor Kaplansky <victork@redhat.com>
8 *
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.
11 */
12
13 /*
14 * TODO:
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
27 * dirty page logging.
28 */
29
30 #define _FILE_OFFSET_BITS 64
31
32 #include "qemu/osdep.h"
33 #include <sys/socket.h>
34 #include <sys/un.h>
35 #include <sys/unistd.h>
36 #include <sys/eventfd.h>
37 #include <arpa/inet.h>
38 #include <netdb.h>
39 #include <linux/vhost.h>
40
41 #include "qemu/atomic.h"
42 #include "standard-headers/linux/virtio_net.h"
43 #include "standard-headers/linux/virtio_ring.h"
44
45 #define VHOST_USER_BRIDGE_DEBUG 1
46
47 #define DPRINT(...) \
48 do { \
49 if (VHOST_USER_BRIDGE_DEBUG) { \
50 printf(__VA_ARGS__); \
51 } \
52 } while (0)
53
54 typedef void (*CallbackFunc)(int sock, void *ctx);
55
56 typedef struct Event {
57 void *ctx;
58 CallbackFunc callback;
59 } Event;
60
61 typedef struct Dispatcher {
62 int max_sock;
63 fd_set fdset;
64 Event events[FD_SETSIZE];
65 } Dispatcher;
66
67 static void
68 vubr_die(const char *s)
69 {
70 perror(s);
71 exit(1);
72 }
73
74 static int
75 dispatcher_init(Dispatcher *dispr)
76 {
77 FD_ZERO(&dispr->fdset);
78 dispr->max_sock = -1;
79 return 0;
80 }
81
82 static int
83 dispatcher_add(Dispatcher *dispr, int sock, void *ctx, CallbackFunc cb)
84 {
85 if (sock >= FD_SETSIZE) {
86 fprintf(stderr,
87 "Error: Failed to add new event. sock %d should be less than %d\n",
88 sock, FD_SETSIZE);
89 return -1;
90 }
91
92 dispr->events[sock].ctx = ctx;
93 dispr->events[sock].callback = cb;
94
95 FD_SET(sock, &dispr->fdset);
96 if (sock > dispr->max_sock) {
97 dispr->max_sock = sock;
98 }
99 DPRINT("Added sock %d for watching. max_sock: %d\n",
100 sock, dispr->max_sock);
101 return 0;
102 }
103
104 /* dispatcher_remove() is not currently in use but may be useful
105 * in the future. */
106 static int
107 dispatcher_remove(Dispatcher *dispr, int sock)
108 {
109 if (sock >= FD_SETSIZE) {
110 fprintf(stderr,
111 "Error: Failed to remove event. sock %d should be less than %d\n",
112 sock, FD_SETSIZE);
113 return -1;
114 }
115
116 FD_CLR(sock, &dispr->fdset);
117 DPRINT("Sock %d removed from dispatcher watch.\n", sock);
118 return 0;
119 }
120
121 /* timeout in us */
122 static int
123 dispatcher_wait(Dispatcher *dispr, uint32_t timeout)
124 {
125 struct timeval tv;
126 tv.tv_sec = timeout / 1000000;
127 tv.tv_usec = timeout % 1000000;
128
129 fd_set fdset = dispr->fdset;
130
131 /* wait until some of sockets become readable. */
132 int rc = select(dispr->max_sock + 1, &fdset, 0, 0, &tv);
133
134 if (rc == -1) {
135 vubr_die("select");
136 }
137
138 /* Timeout */
139 if (rc == 0) {
140 return 0;
141 }
142
143 /* Now call callback for every ready socket. */
144
145 int sock;
146 for (sock = 0; sock < dispr->max_sock + 1; sock++) {
147 /* The callback on a socket can remove other sockets from the
148 * dispatcher, thus we have to check that the socket is
149 * still not removed from dispatcher's list
150 */
151 if (FD_ISSET(sock, &fdset) && FD_ISSET(sock, &dispr->fdset)) {
152 Event *e = &dispr->events[sock];
153 e->callback(sock, e->ctx);
154 }
155 }
156
157 return 0;
158 }
159
160 typedef struct VubrVirtq {
161 int call_fd;
162 int kick_fd;
163 uint32_t size;
164 uint16_t last_avail_index;
165 uint16_t last_used_index;
166 struct vring_desc *desc;
167 struct vring_avail *avail;
168 struct vring_used *used;
169 uint64_t log_guest_addr;
170 int enable;
171 } VubrVirtq;
172
173 /* Based on qemu/hw/virtio/vhost-user.c */
174
175 #define VHOST_MEMORY_MAX_NREGIONS 8
176 #define VHOST_USER_F_PROTOCOL_FEATURES 30
177 /* v1.0 compliant. */
178 #define VIRTIO_F_VERSION_1 32
179
180 #define VHOST_LOG_PAGE 4096
181
182 enum VhostUserProtocolFeature {
183 VHOST_USER_PROTOCOL_F_MQ = 0,
184 VHOST_USER_PROTOCOL_F_LOG_SHMFD = 1,
185 VHOST_USER_PROTOCOL_F_RARP = 2,
186
187 VHOST_USER_PROTOCOL_F_MAX
188 };
189
190 #define VHOST_USER_PROTOCOL_FEATURE_MASK ((1 << VHOST_USER_PROTOCOL_F_MAX) - 1)
191
192 typedef enum VhostUserRequest {
193 VHOST_USER_NONE = 0,
194 VHOST_USER_GET_FEATURES = 1,
195 VHOST_USER_SET_FEATURES = 2,
196 VHOST_USER_SET_OWNER = 3,
197 VHOST_USER_RESET_OWNER = 4,
198 VHOST_USER_SET_MEM_TABLE = 5,
199 VHOST_USER_SET_LOG_BASE = 6,
200 VHOST_USER_SET_LOG_FD = 7,
201 VHOST_USER_SET_VRING_NUM = 8,
202 VHOST_USER_SET_VRING_ADDR = 9,
203 VHOST_USER_SET_VRING_BASE = 10,
204 VHOST_USER_GET_VRING_BASE = 11,
205 VHOST_USER_SET_VRING_KICK = 12,
206 VHOST_USER_SET_VRING_CALL = 13,
207 VHOST_USER_SET_VRING_ERR = 14,
208 VHOST_USER_GET_PROTOCOL_FEATURES = 15,
209 VHOST_USER_SET_PROTOCOL_FEATURES = 16,
210 VHOST_USER_GET_QUEUE_NUM = 17,
211 VHOST_USER_SET_VRING_ENABLE = 18,
212 VHOST_USER_SEND_RARP = 19,
213 VHOST_USER_MAX
214 } VhostUserRequest;
215
216 typedef struct VhostUserMemoryRegion {
217 uint64_t guest_phys_addr;
218 uint64_t memory_size;
219 uint64_t userspace_addr;
220 uint64_t mmap_offset;
221 } VhostUserMemoryRegion;
222
223 typedef struct VhostUserMemory {
224 uint32_t nregions;
225 uint32_t padding;
226 VhostUserMemoryRegion regions[VHOST_MEMORY_MAX_NREGIONS];
227 } VhostUserMemory;
228
229 typedef struct VhostUserLog {
230 uint64_t mmap_size;
231 uint64_t mmap_offset;
232 } VhostUserLog;
233
234 typedef struct VhostUserMsg {
235 VhostUserRequest request;
236
237 #define VHOST_USER_VERSION_MASK (0x3)
238 #define VHOST_USER_REPLY_MASK (0x1<<2)
239 uint32_t flags;
240 uint32_t size; /* the following payload size */
241 union {
242 #define VHOST_USER_VRING_IDX_MASK (0xff)
243 #define VHOST_USER_VRING_NOFD_MASK (0x1<<8)
244 uint64_t u64;
245 struct vhost_vring_state state;
246 struct vhost_vring_addr addr;
247 VhostUserMemory memory;
248 VhostUserLog log;
249 } payload;
250 int fds[VHOST_MEMORY_MAX_NREGIONS];
251 int fd_num;
252 } QEMU_PACKED VhostUserMsg;
253
254 #define VHOST_USER_HDR_SIZE offsetof(VhostUserMsg, payload.u64)
255
256 /* The version of the protocol we support */
257 #define VHOST_USER_VERSION (0x1)
258
259 #define MAX_NR_VIRTQUEUE (8)
260
261 typedef struct VubrDevRegion {
262 /* Guest Physical address. */
263 uint64_t gpa;
264 /* Memory region size. */
265 uint64_t size;
266 /* QEMU virtual address (userspace). */
267 uint64_t qva;
268 /* Starting offset in our mmaped space. */
269 uint64_t mmap_offset;
270 /* Start address of mmaped space. */
271 uint64_t mmap_addr;
272 } VubrDevRegion;
273
274 typedef struct VubrDev {
275 int sock;
276 Dispatcher dispatcher;
277 uint32_t nregions;
278 VubrDevRegion regions[VHOST_MEMORY_MAX_NREGIONS];
279 VubrVirtq vq[MAX_NR_VIRTQUEUE];
280 int log_call_fd;
281 uint64_t log_size;
282 uint8_t *log_table;
283 int backend_udp_sock;
284 struct sockaddr_in backend_udp_dest;
285 int ready;
286 uint64_t features;
287 int hdrlen;
288 } VubrDev;
289
290 static const char *vubr_request_str[] = {
291 [VHOST_USER_NONE] = "VHOST_USER_NONE",
292 [VHOST_USER_GET_FEATURES] = "VHOST_USER_GET_FEATURES",
293 [VHOST_USER_SET_FEATURES] = "VHOST_USER_SET_FEATURES",
294 [VHOST_USER_SET_OWNER] = "VHOST_USER_SET_OWNER",
295 [VHOST_USER_RESET_OWNER] = "VHOST_USER_RESET_OWNER",
296 [VHOST_USER_SET_MEM_TABLE] = "VHOST_USER_SET_MEM_TABLE",
297 [VHOST_USER_SET_LOG_BASE] = "VHOST_USER_SET_LOG_BASE",
298 [VHOST_USER_SET_LOG_FD] = "VHOST_USER_SET_LOG_FD",
299 [VHOST_USER_SET_VRING_NUM] = "VHOST_USER_SET_VRING_NUM",
300 [VHOST_USER_SET_VRING_ADDR] = "VHOST_USER_SET_VRING_ADDR",
301 [VHOST_USER_SET_VRING_BASE] = "VHOST_USER_SET_VRING_BASE",
302 [VHOST_USER_GET_VRING_BASE] = "VHOST_USER_GET_VRING_BASE",
303 [VHOST_USER_SET_VRING_KICK] = "VHOST_USER_SET_VRING_KICK",
304 [VHOST_USER_SET_VRING_CALL] = "VHOST_USER_SET_VRING_CALL",
305 [VHOST_USER_SET_VRING_ERR] = "VHOST_USER_SET_VRING_ERR",
306 [VHOST_USER_GET_PROTOCOL_FEATURES] = "VHOST_USER_GET_PROTOCOL_FEATURES",
307 [VHOST_USER_SET_PROTOCOL_FEATURES] = "VHOST_USER_SET_PROTOCOL_FEATURES",
308 [VHOST_USER_GET_QUEUE_NUM] = "VHOST_USER_GET_QUEUE_NUM",
309 [VHOST_USER_SET_VRING_ENABLE] = "VHOST_USER_SET_VRING_ENABLE",
310 [VHOST_USER_SEND_RARP] = "VHOST_USER_SEND_RARP",
311 [VHOST_USER_MAX] = "VHOST_USER_MAX",
312 };
313
314 static void
315 print_buffer(uint8_t *buf, size_t len)
316 {
317 int i;
318 printf("Raw buffer:\n");
319 for (i = 0; i < len; i++) {
320 if (i % 16 == 0) {
321 printf("\n");
322 }
323 if (i % 4 == 0) {
324 printf(" ");
325 }
326 printf("%02x ", buf[i]);
327 }
328 printf("\n............................................................\n");
329 }
330
331 /* Translate guest physical address to our virtual address. */
332 static uint64_t
333 gpa_to_va(VubrDev *dev, uint64_t guest_addr)
334 {
335 int i;
336
337 /* Find matching memory region. */
338 for (i = 0; i < dev->nregions; i++) {
339 VubrDevRegion *r = &dev->regions[i];
340
341 if ((guest_addr >= r->gpa) && (guest_addr < (r->gpa + r->size))) {
342 return guest_addr - r->gpa + r->mmap_addr + r->mmap_offset;
343 }
344 }
345
346 assert(!"address not found in regions");
347 return 0;
348 }
349
350 /* Translate qemu virtual address to our virtual address. */
351 static uint64_t
352 qva_to_va(VubrDev *dev, uint64_t qemu_addr)
353 {
354 int i;
355
356 /* Find matching memory region. */
357 for (i = 0; i < dev->nregions; i++) {
358 VubrDevRegion *r = &dev->regions[i];
359
360 if ((qemu_addr >= r->qva) && (qemu_addr < (r->qva + r->size))) {
361 return qemu_addr - r->qva + r->mmap_addr + r->mmap_offset;
362 }
363 }
364
365 assert(!"address not found in regions");
366 return 0;
367 }
368
369 static void
370 vubr_message_read(int conn_fd, VhostUserMsg *vmsg)
371 {
372 char control[CMSG_SPACE(VHOST_MEMORY_MAX_NREGIONS * sizeof(int))] = { };
373 struct iovec iov = {
374 .iov_base = (char *)vmsg,
375 .iov_len = VHOST_USER_HDR_SIZE,
376 };
377 struct msghdr msg = {
378 .msg_iov = &iov,
379 .msg_iovlen = 1,
380 .msg_control = control,
381 .msg_controllen = sizeof(control),
382 };
383 size_t fd_size;
384 struct cmsghdr *cmsg;
385 int rc;
386
387 rc = recvmsg(conn_fd, &msg, 0);
388
389 if (rc == 0) {
390 vubr_die("recvmsg");
391 fprintf(stderr, "Peer disconnected.\n");
392 exit(1);
393 }
394 if (rc < 0) {
395 vubr_die("recvmsg");
396 }
397
398 vmsg->fd_num = 0;
399 for (cmsg = CMSG_FIRSTHDR(&msg);
400 cmsg != NULL;
401 cmsg = CMSG_NXTHDR(&msg, cmsg))
402 {
403 if (cmsg->cmsg_level == SOL_SOCKET && cmsg->cmsg_type == SCM_RIGHTS) {
404 fd_size = cmsg->cmsg_len - CMSG_LEN(0);
405 vmsg->fd_num = fd_size / sizeof(int);
406 memcpy(vmsg->fds, CMSG_DATA(cmsg), fd_size);
407 break;
408 }
409 }
410
411 if (vmsg->size > sizeof(vmsg->payload)) {
412 fprintf(stderr,
413 "Error: too big message request: %d, size: vmsg->size: %u, "
414 "while sizeof(vmsg->payload) = %zu\n",
415 vmsg->request, vmsg->size, sizeof(vmsg->payload));
416 exit(1);
417 }
418
419 if (vmsg->size) {
420 rc = read(conn_fd, &vmsg->payload, vmsg->size);
421 if (rc == 0) {
422 vubr_die("recvmsg");
423 fprintf(stderr, "Peer disconnected.\n");
424 exit(1);
425 }
426 if (rc < 0) {
427 vubr_die("recvmsg");
428 }
429
430 assert(rc == vmsg->size);
431 }
432 }
433
434 static void
435 vubr_message_write(int conn_fd, VhostUserMsg *vmsg)
436 {
437 int rc;
438
439 do {
440 rc = write(conn_fd, vmsg, VHOST_USER_HDR_SIZE + vmsg->size);
441 } while (rc < 0 && errno == EINTR);
442
443 if (rc < 0) {
444 vubr_die("write");
445 }
446 }
447
448 static void
449 vubr_backend_udp_sendbuf(VubrDev *dev, uint8_t *buf, size_t len)
450 {
451 int slen = sizeof(struct sockaddr_in);
452
453 if (sendto(dev->backend_udp_sock, buf, len, 0,
454 (struct sockaddr *) &dev->backend_udp_dest, slen) == -1) {
455 vubr_die("sendto()");
456 }
457 }
458
459 static int
460 vubr_backend_udp_recvbuf(VubrDev *dev, uint8_t *buf, size_t buflen)
461 {
462 int slen = sizeof(struct sockaddr_in);
463 int rc;
464
465 rc = recvfrom(dev->backend_udp_sock, buf, buflen, 0,
466 (struct sockaddr *) &dev->backend_udp_dest,
467 (socklen_t *)&slen);
468 if (rc == -1) {
469 vubr_die("recvfrom()");
470 }
471
472 return rc;
473 }
474
475 static void
476 vubr_consume_raw_packet(VubrDev *dev, uint8_t *buf, uint32_t len)
477 {
478 int hdrlen = dev->hdrlen;
479 DPRINT(" hdrlen = %d\n", dev->hdrlen);
480
481 if (VHOST_USER_BRIDGE_DEBUG) {
482 print_buffer(buf, len);
483 }
484 vubr_backend_udp_sendbuf(dev, buf + hdrlen, len - hdrlen);
485 }
486
487 /* Kick the log_call_fd if required. */
488 static void
489 vubr_log_kick(VubrDev *dev)
490 {
491 if (dev->log_call_fd != -1) {
492 DPRINT("Kicking the QEMU's log...\n");
493 eventfd_write(dev->log_call_fd, 1);
494 }
495 }
496
497 /* Kick the guest if necessary. */
498 static void
499 vubr_virtqueue_kick(VubrVirtq *vq)
500 {
501 if (!(vq->avail->flags & VRING_AVAIL_F_NO_INTERRUPT)) {
502 DPRINT("Kicking the guest...\n");
503 eventfd_write(vq->call_fd, 1);
504 }
505 }
506
507 static void
508 vubr_log_page(uint8_t *log_table, uint64_t page)
509 {
510 DPRINT("Logged dirty guest page: %"PRId64"\n", page);
511 atomic_or(&log_table[page / 8], 1 << (page % 8));
512 }
513
514 static void
515 vubr_log_write(VubrDev *dev, uint64_t address, uint64_t length)
516 {
517 uint64_t page;
518
519 if (!(dev->features & (1ULL << VHOST_F_LOG_ALL)) ||
520 !dev->log_table || !length) {
521 return;
522 }
523
524 assert(dev->log_size > ((address + length - 1) / VHOST_LOG_PAGE / 8));
525
526 page = address / VHOST_LOG_PAGE;
527 while (page * VHOST_LOG_PAGE < address + length) {
528 vubr_log_page(dev->log_table, page);
529 page += VHOST_LOG_PAGE;
530 }
531 vubr_log_kick(dev);
532 }
533
534 static void
535 vubr_post_buffer(VubrDev *dev, VubrVirtq *vq, uint8_t *buf, int32_t len)
536 {
537 struct vring_desc *desc = vq->desc;
538 struct vring_avail *avail = vq->avail;
539 struct vring_used *used = vq->used;
540 uint64_t log_guest_addr = vq->log_guest_addr;
541 int32_t remaining_len = len;
542
543 unsigned int size = vq->size;
544
545 uint16_t avail_index = atomic_mb_read(&avail->idx);
546
547 /* We check the available descriptors before posting the
548 * buffer, so here we assume that enough available
549 * descriptors. */
550 assert(vq->last_avail_index != avail_index);
551 uint16_t a_index = vq->last_avail_index % size;
552 uint16_t u_index = vq->last_used_index % size;
553 uint16_t d_index = avail->ring[a_index];
554
555 int i = d_index;
556 uint32_t written_len = 0;
557
558 do {
559 DPRINT("Post packet to guest on vq:\n");
560 DPRINT(" size = %d\n", vq->size);
561 DPRINT(" last_avail_index = %d\n", vq->last_avail_index);
562 DPRINT(" last_used_index = %d\n", vq->last_used_index);
563 DPRINT(" a_index = %d\n", a_index);
564 DPRINT(" u_index = %d\n", u_index);
565 DPRINT(" d_index = %d\n", d_index);
566 DPRINT(" desc[%d].addr = 0x%016"PRIx64"\n", i, desc[i].addr);
567 DPRINT(" desc[%d].len = %d\n", i, desc[i].len);
568 DPRINT(" desc[%d].flags = %d\n", i, desc[i].flags);
569 DPRINT(" avail->idx = %d\n", avail_index);
570 DPRINT(" used->idx = %d\n", used->idx);
571
572 if (!(desc[i].flags & VRING_DESC_F_WRITE)) {
573 /* FIXME: we should find writable descriptor. */
574 fprintf(stderr, "Error: descriptor is not writable. Exiting.\n");
575 exit(1);
576 }
577
578 void *chunk_start = (void *)(uintptr_t)gpa_to_va(dev, desc[i].addr);
579 uint32_t chunk_len = desc[i].len;
580 uint32_t chunk_write_len = MIN(remaining_len, chunk_len);
581
582 memcpy(chunk_start, buf + written_len, chunk_write_len);
583 vubr_log_write(dev, desc[i].addr, chunk_write_len);
584 remaining_len -= chunk_write_len;
585 written_len += chunk_write_len;
586
587 if ((remaining_len == 0) || !(desc[i].flags & VRING_DESC_F_NEXT)) {
588 break;
589 }
590
591 i = desc[i].next;
592 } while (1);
593
594 if (remaining_len > 0) {
595 fprintf(stderr,
596 "Too long packet for RX, remaining_len = %d, Dropping...\n",
597 remaining_len);
598 return;
599 }
600
601 /* Add descriptor to the used ring. */
602 used->ring[u_index].id = d_index;
603 used->ring[u_index].len = len;
604 vubr_log_write(dev,
605 log_guest_addr + offsetof(struct vring_used, ring[u_index]),
606 sizeof(used->ring[u_index]));
607
608 vq->last_avail_index++;
609 vq->last_used_index++;
610
611 atomic_mb_set(&used->idx, vq->last_used_index);
612 vubr_log_write(dev,
613 log_guest_addr + offsetof(struct vring_used, idx),
614 sizeof(used->idx));
615
616 /* Kick the guest if necessary. */
617 vubr_virtqueue_kick(vq);
618 }
619
620 static int
621 vubr_process_desc(VubrDev *dev, VubrVirtq *vq)
622 {
623 struct vring_desc *desc = vq->desc;
624 struct vring_avail *avail = vq->avail;
625 struct vring_used *used = vq->used;
626 uint64_t log_guest_addr = vq->log_guest_addr;
627
628 unsigned int size = vq->size;
629
630 uint16_t a_index = vq->last_avail_index % size;
631 uint16_t u_index = vq->last_used_index % size;
632 uint16_t d_index = avail->ring[a_index];
633
634 uint32_t i, len = 0;
635 size_t buf_size = 4096;
636 uint8_t buf[4096];
637
638 DPRINT("Chunks: ");
639 i = d_index;
640 do {
641 void *chunk_start = (void *)(uintptr_t)gpa_to_va(dev, desc[i].addr);
642 uint32_t chunk_len = desc[i].len;
643
644 assert(!(desc[i].flags & VRING_DESC_F_WRITE));
645
646 if (len + chunk_len < buf_size) {
647 memcpy(buf + len, chunk_start, chunk_len);
648 DPRINT("%d ", chunk_len);
649 } else {
650 fprintf(stderr, "Error: too long packet. Dropping...\n");
651 break;
652 }
653
654 len += chunk_len;
655
656 if (!(desc[i].flags & VRING_DESC_F_NEXT)) {
657 break;
658 }
659
660 i = desc[i].next;
661 } while (1);
662 DPRINT("\n");
663
664 if (!len) {
665 return -1;
666 }
667
668 /* Add descriptor to the used ring. */
669 used->ring[u_index].id = d_index;
670 used->ring[u_index].len = len;
671 vubr_log_write(dev,
672 log_guest_addr + offsetof(struct vring_used, ring[u_index]),
673 sizeof(used->ring[u_index]));
674
675 vubr_consume_raw_packet(dev, buf, len);
676
677 return 0;
678 }
679
680 static void
681 vubr_process_avail(VubrDev *dev, VubrVirtq *vq)
682 {
683 struct vring_avail *avail = vq->avail;
684 struct vring_used *used = vq->used;
685 uint64_t log_guest_addr = vq->log_guest_addr;
686
687 while (vq->last_avail_index != atomic_mb_read(&avail->idx)) {
688 vubr_process_desc(dev, vq);
689 vq->last_avail_index++;
690 vq->last_used_index++;
691 }
692
693 atomic_mb_set(&used->idx, vq->last_used_index);
694 vubr_log_write(dev,
695 log_guest_addr + offsetof(struct vring_used, idx),
696 sizeof(used->idx));
697 }
698
699 static void
700 vubr_backend_recv_cb(int sock, void *ctx)
701 {
702 VubrDev *dev = (VubrDev *) ctx;
703 VubrVirtq *rx_vq = &dev->vq[0];
704 uint8_t buf[4096];
705 struct virtio_net_hdr_v1 *hdr = (struct virtio_net_hdr_v1 *)buf;
706 int hdrlen = dev->hdrlen;
707 int buflen = sizeof(buf);
708 int len;
709
710 if (!dev->ready) {
711 return;
712 }
713
714 DPRINT("\n\n *** IN UDP RECEIVE CALLBACK ***\n\n");
715 DPRINT(" hdrlen = %d\n", hdrlen);
716
717 uint16_t avail_index = atomic_mb_read(&rx_vq->avail->idx);
718
719 /* If there is no available descriptors, just do nothing.
720 * The buffer will be handled by next arrived UDP packet,
721 * or next kick on receive virtq. */
722 if (rx_vq->last_avail_index == avail_index) {
723 DPRINT("Got UDP packet, but no available descriptors on RX virtq.\n");
724 return;
725 }
726
727 memset(buf, 0, hdrlen);
728 /* TODO: support mergeable buffers. */
729 if (hdrlen == 12)
730 hdr->num_buffers = 1;
731 len = vubr_backend_udp_recvbuf(dev, buf + hdrlen, buflen - hdrlen);
732
733 vubr_post_buffer(dev, rx_vq, buf, len + hdrlen);
734 }
735
736 static void
737 vubr_kick_cb(int sock, void *ctx)
738 {
739 VubrDev *dev = (VubrDev *) ctx;
740 eventfd_t kick_data;
741 ssize_t rc;
742
743 rc = eventfd_read(sock, &kick_data);
744 if (rc == -1) {
745 vubr_die("eventfd_read()");
746 } else {
747 DPRINT("Got kick_data: %016"PRIx64"\n", kick_data);
748 vubr_process_avail(dev, &dev->vq[1]);
749 }
750 }
751
752 static int
753 vubr_none_exec(VubrDev *dev, VhostUserMsg *vmsg)
754 {
755 DPRINT("Function %s() not implemented yet.\n", __func__);
756 return 0;
757 }
758
759 static int
760 vubr_get_features_exec(VubrDev *dev, VhostUserMsg *vmsg)
761 {
762 vmsg->payload.u64 =
763 ((1ULL << VIRTIO_NET_F_MRG_RXBUF) |
764 (1ULL << VHOST_F_LOG_ALL) |
765 (1ULL << VIRTIO_NET_F_GUEST_ANNOUNCE) |
766 (1ULL << VHOST_USER_F_PROTOCOL_FEATURES));
767
768 vmsg->size = sizeof(vmsg->payload.u64);
769
770 DPRINT("Sending back to guest u64: 0x%016"PRIx64"\n", vmsg->payload.u64);
771
772 /* Reply */
773 return 1;
774 }
775
776 static int
777 vubr_set_features_exec(VubrDev *dev, VhostUserMsg *vmsg)
778 {
779 DPRINT("u64: 0x%016"PRIx64"\n", vmsg->payload.u64);
780
781 dev->features = vmsg->payload.u64;
782 if ((dev->features & (1ULL << VIRTIO_F_VERSION_1)) ||
783 (dev->features & (1ULL << VIRTIO_NET_F_MRG_RXBUF))) {
784 dev->hdrlen = 12;
785 } else {
786 dev->hdrlen = 10;
787 }
788
789 return 0;
790 }
791
792 static int
793 vubr_set_owner_exec(VubrDev *dev, VhostUserMsg *vmsg)
794 {
795 return 0;
796 }
797
798 static void
799 vubr_close_log(VubrDev *dev)
800 {
801 if (dev->log_table) {
802 if (munmap(dev->log_table, dev->log_size) != 0) {
803 vubr_die("munmap()");
804 }
805
806 dev->log_table = 0;
807 }
808 if (dev->log_call_fd != -1) {
809 close(dev->log_call_fd);
810 dev->log_call_fd = -1;
811 }
812 }
813
814 static int
815 vubr_reset_device_exec(VubrDev *dev, VhostUserMsg *vmsg)
816 {
817 vubr_close_log(dev);
818 dev->ready = 0;
819 dev->features = 0;
820 return 0;
821 }
822
823 static int
824 vubr_set_mem_table_exec(VubrDev *dev, VhostUserMsg *vmsg)
825 {
826 int i;
827 VhostUserMemory *memory = &vmsg->payload.memory;
828 dev->nregions = memory->nregions;
829
830 DPRINT("Nregions: %d\n", memory->nregions);
831 for (i = 0; i < dev->nregions; i++) {
832 void *mmap_addr;
833 VhostUserMemoryRegion *msg_region = &memory->regions[i];
834 VubrDevRegion *dev_region = &dev->regions[i];
835
836 DPRINT("Region %d\n", i);
837 DPRINT(" guest_phys_addr: 0x%016"PRIx64"\n",
838 msg_region->guest_phys_addr);
839 DPRINT(" memory_size: 0x%016"PRIx64"\n",
840 msg_region->memory_size);
841 DPRINT(" userspace_addr 0x%016"PRIx64"\n",
842 msg_region->userspace_addr);
843 DPRINT(" mmap_offset 0x%016"PRIx64"\n",
844 msg_region->mmap_offset);
845
846 dev_region->gpa = msg_region->guest_phys_addr;
847 dev_region->size = msg_region->memory_size;
848 dev_region->qva = msg_region->userspace_addr;
849 dev_region->mmap_offset = msg_region->mmap_offset;
850
851 /* We don't use offset argument of mmap() since the
852 * mapped address has to be page aligned, and we use huge
853 * pages. */
854 mmap_addr = mmap(0, dev_region->size + dev_region->mmap_offset,
855 PROT_READ | PROT_WRITE, MAP_SHARED,
856 vmsg->fds[i], 0);
857
858 if (mmap_addr == MAP_FAILED) {
859 vubr_die("mmap");
860 }
861 dev_region->mmap_addr = (uint64_t)(uintptr_t)mmap_addr;
862 DPRINT(" mmap_addr: 0x%016"PRIx64"\n", dev_region->mmap_addr);
863
864 close(vmsg->fds[i]);
865 }
866
867 return 0;
868 }
869
870 static int
871 vubr_set_log_base_exec(VubrDev *dev, VhostUserMsg *vmsg)
872 {
873 int fd;
874 uint64_t log_mmap_size, log_mmap_offset;
875 void *rc;
876
877 assert(vmsg->fd_num == 1);
878 fd = vmsg->fds[0];
879
880 assert(vmsg->size == sizeof(vmsg->payload.log));
881 log_mmap_offset = vmsg->payload.log.mmap_offset;
882 log_mmap_size = vmsg->payload.log.mmap_size;
883 DPRINT("Log mmap_offset: %"PRId64"\n", log_mmap_offset);
884 DPRINT("Log mmap_size: %"PRId64"\n", log_mmap_size);
885
886 rc = mmap(0, log_mmap_size, PROT_READ | PROT_WRITE, MAP_SHARED, fd,
887 log_mmap_offset);
888 if (rc == MAP_FAILED) {
889 vubr_die("mmap");
890 }
891 dev->log_table = rc;
892 dev->log_size = log_mmap_size;
893
894 vmsg->size = sizeof(vmsg->payload.u64);
895 /* Reply */
896 return 1;
897 }
898
899 static int
900 vubr_set_log_fd_exec(VubrDev *dev, VhostUserMsg *vmsg)
901 {
902 assert(vmsg->fd_num == 1);
903 dev->log_call_fd = vmsg->fds[0];
904 DPRINT("Got log_call_fd: %d\n", vmsg->fds[0]);
905 return 0;
906 }
907
908 static int
909 vubr_set_vring_num_exec(VubrDev *dev, VhostUserMsg *vmsg)
910 {
911 unsigned int index = vmsg->payload.state.index;
912 unsigned int num = vmsg->payload.state.num;
913
914 DPRINT("State.index: %d\n", index);
915 DPRINT("State.num: %d\n", num);
916 dev->vq[index].size = num;
917 return 0;
918 }
919
920 static int
921 vubr_set_vring_addr_exec(VubrDev *dev, VhostUserMsg *vmsg)
922 {
923 struct vhost_vring_addr *vra = &vmsg->payload.addr;
924 unsigned int index = vra->index;
925 VubrVirtq *vq = &dev->vq[index];
926
927 DPRINT("vhost_vring_addr:\n");
928 DPRINT(" index: %d\n", vra->index);
929 DPRINT(" flags: %d\n", vra->flags);
930 DPRINT(" desc_user_addr: 0x%016llx\n", vra->desc_user_addr);
931 DPRINT(" used_user_addr: 0x%016llx\n", vra->used_user_addr);
932 DPRINT(" avail_user_addr: 0x%016llx\n", vra->avail_user_addr);
933 DPRINT(" log_guest_addr: 0x%016llx\n", vra->log_guest_addr);
934
935 vq->desc = (struct vring_desc *)(uintptr_t)qva_to_va(dev, vra->desc_user_addr);
936 vq->used = (struct vring_used *)(uintptr_t)qva_to_va(dev, vra->used_user_addr);
937 vq->avail = (struct vring_avail *)(uintptr_t)qva_to_va(dev, vra->avail_user_addr);
938 vq->log_guest_addr = vra->log_guest_addr;
939
940 DPRINT("Setting virtq addresses:\n");
941 DPRINT(" vring_desc at %p\n", vq->desc);
942 DPRINT(" vring_used at %p\n", vq->used);
943 DPRINT(" vring_avail at %p\n", vq->avail);
944
945 vq->last_used_index = vq->used->idx;
946
947 if (vq->last_avail_index != vq->used->idx) {
948 DPRINT("Last avail index != used index: %d != %d, resuming",
949 vq->last_avail_index, vq->used->idx);
950 vq->last_avail_index = vq->used->idx;
951 }
952
953 return 0;
954 }
955
956 static int
957 vubr_set_vring_base_exec(VubrDev *dev, VhostUserMsg *vmsg)
958 {
959 unsigned int index = vmsg->payload.state.index;
960 unsigned int num = vmsg->payload.state.num;
961
962 DPRINT("State.index: %d\n", index);
963 DPRINT("State.num: %d\n", num);
964 dev->vq[index].last_avail_index = num;
965
966 return 0;
967 }
968
969 static int
970 vubr_get_vring_base_exec(VubrDev *dev, VhostUserMsg *vmsg)
971 {
972 unsigned int index = vmsg->payload.state.index;
973
974 DPRINT("State.index: %d\n", index);
975 vmsg->payload.state.num = dev->vq[index].last_avail_index;
976 vmsg->size = sizeof(vmsg->payload.state);
977 /* FIXME: this is a work-around for a bug in QEMU enabling
978 * too early vrings. When protocol features are enabled,
979 * we have to respect * VHOST_USER_SET_VRING_ENABLE request. */
980 dev->ready = 0;
981
982 if (dev->vq[index].call_fd != -1) {
983 close(dev->vq[index].call_fd);
984 dispatcher_remove(&dev->dispatcher, dev->vq[index].call_fd);
985 dev->vq[index].call_fd = -1;
986 }
987 if (dev->vq[index].kick_fd != -1) {
988 close(dev->vq[index].kick_fd);
989 dispatcher_remove(&dev->dispatcher, dev->vq[index].kick_fd);
990 dev->vq[index].kick_fd = -1;
991 }
992
993 /* Reply */
994 return 1;
995 }
996
997 static int
998 vubr_set_vring_kick_exec(VubrDev *dev, VhostUserMsg *vmsg)
999 {
1000 uint64_t u64_arg = vmsg->payload.u64;
1001 int index = u64_arg & VHOST_USER_VRING_IDX_MASK;
1002
1003 DPRINT("u64: 0x%016"PRIx64"\n", vmsg->payload.u64);
1004
1005 assert((u64_arg & VHOST_USER_VRING_NOFD_MASK) == 0);
1006 assert(vmsg->fd_num == 1);
1007
1008 if (dev->vq[index].kick_fd != -1) {
1009 close(dev->vq[index].kick_fd);
1010 dispatcher_remove(&dev->dispatcher, dev->vq[index].kick_fd);
1011 }
1012 dev->vq[index].kick_fd = vmsg->fds[0];
1013 DPRINT("Got kick_fd: %d for vq: %d\n", vmsg->fds[0], index);
1014
1015 if (index % 2 == 1) {
1016 /* TX queue. */
1017 dispatcher_add(&dev->dispatcher, dev->vq[index].kick_fd,
1018 dev, vubr_kick_cb);
1019
1020 DPRINT("Waiting for kicks on fd: %d for vq: %d\n",
1021 dev->vq[index].kick_fd, index);
1022 }
1023 /* We temporarily use this hack to determine that both TX and RX
1024 * queues are set up and ready for processing.
1025 * FIXME: we need to rely in VHOST_USER_SET_VRING_ENABLE and
1026 * actual kicks. */
1027 if (dev->vq[0].kick_fd != -1 &&
1028 dev->vq[1].kick_fd != -1) {
1029 dev->ready = 1;
1030 DPRINT("vhost-user-bridge is ready for processing queues.\n");
1031 }
1032 return 0;
1033
1034 }
1035
1036 static int
1037 vubr_set_vring_call_exec(VubrDev *dev, VhostUserMsg *vmsg)
1038 {
1039 uint64_t u64_arg = vmsg->payload.u64;
1040 int index = u64_arg & VHOST_USER_VRING_IDX_MASK;
1041
1042 DPRINT("u64: 0x%016"PRIx64"\n", vmsg->payload.u64);
1043 assert((u64_arg & VHOST_USER_VRING_NOFD_MASK) == 0);
1044 assert(vmsg->fd_num == 1);
1045
1046 if (dev->vq[index].call_fd != -1) {
1047 close(dev->vq[index].call_fd);
1048 dispatcher_remove(&dev->dispatcher, dev->vq[index].call_fd);
1049 }
1050 dev->vq[index].call_fd = vmsg->fds[0];
1051 DPRINT("Got call_fd: %d for vq: %d\n", vmsg->fds[0], index);
1052
1053 return 0;
1054 }
1055
1056 static int
1057 vubr_set_vring_err_exec(VubrDev *dev, VhostUserMsg *vmsg)
1058 {
1059 DPRINT("u64: 0x%016"PRIx64"\n", vmsg->payload.u64);
1060 return 0;
1061 }
1062
1063 static int
1064 vubr_get_protocol_features_exec(VubrDev *dev, VhostUserMsg *vmsg)
1065 {
1066 vmsg->payload.u64 = 1ULL << VHOST_USER_PROTOCOL_F_LOG_SHMFD;
1067 DPRINT("u64: 0x%016"PRIx64"\n", vmsg->payload.u64);
1068 vmsg->size = sizeof(vmsg->payload.u64);
1069
1070 /* Reply */
1071 return 1;
1072 }
1073
1074 static int
1075 vubr_set_protocol_features_exec(VubrDev *dev, VhostUserMsg *vmsg)
1076 {
1077 /* FIXME: unimplented */
1078 DPRINT("u64: 0x%016"PRIx64"\n", vmsg->payload.u64);
1079 return 0;
1080 }
1081
1082 static int
1083 vubr_get_queue_num_exec(VubrDev *dev, VhostUserMsg *vmsg)
1084 {
1085 DPRINT("Function %s() not implemented yet.\n", __func__);
1086 return 0;
1087 }
1088
1089 static int
1090 vubr_set_vring_enable_exec(VubrDev *dev, VhostUserMsg *vmsg)
1091 {
1092 unsigned int index = vmsg->payload.state.index;
1093 unsigned int enable = vmsg->payload.state.num;
1094
1095 DPRINT("State.index: %d\n", index);
1096 DPRINT("State.enable: %d\n", enable);
1097 dev->vq[index].enable = enable;
1098 return 0;
1099 }
1100
1101 static int
1102 vubr_send_rarp_exec(VubrDev *dev, VhostUserMsg *vmsg)
1103 {
1104 DPRINT("Function %s() not implemented yet.\n", __func__);
1105 return 0;
1106 }
1107
1108 static int
1109 vubr_execute_request(VubrDev *dev, VhostUserMsg *vmsg)
1110 {
1111 /* Print out generic part of the request. */
1112 DPRINT(
1113 "================== Vhost user message from QEMU ==================\n");
1114 DPRINT("Request: %s (%d)\n", vubr_request_str[vmsg->request],
1115 vmsg->request);
1116 DPRINT("Flags: 0x%x\n", vmsg->flags);
1117 DPRINT("Size: %d\n", vmsg->size);
1118
1119 if (vmsg->fd_num) {
1120 int i;
1121 DPRINT("Fds:");
1122 for (i = 0; i < vmsg->fd_num; i++) {
1123 DPRINT(" %d", vmsg->fds[i]);
1124 }
1125 DPRINT("\n");
1126 }
1127
1128 switch (vmsg->request) {
1129 case VHOST_USER_NONE:
1130 return vubr_none_exec(dev, vmsg);
1131 case VHOST_USER_GET_FEATURES:
1132 return vubr_get_features_exec(dev, vmsg);
1133 case VHOST_USER_SET_FEATURES:
1134 return vubr_set_features_exec(dev, vmsg);
1135 case VHOST_USER_SET_OWNER:
1136 return vubr_set_owner_exec(dev, vmsg);
1137 case VHOST_USER_RESET_OWNER:
1138 return vubr_reset_device_exec(dev, vmsg);
1139 case VHOST_USER_SET_MEM_TABLE:
1140 return vubr_set_mem_table_exec(dev, vmsg);
1141 case VHOST_USER_SET_LOG_BASE:
1142 return vubr_set_log_base_exec(dev, vmsg);
1143 case VHOST_USER_SET_LOG_FD:
1144 return vubr_set_log_fd_exec(dev, vmsg);
1145 case VHOST_USER_SET_VRING_NUM:
1146 return vubr_set_vring_num_exec(dev, vmsg);
1147 case VHOST_USER_SET_VRING_ADDR:
1148 return vubr_set_vring_addr_exec(dev, vmsg);
1149 case VHOST_USER_SET_VRING_BASE:
1150 return vubr_set_vring_base_exec(dev, vmsg);
1151 case VHOST_USER_GET_VRING_BASE:
1152 return vubr_get_vring_base_exec(dev, vmsg);
1153 case VHOST_USER_SET_VRING_KICK:
1154 return vubr_set_vring_kick_exec(dev, vmsg);
1155 case VHOST_USER_SET_VRING_CALL:
1156 return vubr_set_vring_call_exec(dev, vmsg);
1157 case VHOST_USER_SET_VRING_ERR:
1158 return vubr_set_vring_err_exec(dev, vmsg);
1159 case VHOST_USER_GET_PROTOCOL_FEATURES:
1160 return vubr_get_protocol_features_exec(dev, vmsg);
1161 case VHOST_USER_SET_PROTOCOL_FEATURES:
1162 return vubr_set_protocol_features_exec(dev, vmsg);
1163 case VHOST_USER_GET_QUEUE_NUM:
1164 return vubr_get_queue_num_exec(dev, vmsg);
1165 case VHOST_USER_SET_VRING_ENABLE:
1166 return vubr_set_vring_enable_exec(dev, vmsg);
1167 case VHOST_USER_SEND_RARP:
1168 return vubr_send_rarp_exec(dev, vmsg);
1169
1170 case VHOST_USER_MAX:
1171 assert(vmsg->request != VHOST_USER_MAX);
1172 }
1173 return 0;
1174 }
1175
1176 static void
1177 vubr_receive_cb(int sock, void *ctx)
1178 {
1179 VubrDev *dev = (VubrDev *) ctx;
1180 VhostUserMsg vmsg;
1181 int reply_requested;
1182
1183 vubr_message_read(sock, &vmsg);
1184 reply_requested = vubr_execute_request(dev, &vmsg);
1185 if (reply_requested) {
1186 /* Set the version in the flags when sending the reply */
1187 vmsg.flags &= ~VHOST_USER_VERSION_MASK;
1188 vmsg.flags |= VHOST_USER_VERSION;
1189 vmsg.flags |= VHOST_USER_REPLY_MASK;
1190 vubr_message_write(sock, &vmsg);
1191 }
1192 }
1193
1194 static void
1195 vubr_accept_cb(int sock, void *ctx)
1196 {
1197 VubrDev *dev = (VubrDev *)ctx;
1198 int conn_fd;
1199 struct sockaddr_un un;
1200 socklen_t len = sizeof(un);
1201
1202 conn_fd = accept(sock, (struct sockaddr *) &un, &len);
1203 if (conn_fd == -1) {
1204 vubr_die("accept()");
1205 }
1206 DPRINT("Got connection from remote peer on sock %d\n", conn_fd);
1207 dispatcher_add(&dev->dispatcher, conn_fd, ctx, vubr_receive_cb);
1208 }
1209
1210 static VubrDev *
1211 vubr_new(const char *path, bool client)
1212 {
1213 VubrDev *dev = (VubrDev *) calloc(1, sizeof(VubrDev));
1214 dev->nregions = 0;
1215 int i;
1216 struct sockaddr_un un;
1217 CallbackFunc cb;
1218 size_t len;
1219
1220 for (i = 0; i < MAX_NR_VIRTQUEUE; i++) {
1221 dev->vq[i] = (VubrVirtq) {
1222 .call_fd = -1, .kick_fd = -1,
1223 .size = 0,
1224 .last_avail_index = 0, .last_used_index = 0,
1225 .desc = 0, .avail = 0, .used = 0,
1226 .enable = 0,
1227 };
1228 }
1229
1230 /* Init log */
1231 dev->log_call_fd = -1;
1232 dev->log_size = 0;
1233 dev->log_table = 0;
1234 dev->ready = 0;
1235 dev->features = 0;
1236
1237 /* Get a UNIX socket. */
1238 dev->sock = socket(AF_UNIX, SOCK_STREAM, 0);
1239 if (dev->sock == -1) {
1240 vubr_die("socket");
1241 }
1242
1243 un.sun_family = AF_UNIX;
1244 strcpy(un.sun_path, path);
1245 len = sizeof(un.sun_family) + strlen(path);
1246
1247 if (!client) {
1248 unlink(path);
1249
1250 if (bind(dev->sock, (struct sockaddr *) &un, len) == -1) {
1251 vubr_die("bind");
1252 }
1253
1254 if (listen(dev->sock, 1) == -1) {
1255 vubr_die("listen");
1256 }
1257 cb = vubr_accept_cb;
1258
1259 DPRINT("Waiting for connections on UNIX socket %s ...\n", path);
1260 } else {
1261 if (connect(dev->sock, (struct sockaddr *)&un, len) == -1) {
1262 vubr_die("connect");
1263 }
1264 cb = vubr_receive_cb;
1265 }
1266
1267 dispatcher_init(&dev->dispatcher);
1268 dispatcher_add(&dev->dispatcher, dev->sock, (void *)dev, cb);
1269
1270 return dev;
1271 }
1272
1273 static void
1274 vubr_set_host(struct sockaddr_in *saddr, const char *host)
1275 {
1276 if (isdigit(host[0])) {
1277 if (!inet_aton(host, &saddr->sin_addr)) {
1278 fprintf(stderr, "inet_aton() failed.\n");
1279 exit(1);
1280 }
1281 } else {
1282 struct hostent *he = gethostbyname(host);
1283
1284 if (!he) {
1285 fprintf(stderr, "gethostbyname() failed.\n");
1286 exit(1);
1287 }
1288 saddr->sin_addr = *(struct in_addr *)he->h_addr;
1289 }
1290 }
1291
1292 static void
1293 vubr_backend_udp_setup(VubrDev *dev,
1294 const char *local_host,
1295 const char *local_port,
1296 const char *remote_host,
1297 const char *remote_port)
1298 {
1299 int sock;
1300 const char *r;
1301
1302 int lport, rport;
1303
1304 lport = strtol(local_port, (char **)&r, 0);
1305 if (r == local_port) {
1306 fprintf(stderr, "lport parsing failed.\n");
1307 exit(1);
1308 }
1309
1310 rport = strtol(remote_port, (char **)&r, 0);
1311 if (r == remote_port) {
1312 fprintf(stderr, "rport parsing failed.\n");
1313 exit(1);
1314 }
1315
1316 struct sockaddr_in si_local = {
1317 .sin_family = AF_INET,
1318 .sin_port = htons(lport),
1319 };
1320
1321 vubr_set_host(&si_local, local_host);
1322
1323 /* setup destination for sends */
1324 dev->backend_udp_dest = (struct sockaddr_in) {
1325 .sin_family = AF_INET,
1326 .sin_port = htons(rport),
1327 };
1328 vubr_set_host(&dev->backend_udp_dest, remote_host);
1329
1330 sock = socket(AF_INET, SOCK_DGRAM, IPPROTO_UDP);
1331 if (sock == -1) {
1332 vubr_die("socket");
1333 }
1334
1335 if (bind(sock, (struct sockaddr *)&si_local, sizeof(si_local)) == -1) {
1336 vubr_die("bind");
1337 }
1338
1339 dev->backend_udp_sock = sock;
1340 dispatcher_add(&dev->dispatcher, sock, dev, vubr_backend_recv_cb);
1341 DPRINT("Waiting for data from udp backend on %s:%d...\n",
1342 local_host, lport);
1343 }
1344
1345 static void
1346 vubr_run(VubrDev *dev)
1347 {
1348 while (1) {
1349 /* timeout 200ms */
1350 dispatcher_wait(&dev->dispatcher, 200000);
1351 /* Here one can try polling strategy. */
1352 }
1353 }
1354
1355 static int
1356 vubr_parse_host_port(const char **host, const char **port, const char *buf)
1357 {
1358 char *p = strchr(buf, ':');
1359
1360 if (!p) {
1361 return -1;
1362 }
1363 *p = '\0';
1364 *host = strdup(buf);
1365 *port = strdup(p + 1);
1366 return 0;
1367 }
1368
1369 #define DEFAULT_UD_SOCKET "/tmp/vubr.sock"
1370 #define DEFAULT_LHOST "127.0.0.1"
1371 #define DEFAULT_LPORT "4444"
1372 #define DEFAULT_RHOST "127.0.0.1"
1373 #define DEFAULT_RPORT "5555"
1374
1375 static const char *ud_socket_path = DEFAULT_UD_SOCKET;
1376 static const char *lhost = DEFAULT_LHOST;
1377 static const char *lport = DEFAULT_LPORT;
1378 static const char *rhost = DEFAULT_RHOST;
1379 static const char *rport = DEFAULT_RPORT;
1380
1381 int
1382 main(int argc, char *argv[])
1383 {
1384 VubrDev *dev;
1385 int opt;
1386 bool client = false;
1387
1388 while ((opt = getopt(argc, argv, "l:r:u:c")) != -1) {
1389
1390 switch (opt) {
1391 case 'l':
1392 if (vubr_parse_host_port(&lhost, &lport, optarg) < 0) {
1393 goto out;
1394 }
1395 break;
1396 case 'r':
1397 if (vubr_parse_host_port(&rhost, &rport, optarg) < 0) {
1398 goto out;
1399 }
1400 break;
1401 case 'u':
1402 ud_socket_path = strdup(optarg);
1403 break;
1404 case 'c':
1405 client = true;
1406 break;
1407 default:
1408 goto out;
1409 }
1410 }
1411
1412 DPRINT("ud socket: %s (%s)\n", ud_socket_path,
1413 client ? "client" : "server");
1414 DPRINT("local: %s:%s\n", lhost, lport);
1415 DPRINT("remote: %s:%s\n", rhost, rport);
1416
1417 dev = vubr_new(ud_socket_path, client);
1418 if (!dev) {
1419 return 1;
1420 }
1421
1422 vubr_backend_udp_setup(dev, lhost, lport, rhost, rport);
1423 vubr_run(dev);
1424 return 0;
1425
1426 out:
1427 fprintf(stderr, "Usage: %s ", argv[0]);
1428 fprintf(stderr, "[-c] [-u ud_socket_path] [-l lhost:lport] [-r rhost:rport]\n");
1429 fprintf(stderr, "\t-u path to unix doman socket. default: %s\n",
1430 DEFAULT_UD_SOCKET);
1431 fprintf(stderr, "\t-l local host and port. default: %s:%s\n",
1432 DEFAULT_LHOST, DEFAULT_LPORT);
1433 fprintf(stderr, "\t-r remote host and port. default: %s:%s\n",
1434 DEFAULT_RHOST, DEFAULT_RPORT);
1435 fprintf(stderr, "\t-c client mode\n");
1436
1437 return 1;
1438 }