mirror of https://github.com/oxen-io/lokinet
You cannot select more than 25 topics
Topics must start with a letter or number, can include dashes ('-') and can be up to 35 characters long.
668 lines
15 KiB
C++
668 lines
15 KiB
C++
#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>
|
|
#include <algorithm>
|
|
|
|
#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
|
|
#ifndef EV_WRITE_BUF_SZ
|
|
#define EV_WRITE_BUF_SZ (2 * 1024)
|
|
#endif
|
|
|
|
namespace llarp
|
|
{
|
|
#ifdef _WIN32
|
|
struct win32_ev_io
|
|
{
|
|
struct WriteBuffer
|
|
{
|
|
llarp_time_t timestamp = 0;
|
|
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 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;
|
|
}
|
|
};
|
|
};
|
|
|
|
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;
|
|
|
|
// on windows, tcp/udp event loops are socket fds
|
|
// and TUN device is a plain old fd
|
|
std::variant< SOCKET, HANDLE > fd;
|
|
ULONG_PTR listener_id = 0;
|
|
bool isTCP = false;
|
|
bool write = false;
|
|
WSAOVERLAPPED portfd[2];
|
|
|
|
// constructors
|
|
// for udp
|
|
win32_ev_io(SOCKET f) : fd(f)
|
|
{
|
|
memset((void*)&portfd[0], 0, sizeof(WSAOVERLAPPED) * 2);
|
|
};
|
|
// for tun
|
|
win32_ev_io(HANDLE t, LossyWriteQueue_t* q) : fd(t), m_LossyWriteQueue(q)
|
|
{
|
|
memset((void*)&portfd[0], 0, sizeof(WSAOVERLAPPED) * 2);
|
|
}
|
|
// for tcp
|
|
win32_ev_io(SOCKET f, LosslessWriteQueue_t* q)
|
|
: fd(f), m_BlockingWriteQueue(q)
|
|
{
|
|
memset((void*)&portfd[0], 0, sizeof(WSAOVERLAPPED) * 2);
|
|
isTCP = true;
|
|
}
|
|
|
|
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
|
|
virtual ssize_t
|
|
do_write(void* data, size_t sz)
|
|
{
|
|
DWORD w;
|
|
if(std::holds_alternative< HANDLE >(fd))
|
|
{
|
|
WriteFile(std::get< HANDLE >(fd), data, sz, nullptr, &portfd[1]);
|
|
GetOverlappedResult(std::get< HANDLE >(fd), &portfd[1], &w, TRUE);
|
|
}
|
|
else
|
|
{
|
|
WriteFile((HANDLE)std::get< SOCKET >(fd), data, sz, nullptr,
|
|
&portfd[1]);
|
|
GetOverlappedResult((HANDLE)std::get< SOCKET >(fd), &portfd[1], &w,
|
|
TRUE);
|
|
}
|
|
return w;
|
|
}
|
|
|
|
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 ~win32_ev_io()
|
|
{
|
|
closesocket(std::get< SOCKET >(fd));
|
|
};
|
|
};
|
|
#endif
|
|
|
|
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 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;
|
|
}
|
|
};
|
|
};
|
|
|
|
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;
|
|
|
|
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(void* 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;
|
|
}
|
|
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);
|
|
};
|
|
};
|
|
|
|
// finally create aliases by platform
|
|
#ifdef _WIN32
|
|
using ev_io = win32_ev_io;
|
|
#define sizeof(sockaddr_un) 115
|
|
#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;
|
|
|
|
/// 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)
|
|
{
|
|
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()
|
|
{
|
|
}
|
|
|
|
/// 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)
|
|
{
|
|
llarp::LogError("tcp_conn error: ", strerror(errno));
|
|
if(_conn->error)
|
|
_conn->error(_conn);
|
|
}
|
|
}
|
|
|
|
virtual ssize_t
|
|
do_write(void* buf, size_t sz);
|
|
|
|
virtual int
|
|
read(void* 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)
|
|
{
|
|
tcp->impl = this;
|
|
}
|
|
|
|
bool
|
|
tick()
|
|
{
|
|
if(tcp->tick)
|
|
tcp->tick(tcp);
|
|
return true;
|
|
}
|
|
|
|
/// actually does accept() :^)
|
|
virtual int
|
|
read(void*, size_t);
|
|
};
|
|
|
|
}; // namespace llarp
|
|
|
|
// this (nearly!) abstract base class
|
|
// is overriden for each platform
|
|
struct llarp_ev_loop
|
|
{
|
|
byte_t readbuf[EV_READ_BUF_SZ] = {0};
|
|
llarp_time_t _now = 0;
|
|
virtual bool
|
|
init() = 0;
|
|
virtual int
|
|
run() = 0;
|
|
|
|
virtual int
|
|
tick(int ms) = 0;
|
|
|
|
virtual void
|
|
stop() = 0;
|
|
|
|
virtual bool
|
|
udp_listen(llarp_udp_io* l, const sockaddr* src) = 0;
|
|
|
|
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;
|
|
|
|
virtual llarp::ev_io*
|
|
bind_tcp(llarp_tcp_acceptor* tcp, const sockaddr* addr) = 0;
|
|
|
|
/// return false on socket error (non blocking)
|
|
virtual bool
|
|
tcp_connect(llarp_tcp_connecter* tcp, const sockaddr* addr) = 0;
|
|
|
|
/// register event listener
|
|
virtual bool
|
|
add_ev(llarp::ev_io* ev, bool write) = 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
|