[7/9] tap: make tap peer work on dedicated data-plane thread

diff --git a/hw/dataplane/virtio-net.c b/hw/dataplane/virtio-net.c
index 9a1795d..c3aac08 100644
--- a/hw/dataplane/virtio-net.c
+++ b/hw/dataplane/virtio-net.c
@@ -246,6 +246,7 @@  static void rx_cb(EventHandler *handler, uint32_t events)
 {
     VirtIONet *n = handler->opaque;
 
+    tap_enable_read(n->nic->nc.peer);
     event_notifier_test_and_clear(handler->notifier);
     qemu_flush_queued_packets(&n->nic->nc);
 }
@@ -388,6 +389,7 @@  void virtnet_dataplane_start(VirtIONet *n)
 
     event_poll_add(&t->polltbl, n->dp->rx_handler, rx_e, rx_cb);
     event_poll_add(&t->polltbl, n->dp->tx_handler, tx_e, tx_cb);
+    tap_create_epoll_event(n->nic->nc.peer);
 }
 
 void virtnet_dataplane_stop(VirtIONet *n)
@@ -398,6 +400,7 @@  void virtnet_dataplane_stop(VirtIONet *n)
 
     event_poll_del_fd(&t->polltbl, event_notifier_get_fd(rx_e));
     event_poll_del_fd(&t->polltbl, event_notifier_get_fd(tx_e));
+    tap_del_epoll_event(n->nic->nc.peer);
 
     t->state = THREAD_EXIT;
     event_notifier_set(&t->e);
@@ -420,3 +423,22 @@  void virtnet_dataplane_destroy(VirtIONet *n)
     g_free(n->dp->tx_handler);
     g_free(n->dp);
 }
+
+void thread_add_event(WorkThread *t, uint32_t event_type, EventHandler *handler)
+{
+    event_poll_add_fd(&t->polltbl, handler->fd, event_type, handler);
+    event_notifier_set(&t->e);
+}
+
+void thread_del_event(WorkThread *t, EventHandler *handler)
+{
+    event_poll_del_fd(&t->polltbl, handler->fd);
+    event_notifier_set(&t->e);
+}
+
+void thread_modify_event(WorkThread *t, uint32_t event_type,
+    EventHandler *handler)
+{
+    event_poll_modify_fd(&t->polltbl, handler->fd, event_type, handler);
+    event_notifier_set(&t->e);
+}
diff --git a/hw/dataplane/virtio-net.h b/hw/dataplane/virtio-net.h
index e50b2de..5d6bd34 100644
--- a/hw/dataplane/virtio-net.h
+++ b/hw/dataplane/virtio-net.h
@@ -23,4 +23,10 @@  typedef struct WorkThread {
 } WorkThread;
 
 extern WorkThread virt_net_thread;
+
+void thread_add_event(WorkThread *t, uint32_t event_type,
+    EventHandler *handler);
+void thread_del_event(WorkThread *t, EventHandler *handler);
+void thread_modify_event(WorkThread *t, uint32_t event_type,
+    EventHandler *handler);
 #endif
diff --git a/include/net/tap.h b/include/net/tap.h
index bb7efb5..3f78d15 100644
--- a/include/net/tap.h
+++ b/include/net/tap.h
@@ -29,6 +29,12 @@ 
 #include "qemu-common.h"
 #include "qapi-types.h"
 
+#ifdef CONFIG_VIRTIO_NET_DATA_PLANE
+int tap_create_epoll_event(NetClientState *nc);
+int tap_del_epoll_event(NetClientState *nc);
+void tap_enable_read(NetClientState *nc);
+#endif
+
 int tap_has_ufo(NetClientState *nc);
 int tap_has_vnet_hdr(NetClientState *nc);
 int tap_has_vnet_hdr_len(NetClientState *nc, int len);
diff --git a/net/tap.c b/net/tap.c
index eb40c42..b6bfed4 100644
--- a/net/tap.c
+++ b/net/tap.c
@@ -27,6 +27,7 @@ 
 
 #include "config-host.h"
 
+#include <sys/epoll.h>
 #include <sys/ioctl.h>
 #include <sys/stat.h>
 #include <sys/wait.h>
@@ -43,6 +44,7 @@ 
 #include "net/tap.h"
 
 #include "hw/vhost_net.h"
