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