start removing libuv

include/llarp/ev.h

@@ -19,20 +19,21 @@ int llarp_ev_loop_run(struct llarp_ev_loop *ev);
 /** stop event loop and wait for it to complete all jobs */
 void llarp_ev_loop_stop(struct llarp_ev_loop *ev);
 
-struct llarp_udp_listener {
+struct llarp_udp_io {
   struct sockaddr_in6 *addr;
   void *user;
   void *impl;
-  void (*recvfrom)(struct llarp_udp_listener *, const struct sockaddr *, char *,
+  void (*recvfrom)(struct llarp_udp_io *, const struct sockaddr *, char *,
                    ssize_t);
-  void (*closed)(struct llarp_udp_listener *);
 };
+  
+int llarp_ev_add_udp(struct llarp_ev_loop *ev,
+                              struct llarp_udp_io *udp);
 
-int llarp_ev_add_udp_listener(struct llarp_ev_loop *ev,
-                              struct llarp_udp_listener *listener);
-
-int llarp_ev_close_udp_listener(struct llarp_udp_listener *listener);
+int llarp_ev_udp_sendto(struct llarp_udp_io * udp, const struct sockaddr * to, const void * data, size_t sz);
 
+int llarp_ev_close_udp(struct llarp_udp_io * udp);
+  
 struct llarp_ev_async_call;
 
 typedef void (*llarp_ev_work_func)(struct llarp_ev_async_call *);

llarp/ev.cpp

@@ -1,7 +1,13 @@
 #include <llarp/ev.h>
-#include <uv.h>
 #include "mem.hpp"
 
+#ifdef __linux__
+#include "ev_epoll.hpp"
+#endif
+#ifdef __freebsd__
+#include "ev_kqueue.hpp"
+#endif
+
 #include <mutex>
 #include <queue>
 
@@ -14,7 +20,6 @@ struct llarp_ev_caller {
 
   llarp_ev_caller(llarp_ev_loop *ev, llarp_ev_work_func func)
       : loop(ev), work(func) {
-    async.data = this;
   }
 
   ~llarp_ev_caller() {}
@@ -49,167 +54,57 @@ struct llarp_ev_caller {
 
   std::mutex access;
   struct llarp_ev_loop *loop;
-  uv_async_t async;
   std::deque<llarp_ev_async_call> pending;
   llarp_ev_work_func work;
 };
 
-struct llarp_ev_loop {
-  uv_loop_t _loop;
-
-  static void *operator new(size_t sz) {
-    return llarp_g_mem.alloc(sz, llarp::alignment<llarp_ev_loop>());
-  }
-
-  static void operator delete(void *ptr) { llarp_g_mem.free(ptr); }
-
-  uv_loop_t *loop() { return &_loop; }
-};
-
-namespace llarp {
-struct udp_listener {
-  static void *operator new(size_t sz) {
-    return llarp_g_mem.alloc(sz, alignment<udp_listener>());
-  }
-
-  static void operator delete(void *ptr) { llarp_g_mem.free(ptr); }
-
-  uv_udp_t _handle;
-  struct llarp_udp_listener *listener;
-
-  void recvfrom(const struct sockaddr *addr, char *buff, ssize_t sz) {
-    if (listener->recvfrom) listener->recvfrom(listener, addr, buff, sz);
-  }
-
-  /** called after closed */
-  void closed() {
-    if (listener->closed) listener->closed(listener);
-    listener->impl = nullptr;
-  }
-
-  uv_udp_t *udp() { return &_handle; }
-};
-
-static void udp_alloc_cb(uv_handle_t *h, size_t sz, uv_buf_t *buf) {
-  buf->base = static_cast<char *>(llarp_g_mem.alloc(sz, 1024));
-  buf->len = sz;
-}
-
-static void udp_recv_cb(uv_udp_t *handle, ssize_t nread, const uv_buf_t *buf,
-                        const struct sockaddr *addr, unsigned flags) {
-  udp_listener *l = static_cast<udp_listener *>(handle->data);
-  l->recvfrom(addr, buf->base, nread);
-  llarp_g_mem.free(buf->base);
-}
-
-static void udp_close_cb(uv_handle_t *handle) {
-  udp_listener *l = static_cast<udp_listener *>(handle->data);
-  l->closed();
-  delete l;
-}
-
-}  // namespace llarp
-
-namespace llarp {
-
-static void ev_caller_async_closed(uv_handle_t *handle) {
-  llarp_ev_caller *caller = static_cast<llarp_ev_caller *>(handle->data);
-  delete caller;
-}
-
-static void ev_handle_async_call(uv_async_t *handle) {
-  llarp_ev_caller *caller = static_cast<llarp_ev_caller *>(handle->data);
-  caller->Call();
-}
-}  // namespace llarp
-
 extern "C" {
 void llarp_ev_loop_alloc(struct llarp_ev_loop **ev) {
-  *ev = new llarp_ev_loop;
-  if (*ev) {
-    uv_loop_init((*ev)->loop());
-  }
+#ifdef __linux__
+  *ev = new llarp_epoll_loop;
+#endif
+#ifdef __freebsd__
+  *ev = new llarp_kqueue_loop;
+#endif
 }
 
 void llarp_ev_loop_free(struct llarp_ev_loop **ev) {
-  if (*ev) {
-    uv_loop_close((*ev)->loop());
-    llarp_g_mem.free(*ev);
-  }
+  delete *ev;
   *ev = nullptr;
 }
 
 int llarp_ev_loop_run(struct llarp_ev_loop *ev) {
-  return uv_run(ev->loop(), UV_RUN_DEFAULT);
+  return ev->run();
 }
 
 int llarp_ev_add_udp_listener(struct llarp_ev_loop *ev,
                               struct llarp_udp_listener *listener) {
-  int ret = 0;
-  llarp::udp_listener *l = new llarp::udp_listener;
-  listener->impl = l;
-  l->udp()->data = l;
-  l->listener = listener;
-
-  ret = uv_udp_init(ev->loop(), l->udp());
-  if (ret == 0) {
-    ret = uv_udp_bind(l->udp(), (sockaddr *)listener->addr, 0);
-    if (ret == 0) {
-      char addr[128] = {0};
-      uv_ip6_name(listener->addr, addr, sizeof(addr));
-      printf("bound udp listener at %s port %d\n", addr,
-             ntohs(listener->addr->sin6_port));
-      ret = uv_udp_recv_start(l->udp(), &llarp::udp_alloc_cb,
-                              &llarp::udp_recv_cb);
-    }
-  }
+  int ret = -1;
   return ret;
 }
 
 int llarp_ev_close_udp_listener(struct llarp_udp_listener *listener) {
   int ret = -1;
-  if (listener) {
-    llarp::udp_listener *l = static_cast<llarp::udp_listener *>(listener->impl);
-    if (l) {
-      if (!uv_udp_recv_stop(l->udp())) {
-        uv_close((uv_handle_t *)l->udp(), &llarp::udp_close_cb);
-        ret = 0;
-      }
-    }
-  }
   return ret;
 }
 
-void llarp_ev_loop_stop(struct llarp_ev_loop *loop) { uv_stop(loop->loop()); }
+void llarp_ev_loop_stop(struct llarp_ev_loop *loop) { loop->stop(); }
 
 struct llarp_ev_caller *llarp_ev_prepare_async(struct llarp_ev_loop *loop,
                                                llarp_ev_work_func work) {
-  llarp_ev_caller *caller = new llarp_ev_caller(loop, work);
-  if (uv_async_init(loop->loop(), &caller->async,
-                    llarp::ev_handle_async_call) == 0)
-    return caller;
-  else {
-    delete caller;
-    return nullptr;
-  }
+  return new llarp_ev_caller(loop, work);
 }
 
 bool llarp_ev_call_async(struct llarp_ev_caller *caller, void *user) {
-  if (caller->appendCall(user))
-    return uv_async_send(&caller->async) == 0;
-  else
-    return true;
+  return false;
 }
 
 bool llarp_ev_call_many_async(struct llarp_ev_caller *caller, void **users,
                               size_t n) {
-  if (caller->appendManyCalls(users, n))
-    return uv_async_send(&caller->async) == 0;
-  else
-    return true;
+  return false;
 }
 
 void llarp_ev_caller_stop(struct llarp_ev_caller *caller) {
-  uv_close((uv_handle_t *)&caller->async, &llarp::ev_caller_async_closed);
+  delete caller;
 }
 }

llarp/ev.hpp

@@ -0,0 +1,35 @@
+#ifndef LLARP_EV_HPP
+#define LLARP_EV_HPP
+#include <llarp/ev.h>
+
+#include <unistd.h>
+
+namespace llarp
+{
+  struct ev_io
+  {
+    char buff[2048];
+    int fd;
+    ev_io(int f) : fd(f) {};
+    virtual int read() = 0;
+    virtual int sendto(const sockaddr * dst, const void * data, size_t sz) = 0;
+    virtual ~ev_io()
+    {
+      ::close(fd);
+    };
+  };
+};
+
+struct llarp_ev_loop {
+
+  virtual bool init() = 0;
+  virtual int run() = 0;
+  virtual void stop() = 0;
+
+  virtual bool udp_listen(llarp_udp_io * l) = 0;
+  
+  virtual ~llarp_ev_loop() {};
+  
+};
+
+#endif

llarp/ev_epoll.hpp

@@ -0,0 +1,148 @@
+#ifndef EV_EPOLL_HPP
+#define EV_EPOLL_HPP
+#include "ev.hpp"
+#include <unistd.h>
+#include <sys/epoll.h>
+#include <netinet/in.h>
+
+
+namespace llarp
+{
+  struct udp_listener : public ev_io
+  {
+    llarp_udp_io * udp;
+    
+    udp_listener(int fd, llarp_udp_io * u) :
+      ev_io(fd),
+      udp(u) {};
+
+    ~udp_listener() {}
+    
+    virtual int read()
+    {
+      sockaddr src;
+      socklen_t slen;
+      int ret = ::recvfrom(fd, buff, sizeof(buff), 0, &src, &slen);
+      if (ret == -1) return -1;
+      udp->recvfrom(udp, &src, buff, ret);
+      return 0;
+    }
+
+    virtual int sendto(const sockaddr * to, const void * data, size_t sz)
+    {
+      socklen_t slen;
+      switch(to->sa_family)
+      {
+      case AF_INET:
+        slen = sizeof(struct sockaddr_in);
+        break;
+      case AF_INET6:
+        slen = sizeof(struct sockaddr_in6);
+        break;
+      default:
+        return -1;
+      }
+      return ::sendto(fd, data, sz, 0, to, slen);
+    }
+  };
+};
+
+struct llarp_epoll_loop : public llarp_ev_loop
+{
+
+  int epollfd;
+  
+  llarp_epoll_loop() : epollfd(-1)
+  {
+    
+  }
+
+  ~llarp_epoll_loop()
+  {
+  }
+
+  bool init()
+  {
+    if(epollfd == -1)
+      epollfd = epoll_create1(EPOLL_CLOEXEC);
+    return epollfd != -1;
+  }
+
+  int run()
+  {
+    epoll_event events[1024];
+    int result;
+    do
+    {
+      result = epoll_wait(epollfd, events, 1024, 0);
+      if(result > 0)
+      {
+        int idx = 0;
+        while(idx < result)
+        {
+          llarp::ev_io * ev = static_cast<llarp::ev_io*>(events[idx].data.ptr);
+          if(events[idx].events & EPOLLIN)
+          {
+            if ( ev->read() == -1)
+            {
+              epoll_ctl(epollfd, EPOLL_CTL_DEL, ev->fd, nullptr);
+            }
+          }
+          ++idx;
+        }
+      }
+    }
+    while(result != -1);
+    return result;
+  }
+
+  int udp_bind(const sockaddr * addr)
+  {
+    socklen_t slen;
+    switch(addr->sa_family)
+    {
+    case AF_INET:
+      slen = sizeof(struct sockaddr_in);
+      break;
+    case AF_INET6:
+      slen = sizeof(struct sockaddr_in6);
+      break;
+    default:
+      return -1;
+    }
+    int fd = socket(addr->sa_family, SOCK_DGRAM, 0);
+    if (fd == -1) return -1;
+    if(bind(fd, addr, slen) == -1)
+    {
+      close(fd);
+      return -1;
+    }
+    return fd;
+  }
+
+  bool udp_listen(llarp_udp_io * l)
+  {
+    int fd = udp_bind((sockaddr*)l->addr);
+    if (fd == -1) return false;
+    llarp::udp_listener * listener = new llarp::udp_listener(fd, l);
+    epoll_event ev;
+    ev.data.ptr = listener;
+    ev.events = EPOLLIN;
+    if(epoll_ctl(epollfd, EPOLL_CTL_ADD, fd, &ev) == -1)
+    {
+      delete listener;
+      return false;
+    }
+    return true;
+  }
+  
+  void stop()
+  {
+    if(epollfd != -1)
+      ::close(epollfd);
+    
+    epollfd = -1;
+  }
+};
+
+#endif