+#include "hw/dataplane/virtio-net.h"
 
 /* Maximum GSO packet size (64k) plus plenty of room for
  * the ethernet and virtio_net headers
@@ -61,14 +63,22 @@  typedef struct TAPState {
     unsigned int has_ufo: 1;
     VHostNetState *vhost_net;
     unsigned host_vnet_hdr_len;
+
+#ifdef CONFIG_VIRTIO_NET_DATA_PLANE
+    EventHandler tap_evt;
+    uint32_t tap_events;
+    bool start;
+#endif
 } TAPState;
 
 static int launch_script(const char *setup_script, const char *ifname, int fd);
-
+#ifndef CONFIG_VIRTIO_NET_DATA_PLANE
 static int tap_can_send(void *opaque);
+#endif
 static void tap_send(void *opaque);
 static void tap_writable(void *opaque);
 
+#ifndef CONFIG_VIRTIO_NET_DATA_PLANE
 static void tap_update_fd_handler(TAPState *s)
 {
     qemu_set_fd_handler2(s->fd,
@@ -77,18 +87,96 @@  static void tap_update_fd_handler(TAPState *s)
                          s->write_poll ? tap_writable : NULL,
                          s);
 }
+#endif
 
 static void tap_read_poll(TAPState *s, int enable)
 {
+#ifdef CONFIG_VIRTIO_NET_DATA_PLANE
+    if (!s->start) {
+        return;
+    }
+    if (s->read_poll == !!enable) {
+        return;
+    }
+
+    s->read_poll = !!enable;
+    if (enable) {
+        s->tap_events |= EPOLLIN;
+    } else {
+        s->tap_events &= ~EPOLLIN;
+    }
+    thread_modify_event(&virt_net_thread, s->tap_events, &s->tap_evt);
+#else
     s->read_poll = !!enable;
     tap_update_fd_handler(s);
+#endif
 }
 
 static void tap_write_poll(TAPState *s, int enable)
 {
+#ifdef CONFIG_VIRTIO_NET_DATA_PLANE
+    if (!s->start) {
+        return;
+    }
+    if (s->write_poll == !!enable) {
+        return;
+    }
+    s->write_poll = !!enable;
+    if (enable) {
+        s->tap_events |= EPOLLOUT;
+    } else {
+        s->tap_events &= ~EPOLLOUT;
+    }
+    thread_modify_event(&virt_net_thread, s->tap_events, &s->tap_evt);
+#else
     s->write_poll = !!enable;
     tap_update_fd_handler(s);
+#endif
+}
+
+#ifdef CONFIG_VIRTIO_NET_DATA_PLANE
+static void tap_epoll_handle(EventHandler *handler, uint32_t events)
+{
+    if (events & EPOLLIN) {
+        tap_send(handler->opaque);
+    }
+    if (events & EPOLLOUT) {
+        tap_writable(handler->opaque);
+    }
+}
+
+int tap_create_epoll_event(NetClientState *nc)
+{
+    TAPState *s = DO_UPCAST(TAPState, nc, nc);
+
+    s->tap_evt.callback = tap_epoll_handle;
+    s->tap_evt.opaque = s;
+    s->tap_evt.notifier = NULL;
+    s->tap_evt.fd = s->fd;
+
+    thread_add_event(&virt_net_thread, EPOLLIN|EPOLLOUT, &s->tap_evt);
+    s->start = true;
+    return 0;
+}
+
+int tap_del_epoll_event(NetClientState *nc)
+{
+    TAPState *s = DO_UPCAST(TAPState, nc, nc);
+
+    thread_del_event(&virt_net_thread, &s->tap_evt);
+    s->start = false;
+    return 0;
+}
+
+void tap_enable_read(NetClientState *nc)
+{
+    TAPState *s = DO_UPCAST(TAPState, nc, nc);
+
+    if (s->start) {
+        tap_read_poll(s, 1);
+    }
 }
+#endif
 
 static void tap_writable(void *opaque)
 {
@@ -169,12 +257,14 @@  static ssize_t tap_receive(NetClientState *nc, const uint8_t *buf, size_t size)
     return tap_write_packet(s, iov, 1);
 }
 
+#ifndef CONFIG_VIRTIO_NET_DATA_PLANE
 static int tap_can_send(void *opaque)
 {
     TAPState *s = opaque;
 
     return qemu_can_send_packet(&s->nc);
 }
+#endif
 
 #ifndef __sun__
 ssize_t tap_read_packet(int tapfd, uint8_t *buf, int maxlen)