lokinet/llarp/ev/ev.hpp

#ifndef LLARP_EV_HPP
#define LLARP_EV_HPP

#include <net/ip_address.hpp>
#include <ev/ev.h>
#include <util/buffer.hpp>
#include <util/codel.hpp>
#include <util/thread/threading.hpp>

// writev
#ifndef _WIN32
#include <sys/uio.h>
#include <unistd.h>
#endif

#include <algorithm>
#include <deque>
#include <list>
#include <future>
#include <utility>

#ifdef _WIN32
// From the preview SDK, should take a look at that
// periodically in case its definition changes
#define UNIX_PATH_MAX 108

typedef struct sockaddr_un
{
  ADDRESS_FAMILY sun_family;    /* AF_UNIX */
  char sun_path[UNIX_PATH_MAX]; /* pathname */
} SOCKADDR_UN, *PSOCKADDR_UN;
#else

#if defined(__FreeBSD__) || defined(__OpenBSD__) || defined(__NetBSD__) || (__APPLE__ && __MACH__)
#include <sys/event.h>
#endif

#include <sys/un.h>
#endif

struct llarp_ev_pkt_pipe;

#ifndef MAX_WRITE_QUEUE_SIZE
#define MAX_WRITE_QUEUE_SIZE (1024UL)
#endif

#ifndef EV_READ_BUF_SZ
#define EV_READ_BUF_SZ (4 * 1024UL)
#endif
#ifndef EV_WRITE_BUF_SZ
#define EV_WRITE_BUF_SZ (4 * 1024UL)
#endif

/// do io and reset errno after
static ssize_t
IO(std::function<ssize_t(void)> iofunc)
{
  ssize_t ret = iofunc();
#ifndef _WIN32
  errno = 0;
#else
  WSASetLastError(0);
#endif
  return ret;
}

namespace llarp
{
#ifdef _WIN32
  struct win32_ev_io
  {
    struct WriteBuffer
    {
      llarp_time_t timestamp = 0s;
      size_t bufsz;
      byte_t buf[EV_WRITE_BUF_SZ] = {0};

      WriteBuffer() = default;

      WriteBuffer(const byte_t* ptr, size_t sz)
      {
        if (sz <= sizeof(buf))
        {
          bufsz = sz;
          memcpy(buf, ptr, bufsz);
        }
        else
          bufsz = 0;
      }

      struct GetTime
      {
        llarp_time_t
        operator()(const WriteBuffer& buf) const
        {
          return buf.timestamp;
        }
      };

      struct GetNow
      {
        llarp_ev_loop_ptr loop;
        GetNow(llarp_ev_loop_ptr l) : loop(l)
        {}

        llarp_time_t
        operator()() const
        {
          return llarp_ev_loop_time_now_ms(loop);
        }
      };

      struct PutTime
      {
        llarp_ev_loop_ptr loop;
        PutTime(llarp_ev_loop_ptr l) : loop(l)
        {}
        void
        operator()(WriteBuffer& buf)
        {
          buf.timestamp = llarp_ev_loop_time_now_ms(loop);
        }
      };

      struct Compare
      {
        bool
        operator()(const WriteBuffer& left, const WriteBuffer& right) const
        {
          return left.timestamp < right.timestamp;
        }
      };
    };

    using LosslessWriteQueue_t = std::deque<WriteBuffer>;

    intptr_t fd;  // Sockets only, fuck UNIX-style reactive IO with a rusty knife

    int flags = 0;
    win32_ev_io(intptr_t f) : fd(f){};

    /// for tcp
    win32_ev_io(intptr_t f, LosslessWriteQueue_t* q) : fd(f), m_BlockingWriteQueue(q)
    {}

    virtual void
    error()
    {
      char ebuf[1024];
      int err = WSAGetLastError();
      FormatMessage(FORMAT_MESSAGE_FROM_SYSTEM, nullptr, err, LANG_NEUTRAL, ebuf, 1024, nullptr);
      llarp::LogError(ebuf);
    }

