lokinet/llarp/ev/ev.hpp
Stephen Shelton 273270916e
The Great Wall of Blame
This commit reflects changes to clang-format rules. Unfortunately,
these rule changes create a massive change to the codebase, which
causes an apparent rewrite of git history.

Git blame's --ignore-rev flag can be used to ignore this commit when
attempting to `git blame` some code.
2020-04-07 12:38:56 -06:00

890 lines
18 KiB
C++

#ifndef LLARP_EV_HPP
#define LLARP_EV_HPP
#include <net/net_addr.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
{
sockaddr_storage _addr;
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)
{
socklen_t slen = sizeof(sockaddr_in);
if (addr->sa_family == AF_INET6)
slen = sizeof(sockaddr_in6);
else if (addr->sa_family == AF_UNIX)
slen = sizeof(sockaddr_un);
memcpy(&_addr, addr, slen);
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 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 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 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 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;
};
struct PacketBuffer
{
PacketBuffer(PacketBuffer&& other)
{
_ptr = other._ptr;
_sz = other._sz;
other._ptr = nullptr;
other._sz = 0;
}
PacketBuffer(const PacketBuffer&) = delete;
PacketBuffer&
operator=(const PacketBuffer&) = delete;
PacketBuffer() : PacketBuffer(nullptr, 0){};
explicit PacketBuffer(size_t sz) : _sz{sz}
{
_ptr = new char[sz];
}
PacketBuffer(char* buf, size_t sz)
{
_ptr = buf;
_sz = sz;
}
~PacketBuffer()
{
if (_ptr)
delete[] _ptr;
}
byte_t*
data()
{
return (byte_t*)_ptr;
}
size_t
size()
{
return _sz;
}
byte_t& operator[](size_t sz)
{
return data()[sz];
}
void
reserve(size_t sz)
{
if (_ptr)
delete[] _ptr;
_ptr = new char[sz];
_sz = sz;
}
private:
char* _ptr = nullptr;
size_t _sz = 0;
};
struct PacketEvent
{
llarp::Addr remote;
PacketBuffer pkt;
};
struct llarp_pkt_list : public std::vector<PacketEvent>
{
};
#endif