lokinet/llarp/ev.hpp

731 lines
16 KiB
C++
Raw Normal View History

2018-04-30 13:18:57 +00:00
#ifndef LLARP_EV_HPP
#define LLARP_EV_HPP
#include <ev.h>
#include <codel.hpp>
#include <threading.hpp>
// writev
#ifndef _WIN32
#include <sys/uio.h>
#endif
#include <algorithm>
#include <buffer.h>
#include <deque>
#include <list>
#include <unistd.h>
#ifdef _WIN32
#include <win32_up.h>
#include <win32_upoll.h>
2018-12-13 23:02:41 +00:00
// 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;
2018-12-03 20:37:25 +00:00
#else
2018-12-04 23:45:08 +00:00
#if defined(__FreeBSD__) || defined(__OpenBSD__) || defined(__NetBSD__) \
|| (__APPLE__ && __MACH__)
#include <sys/event.h>
#endif
2018-12-03 20:37:25 +00:00
#include <sys/un.h>
#endif
2018-08-15 15:36:34 +00:00
#ifndef MAX_WRITE_QUEUE_SIZE
2018-11-09 14:48:43 +00:00
#define MAX_WRITE_QUEUE_SIZE (1024UL)
2018-08-15 15:36:34 +00:00
#endif
2018-04-30 13:18:57 +00:00
2018-09-06 20:31:58 +00:00
#ifndef EV_READ_BUF_SZ
2018-11-09 14:48:43 +00:00
#define EV_READ_BUF_SZ (4 * 1024UL)
2018-09-06 20:31:58 +00:00
#endif
2018-10-27 18:26:08 +00:00
#ifndef EV_WRITE_BUF_SZ
2018-11-09 14:48:43 +00:00
#define EV_WRITE_BUF_SZ (2 * 1024UL)
2018-10-27 18:26:08 +00:00
#endif
2018-09-06 20:31:58 +00:00
namespace llarp
{
#ifdef _WIN32
struct win32_ev_io
2018-10-29 16:48:36 +00:00
{
2018-10-03 10:40:32 +00:00
struct WriteBuffer
2018-08-15 15:36:34 +00:00
{
2018-10-29 16:48:36 +00:00
llarp_time_t timestamp = 0;
2018-10-03 10:40:32 +00:00
size_t bufsz;
2018-10-31 01:48:41 +00:00
byte_t buf[EV_WRITE_BUF_SZ] = {0};
2018-10-03 10:40:32 +00:00
WriteBuffer() = default;
2018-10-25 12:09:29 +00:00
WriteBuffer(const byte_t* ptr, size_t sz)
2018-10-03 10:40:32 +00:00
{
if(sz <= sizeof(buf))
{
bufsz = sz;
memcpy(buf, ptr, bufsz);
}
else
bufsz = 0;
}
struct GetTime
{
llarp_time_t
operator()(const WriteBuffer& buf) const
2018-10-03 10:40:32 +00:00
{
2018-10-29 16:48:36 +00:00
return buf.timestamp;
2018-10-03 10:40:32 +00:00
}
};
2018-12-03 14:39:30 +00:00
struct GetNow
{
llarp_ev_loop* loop;
GetNow(llarp_ev_loop* l) : loop(l)
{
}
llarp_time_t
operator()() const
{
return llarp_ev_loop_time_now_ms(loop);
}
};
2018-10-03 10:40:32 +00:00
struct PutTime
{
llarp_ev_loop* loop;
PutTime(llarp_ev_loop* l) : loop(l)
{
}
void
operator()(WriteBuffer& buf)
2018-10-03 10:40:32 +00:00
{
2018-10-29 16:48:36 +00:00
buf.timestamp = llarp_ev_loop_time_now_ms(loop);
2018-10-03 10:40:32 +00:00
}
};
struct Compare
{
bool
operator()(const WriteBuffer& left, const WriteBuffer& right) const
{
return left.timestamp < right.timestamp;
}
};
};
using LosslessWriteQueue_t = std::deque< WriteBuffer >;
2018-12-23 13:29:11 +00:00
intptr_t
fd; // Sockets only, fuck UNIX-style reactive IO with a rusty knife
2018-12-23 13:29:11 +00:00
int flags = 0;
win32_ev_io(intptr_t f) : fd(f){};
/// for tcp
2018-12-23 13:29:11 +00:00
win32_ev_io(intptr_t f, LosslessWriteQueue_t* q)
: fd(f), m_BlockingWriteQueue(q)
{
}
virtual void
error()
{
2018-12-14 14:38:37 +00:00
char ebuf[1024];
int err = WSAGetLastError();
FormatMessage(FORMAT_MESSAGE_FROM_SYSTEM, nullptr, err, LANG_NEUTRAL,
ebuf, 1024, nullptr);
llarp::LogError(ebuf);
}
virtual int
2018-12-03 14:39:30 +00:00
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;
};
2018-10-25 12:39:32 +00:00
/// return false if we want to deregister and remove ourselves
virtual bool
tick()
{
return true;
};
/// used for tun interface and tcp conn
2018-10-27 20:02:37 +00:00
virtual ssize_t
do_write(void* data, size_t sz)
{
return uwrite(fd, (char*)data, sz);
}
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)
{
2018-12-23 13:29:11 +00:00
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()
{
uclose(fd);
2018-10-03 10:40:32 +00:00
};
};
2018-12-14 14:38:37 +00:00
#else
struct posix_ev_io
{
struct WriteBuffer
{
llarp_time_t timestamp = 0;
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& buf) const
{
return buf.timestamp;
}
};
struct GetNow
{
llarp_ev_loop* loop;
GetNow(llarp_ev_loop* l) : loop(l)
{
}
llarp_time_t
operator()() const
{
return llarp_ev_loop_time_now_ms(loop);
}
};
struct PutTime
{
llarp_ev_loop* loop;
PutTime(llarp_ev_loop* 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 LossyWriteQueue_t =
llarp::util::CoDelQueue< WriteBuffer, WriteBuffer::GetTime,
WriteBuffer::PutTime, WriteBuffer::Compare,
WriteBuffer::GetNow, llarp::util::NullMutex,
llarp::util::NullLock, 5, 100, 1024 >;
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)
{
}
2018-11-01 12:47:14 +00:00
virtual void
error()
{
llarp::LogError(strerror(errno));
}
virtual int
2018-12-01 14:35:11 +00:00
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
2018-10-27 20:02:37 +00:00
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;
}
else 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_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 == -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();
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);
};
};
2018-12-14 14:38:37 +00:00
#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
{
2018-11-01 12:47:14 +00:00
sockaddr_storage _addr;
bool _shouldClose = false;
bool _calledConnected = false;
llarp_tcp_conn tcp;
// null if inbound otherwise outbound
llarp_tcp_connecter* _conn;
/// 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;
}
/// outbound
tcp_conn(llarp_ev_loop* loop, int fd, const sockaddr* addr,
llarp_tcp_connecter* conn)
: ev_io(fd, new LosslessWriteQueue_t{}), _conn(conn)
{
2018-11-01 12:47:14 +00:00
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;
}
virtual ~tcp_conn()
{
}
2018-11-01 12:47:14 +00:00
/// start connecting
void
connect();
/// calls connected hooks
void
connected()
{
// we are connected yeh boi
if(_conn)
{
if(_conn->connected && !_calledConnected)
_conn->connected(_conn, &tcp);
}
_calledConnected = true;
}
void
flush_write();
void
flush_write_buffers(size_t a)
{
connected();
ev_io::flush_write_buffers(a);
}
void
error()
{
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);
}
}
2018-11-01 12:47:14 +00:00
virtual ssize_t
do_write(void* buf, size_t sz);
virtual int
2018-12-01 14:35:11 +00:00
read(byte_t* buf, size_t sz);
bool
tick();
};
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)
{
2018-11-01 12:47:14 +00:00
tcp->impl = this;
}
2018-10-25 12:39:32 +00:00
bool
tick()
{
if(tcp->tick)
tcp->tick(tcp);
2018-10-25 12:39:32 +00:00
return true;
}
/// actually does accept() :^)
virtual int
2018-12-01 14:35:11 +00:00
read(byte_t*, size_t);
};
2018-10-03 10:40:32 +00:00
}; // namespace llarp
2018-04-30 13:18:57 +00:00
2018-12-03 14:39:30 +00:00
#ifdef _WIN32
struct llarp_fd_promise
{
void
Set(int)
{
}
int
Get()
{
return -1;
}
};
#else
struct llarp_fd_promise
{
llarp_fd_promise(std::promise< int >* p) : _impl(p)
{
}
std::promise< int >* _impl;
void
Set(int fd)
{
_impl->set_value(fd);
}
int
Get()
{
auto future = _impl->get_future();
future.wait();
return future.get();
}
};
2018-12-03 14:39:30 +00:00
#endif
// this (nearly!) abstract base class
// is overriden for each platform
struct llarp_ev_loop
{
byte_t readbuf[EV_READ_BUF_SZ] = {0};
2018-11-02 14:58:12 +00:00
llarp_time_t _now = 0;
virtual bool
init() = 0;
virtual int
run() = 0;
2018-06-06 12:46:26 +00:00
virtual int
2018-06-06 21:23:57 +00:00
tick(int ms) = 0;
2018-06-06 12:46:26 +00:00
virtual void
stop() = 0;
2018-04-30 13:18:57 +00:00
virtual bool
udp_listen(llarp_udp_io* l, const sockaddr* src) = 0;
2018-08-15 15:36:34 +00:00
virtual llarp::ev_io*
create_udp(llarp_udp_io* l, const sockaddr* src) = 0;
virtual bool
udp_close(llarp_udp_io* l) = 0;
2018-10-25 12:39:32 +00:00
/// deregister event listener
virtual bool
close_ev(llarp::ev_io* ev) = 0;
2018-04-30 16:14:20 +00:00
2018-08-15 15:36:34 +00:00
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;
/// return false on socket error (non blocking)
virtual bool
2018-11-01 12:47:14 +00:00
tcp_connect(llarp_tcp_connecter* tcp, const sockaddr* addr) = 0;
2018-10-25 12:39:32 +00:00
/// register event listener
2018-08-15 15:36:34 +00:00
virtual bool
add_ev(llarp::ev_io* ev, bool write) = 0;
2018-08-15 15:36:34 +00:00
2018-08-10 03:51:38 +00:00
virtual bool
running() const = 0;
2018-04-30 16:14:20 +00:00
virtual ~llarp_ev_loop(){};
std::list< std::unique_ptr< llarp::ev_io > > handlers;
2018-08-15 15:36:34 +00:00
void
tick_listeners()
{
2018-11-08 12:31:50 +00:00
auto itr = handlers.begin();
while(itr != handlers.end())
2018-10-25 12:39:32 +00:00
{
if((*itr)->tick())
++itr;
else
{
close_ev(itr->get());
itr = handlers.erase(itr);
}
}
2018-08-15 15:36:34 +00:00
}
2018-04-30 13:18:57 +00:00
};
#endif