    virtual int
    read(byte_t* buf, size_t sz) = 0;

    virtual int
    sendto(const SockAddr& dst, const void* data, size_t sz)
    {
      UNREFERENCED_PARAMETER(dst);
      UNREFERENCED_PARAMETER(data);
      UNREFERENCED_PARAMETER(sz);
      return -1;
    };

    /// return false if we want to deregister and remove ourselves
    virtual bool
    tick()
    {
      return true;
    };

    /// used for tun interface and tcp conn
    virtual ssize_t
    do_write(void* data, size_t sz)
    {
      return send(fd, (char*)data, sz, 0);
    }

    bool
    queue_write(const byte_t* buf, size_t sz)
    {
      if (m_BlockingWriteQueue)
      {
        m_BlockingWriteQueue->emplace_back(buf, sz);
        return true;
      }
      else
        return false;
    }

    virtual void
    flush_write()
    {
      flush_write_buffers(0);
    }

    /// called in event loop when fd is ready for writing
    /// requeues anything not written
    /// this assumes fd is set to non blocking
    virtual void
    flush_write_buffers(size_t amount)
    {
      if (m_BlockingWriteQueue)
      {
        if (amount)
        {
          while (amount && m_BlockingWriteQueue->size())
          {
            auto& itr = m_BlockingWriteQueue->front();
            ssize_t result = do_write(itr.buf, std::min(amount, itr.bufsz));
            if (result == -1)
              return;
            ssize_t dlt = itr.bufsz - result;
            if (dlt > 0)
            {
              // queue remaining to front of queue
              WriteBuffer buff(itr.buf + dlt, itr.bufsz - dlt);
              m_BlockingWriteQueue->pop_front();
              m_BlockingWriteQueue->push_front(buff);
              // TODO: errno?
              return;
            }
            m_BlockingWriteQueue->pop_front();
            amount -= result;
          }
        }
        else
        {
          // write buffers
          while (m_BlockingWriteQueue->size())
          {
            auto& itr = m_BlockingWriteQueue->front();
            ssize_t result = do_write(itr.buf, itr.bufsz);
            if (result == -1)
              return;
            ssize_t dlt = itr.bufsz - result;
            if (dlt > 0)
            {
              // queue remaining to front of queue
              WriteBuffer buff(itr.buf + dlt, itr.bufsz - dlt);
              m_BlockingWriteQueue->pop_front();
              m_BlockingWriteQueue->push_front(buff);
              // TODO: errno?
              return;
            }
            m_BlockingWriteQueue->pop_front();
            int wsaerr = WSAGetLastError();
            if (wsaerr == WSA_IO_PENDING || wsaerr == WSAEWOULDBLOCK)
            {
              WSASetLastError(0);
              return;
            }
          }
        }
      }
      /// reset errno
      WSASetLastError(0);
    }

    std::unique_ptr<LosslessWriteQueue_t> m_BlockingWriteQueue;

    virtual ~win32_ev_io()
    {
      closesocket(fd);
    };
  };
#else
  struct posix_ev_io
  {
    struct WriteBuffer
    {
      llarp_time_t timestamp = 0s;
      size_t bufsz;
      byte_t buf[EV_WRITE_BUF_SZ];

      WriteBuffer() = default;

      WriteBuffer(const byte_t* ptr, size_t sz)
      {
        if (sz <= sizeof(buf))
        {
          bufsz = sz;
          memcpy(buf, ptr, bufsz);
        }
        else
          bufsz = 0;
      }

      struct GetTime
      {
        llarp_time_t
        operator()(const WriteBuffer& writebuf) const
        {
          return writebuf.timestamp;
        }
      };

      struct GetNow
      {
        llarp_ev_loop_ptr loop;
        GetNow(llarp_ev_loop_ptr l) : loop(std::move(l))
        {}

        llarp_time_t
        operator()() const
        {
          return llarp_ev_loop_time_now_ms(loop);
        }
      };

