lokinet/llarp/ev.hpp

#ifndef LLARP_EV_HPP
#define LLARP_EV_HPP
#include <llarp/ev.h>
// writev
#ifndef _WIN32
#include <sys/uio.h>
#endif
#include <unistd.h>
#include <llarp/buffer.h>
#include <llarp/codel.hpp>
#include <list>
#include <deque>

#ifdef _WIN32
#include <variant>
#endif

#ifndef MAX_WRITE_QUEUE_SIZE
#define MAX_WRITE_QUEUE_SIZE 1024
#endif

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

namespace llarp
{
  struct ev_io
  {
    struct WriteBuffer
    {
      static constexpr size_t BufferSize = 2048;
      llarp_time_t timestamp             = 0;
      size_t bufsz;
      byte_t buf[BufferSize];

      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& w) const
        {
          return w.timestamp;
        }
      };

      struct PutTime
      {
        void
        operator()(WriteBuffer& w) const
        {
          w.timestamp = llarp_time_now_ms();
        }
      };

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

    typedef llarp::util::CoDelQueue< WriteBuffer, WriteBuffer::GetTime,
                                     WriteBuffer::PutTime, WriteBuffer::Compare,
                                     llarp::util::NullMutex,
                                     llarp::util::NullLock, 5, 100, 128 >
        LossyWriteQueue_t;

    typedef std::deque< WriteBuffer > LosslessWriteQueue_t;

#ifndef _WIN32
    int fd;
    int flags = 0;
    ev_io(int f) : fd(f)
    {
    }

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

    /// for tcp
    ev_io(int f, LosslessWriteQueue_t* q) : fd(f), m_BlockingWriteQueue(q)
    {
    }
#else
    // on windows, udp event loops are socket fds
    // and TUN device is a plain old fd
    std::variant< SOCKET, HANDLE > fd;
    // the unique completion key that helps us to
    // identify the object instance for which we receive data
    // Here, we'll use the address of the udp_listener instance, converted
    // to its literal int/int64 representation.
    ULONG_PTR listener_id = 0;
    ev_io(SOCKET f) : fd(f), m_writeq("writequeue"){};
    ev_io(HANDLE t)
        : fd(t), m_writeq("writequeue"){};  // overload for TUN device, which
                                            // _is_ a regular file descriptor
#endif
    virtual int
    read(void* buf, size_t sz) = 0;

    virtual int
    sendto(const sockaddr* dst, const void* data, 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
    ssize_t
    do_write(void* data, size_t sz)
    {
#ifndef _WIN32
      return write(fd, data, sz);
#else
      DWORD w;
      WriteFile(std::get< HANDLE >(fd), data, sz, &w, nullptr);
      return w;
#endif
    }

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

    /// 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()
    {
      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)
      {
        // 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();
          if(errno == EAGAIN || errno == EWOULDBLOCK)
          {
            errno = 0;
            return;
          }
        }
      }
      /// reset errno
      errno = 0;
#if _WIN32
      SetLastError(0);
#endif
    }

    std::unique_ptr< LossyWriteQueue_t > m_LossyWriteQueue;
    std::unique_ptr< LosslessWriteQueue_t > m_BlockingWriteQueue;
    virtual ~ev_io()
    {
#ifndef _WIN32
      ::close(fd);
#else
      closesocket(std::get< SOCKET >(fd));
#endif
    };
  };

  struct tcp_conn : public ev_io
  {
    bool _shouldClose = false;
    llarp_tcp_conn* tcp;
    tcp_conn(int fd, llarp_tcp_conn* conn)
        : ev_io(fd, new LosslessWriteQueue_t{}), tcp(conn)
    {
    }

    virtual ~tcp_conn()
    {
      delete tcp;
    }

    virtual int
    do_write(const void* buf, size_t sz)
    {
      if(_shouldClose)
        return -1;
      #ifdef __linux__
      return ::send(fd, buf, sz, MSG_NOSIGNAL);  // ignore sigpipe
      #else
      return ::send(fd, buf, sz, 0 );
      #endif
    }

    int
    read(void* buf, size_t sz)
    {
      if(_shouldClose)
        return -1;
      ssize_t amount = ::read(fd, buf, sz);
      if(amount > 0)
      {
        if(tcp->read)
          tcp->read(tcp, buf, amount);
      }
      else
      {
        // error
        _shouldClose = true;
        return -1;
      }
      return 0;
    }

    bool
    tick();

    int
    sendto(const sockaddr*, const void*, size_t)
    {
      return -1;
    }
  };

  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)
    {
    }

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

    /// actually does accept() :^)
    virtual int
    read(void*, size_t);
  };

};  // namespace llarp

struct llarp_ev_loop
{
  byte_t readbuf[EV_READ_BUF_SZ];

  virtual bool
  init() = 0;
  virtual int
  run() = 0;

  virtual int
  tick(int ms) = 0;

  virtual void
  stop() = 0;

  bool
  udp_listen(llarp_udp_io* l, const sockaddr* src)
  {
    auto ev = create_udp(l, src);
    if(ev)
    {
#ifdef _WIN32
      l->fd = std::get< SOCKET >(ev->fd);
#else
      l->fd = ev->fd;
#endif
    }
    return ev && add_ev(ev, false);
  }

  virtual llarp::ev_io*
  create_udp(llarp_udp_io* l, const 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 llarp::ev_io*
  create_tun(llarp_tun_io* tun) = 0;

  llarp::ev_io*
  bind_tcp(llarp_tcp_acceptor* tcp, const sockaddr* addr);

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

  virtual bool
  running() const = 0;

  virtual ~llarp_ev_loop(){};

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

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

#endif