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