      struct PutTime
      {
        llarp_ev_loop_ptr loop;
        PutTime(llarp_ev_loop_ptr l) : loop(std::move(l))
        {}
        void
        operator()(WriteBuffer& writebuf)
        {
          writebuf.timestamp = llarp_ev_loop_time_now_ms(loop);
        }
      };

      struct Compare
      {
        bool
        operator()(const WriteBuffer& left, const WriteBuffer& right) const
        {
          return left.timestamp < right.timestamp;
        }
      };
    };

    using LossyWriteQueue_t = llarp::util::CoDelQueue<
        WriteBuffer,
        WriteBuffer::GetTime,
        WriteBuffer::PutTime,
        WriteBuffer::Compare,
        WriteBuffer::GetNow,
        llarp::util::NullMutex,
        llarp::util::NullLock>;

    using LosslessWriteQueue_t = std::deque<WriteBuffer>;

    int fd;
    int flags = 0;
#if defined(__FreeBSD__) || defined(__OpenBSD__) || defined(__NetBSD__) || (__APPLE__ && __MACH__)
    struct kevent change;
#endif

    posix_ev_io(int f) : fd(f)
    {}

    /// for tun
    posix_ev_io(int f, LossyWriteQueue_t* q) : fd(f), m_LossyWriteQueue(q)
    {}

    /// for tcp
    posix_ev_io(int f, LosslessWriteQueue_t* q) : fd(f), m_BlockingWriteQueue(q)
    {}

    virtual void
    error()
    {
      llarp::LogError(strerror(errno));
    }

    virtual int
    read(byte_t* buf, size_t sz) = 0;

    virtual int
    sendto(
        __attribute__((unused)) const SockAddr& dst,
        __attribute__((unused)) const void* data,
        __attribute__((unused)) size_t sz)
    {
      return -1;
    }

    /// return false if we want to deregister and remove ourselves
    virtual bool
    tick()
    {
      return true;
    }

    /// used for tun interface and tcp conn
    virtual ssize_t
    do_write(void* data, size_t sz)
    {
      return write(fd, data, sz);
    }

    bool
    queue_write(const byte_t* buf, size_t sz)
    {
      if (m_LossyWriteQueue)
      {
        m_LossyWriteQueue->Emplace(buf, sz);
        return true;
      }
      if (m_BlockingWriteQueue)
      {
        m_BlockingWriteQueue->emplace_back(buf, sz);
        return true;
      }

      return false;
    }

    virtual void
    flush_write()
    {
      flush_write_buffers(0);
    }

    virtual void
    before_flush_write()
    {}

    /// called in event loop when fd is ready for writing
    /// requeues anything not written
    /// this assumes fd is set to non blocking
    virtual void
    flush_write_buffers(size_t amount)
    {
      before_flush_write();
      if (m_LossyWriteQueue)
      {
        m_LossyWriteQueue->Process([&](WriteBuffer& buffer) {
          do_write(buffer.buf, buffer.bufsz);
          // if we would block we save the entries for later
          // discard entry
        });
      }
      else if (m_BlockingWriteQueue)
      {
        if (amount)
        {
          while (amount && m_BlockingWriteQueue->size())
          {
            auto& itr = m_BlockingWriteQueue->front();
            ssize_t result = do_write(itr.buf, std::min(amount, itr.bufsz));
            if (result <= 0)
              return;
            ssize_t dlt = itr.bufsz - result;
            if (dlt > 0)
            {
              // queue remaining to front of queue
              WriteBuffer buff(itr.buf + dlt, itr.bufsz - dlt);
              m_BlockingWriteQueue->pop_front();
              m_BlockingWriteQueue->push_front(buff);
              // TODO: errno?
              return;
            }
            m_BlockingWriteQueue->pop_front();
            amount -= result;
          }
        }
        else
        {
          // write buffers
          while (m_BlockingWriteQueue->size())
          {
            auto& itr = m_BlockingWriteQueue->front();
            ssize_t result = do_write(itr.buf, itr.bufsz);
            if (result <= 0)
            {
              errno = 0;
              return;
            }
            ssize_t dlt = itr.bufsz - result;
            if (dlt > 0)
            {
              // queue remaining to front of queue
              WriteBuffer buff(itr.buf + dlt, itr.bufsz - dlt);
              m_BlockingWriteQueue->pop_front();
              m_BlockingWriteQueue->push_front(buff);
              // TODO: errno?
              return;
            }
            m_BlockingWriteQueue->pop_front();
            if (errno == EAGAIN || errno == EWOULDBLOCK)
            {
              errno = 0;
              return;
            }
          }
        }
      }
      /// reset errno
      errno = 0;
    }

    std::unique_ptr<LossyWriteQueue_t> m_LossyWriteQueue;
    std::unique_ptr<LosslessWriteQueue_t> m_BlockingWriteQueue;

    virtual ~posix_ev_io()
    {
      close(fd);
    }
  };
#endif

// finally create aliases by platform
#ifdef _WIN32
  using ev_io = win32_ev_io;
#else
  using ev_io = posix_ev_io;
#endif

  // wew, managed to get away with using
  // 'int fd' across all platforms
  // since we're operating entirely
  // on sockets
  struct tcp_conn : public ev_io
  {
    bool _shouldClose = false;
    bool _calledConnected = false;
    llarp_tcp_conn tcp;
    // null if inbound otherwise outbound
    llarp_tcp_connecter* _conn;

    static void
    DoClose(llarp_tcp_conn* conn)
    {
      static_cast<tcp_conn*>(conn->impl)->_shouldClose = true;
    }

    /// inbound
    tcp_conn(llarp_ev_loop* loop, int _fd) : ev_io(_fd, new LosslessWriteQueue_t{}), _conn(nullptr)
    {
      tcp.impl = this;
      tcp.loop = loop;
      tcp.closed = nullptr;
      tcp.user = nullptr;
      tcp.read = nullptr;
      tcp.tick = nullptr;
      tcp.close = &DoClose;
    }

    /// outbound
    tcp_conn(llarp_ev_loop* loop, int _fd, const SockAddr& addr, llarp_tcp_connecter* conn)
        : ev_io(_fd, new LosslessWriteQueue_t{}), _conn(conn)
    {
      (void)addr;
      tcp.impl = this;
      tcp.loop = loop;
      tcp.closed = nullptr;
      tcp.user = nullptr;
      tcp.read = nullptr;
      tcp.tick = nullptr;
      tcp.close = &DoClose;
    }

    ~tcp_conn() override = default;

    /// start connecting
    void
    connect();

    /// calls connected hooks
    void
    connected()
    {
      sockaddr_storage st;
      socklen_t sl;
      if (getpeername(fd, (sockaddr*)&st, &sl) == 0)
      {
        // we are connected yeh boi
        if (_conn)
        {
          if (_conn->connected && !_calledConnected)
            _conn->connected(_conn, &tcp);
        }
        _calledConnected = true;
      }
      else
      {
        error();
      }
    }

    void
    flush_write() override;

    void
    flush_write_buffers(size_t a) override
    {
      connected();
      ev_io::flush_write_buffers(a);
    }

    void
    error() override
    {
      _shouldClose = true;
      if (_conn)
      {
#ifndef _WIN32
        llarp::LogError("tcp_conn error: ", strerror(errno));
#else
        char ebuf[1024];
        int err = WSAGetLastError();
        FormatMessage(FORMAT_MESSAGE_FROM_SYSTEM, nullptr, err, LANG_NEUTRAL, ebuf, 1024, nullptr);
        llarp::LogError("tcp_conn error: ", ebuf);
#endif
        if (_conn->error)
          _conn->error(_conn);
      }
      errno = 0;
    }

    ssize_t
    do_write(void* buf, size_t sz) override;

    int
    read(byte_t* buf, size_t sz) override;

    bool
    tick() override;
  };

  struct tcp_serv : public ev_io
  {
    llarp_ev_loop* loop;
    llarp_tcp_acceptor* tcp;
    tcp_serv(llarp_ev_loop* l, int _fd, llarp_tcp_acceptor* t) : ev_io(_fd), loop(l), tcp(t)
    {
      tcp->impl = this;
    }

    bool
    tick() override
    {
      if (tcp->tick)
        tcp->tick(tcp);
      return true;
    }

    /// actually does accept() :^)
    int
    read(byte_t*, size_t) override;
  };

}  // namespace llarp

#ifdef _WIN32
struct llarp_fd_promise
{
  void Set(std::pair<int, int>)
  {}

  int
  Get()
  {
    return -1;
  }
};
#else
struct llarp_fd_promise
{
  using promise_val_t = std::pair<int, int>;
  llarp_fd_promise(std::promise<promise_val_t>* p) : _impl(p)
  {}
  std::promise<promise_val_t>* _impl;

  void
  Set(promise_val_t fds)
  {
    _impl->set_value(fds);
  }

  promise_val_t
  Get()
  {
    auto future = _impl->get_future();
    future.wait();
    return future.get();
  }
};
#endif

// this (nearly!) abstract base class
// is overriden for each platform
struct llarp_ev_loop
{
  byte_t readbuf[EV_READ_BUF_SZ] = {0};

  virtual bool
  init() = 0;

  virtual int
  run() = 0;

  virtual bool
  running() const = 0;

  virtual void
  update_time()
  {}

  virtual llarp_time_t
  time_now() const
  {
    return llarp::time_now_ms();
  }

  virtual void
  stopped(){};

  /// return false on socket error (non blocking)
  virtual bool
  tcp_connect(llarp_tcp_connecter* tcp, const llarp::SockAddr& addr) = 0;

  virtual int
  tick(int ms) = 0;

  virtual uint32_t
  call_after_delay(llarp_time_t delay_ms, std::function<void(void)> callback) = 0;

  virtual void
  cancel_delayed_call(uint32_t call_id) = 0;

  virtual bool
  add_ticker(std::function<void(void)> ticker) = 0;

  virtual void
  stop() = 0;

  virtual bool
  udp_listen(llarp_udp_io* l, const llarp::SockAddr& src) = 0;

  virtual bool
  udp_close(llarp_udp_io* l) = 0;
  /// deregister event listener
  virtual bool
  close_ev(llarp::ev_io* ev) = 0;

  virtual bool
  tun_listen(llarp_tun_io* tun)
  {
    auto dev = create_tun(tun);
    tun->impl = dev;
    if (dev)
    {
      return add_ev(dev, false);
    }
    return false;
  }

  virtual llarp::ev_io*
  create_tun(llarp_tun_io* tun) = 0;

  virtual llarp::ev_io*
  bind_tcp(llarp_tcp_acceptor* tcp, const llarp::SockAddr& addr) = 0;

  virtual bool
  add_pipe(llarp_ev_pkt_pipe*)
  {
    return false;
  }

  /// give this event loop a logic thread for calling
  virtual void set_logic(std::shared_ptr<llarp::Logic>) = 0;

  /// register event listener
  virtual bool
  add_ev(llarp::ev_io* ev, bool write) = 0;

  virtual bool
  tcp_listen(llarp_tcp_acceptor* tcp, const llarp::SockAddr& addr)
  {
    auto conn = bind_tcp(tcp, addr);
    return conn && add_ev(conn, true);
  }

  virtual ~llarp_ev_loop() = default;

  std::list<std::unique_ptr<llarp::ev_io>> handlers;

  virtual void
  tick_listeners()
  {
    auto itr = handlers.begin();
    while (itr != handlers.end())
    {
      if ((*itr)->tick())
        ++itr;
      else
      {
        close_ev(itr->get());
        itr = handlers.erase(itr);
      }
    }
  }

  virtual void
  call_soon(std::function<void(void)> f) = 0;

  virtual void
  register_poll_fd_readable(int fd, std::function<void(void)> callback) = 0;

  virtual void
  deregister_poll_fd_readable(int fd) = 0;
};

#endif