talezero/integration
2
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

A bit of refactoring before SSL conversion

Browse Source
This commit is contained in:
jyelon
2022-01-05 13:14:44 -05:00
parent c733023823
commit 0833662539
1 changed files with 105 additions and 96 deletions
Download Patch File Download Diff File Expand all files Collapse all files

97
luprex/core/cpp/driver-linux.cpp
View File

@@ -16,6 +16,8 @@
#include <arpa/inet.h>
#include <sys/types.h>
#include <netdb.h>
using SOCKET=int;
const int INVALID_SOCKET = -1;
struct termios orig_termios;
@@ -45,29 +47,6 @@ static void enableRawMode() {
assert(status >= 0);
}
class Driver {
public:
using PollVec = std::vector<struct pollfd>;
using SOCKET=int;
const int INVALID_SOCKET = -1;
static const int MAX_CHAN = DrivenEngine::MAX_CHAN;
DrivenEngine *driven_;
SOCKET socket_[MAX_CHAN];
bool connected_[MAX_CHAN];
bool short_sleep_;
std::map<int, SOCKET> listen_sockets_;
std::unique_ptr<char[]> chbuf;
int64_t basetime_;
int64_t get_now() {
struct timeval tv;
gettimeofday(&tv, nullptr);
int64_t tv_sec = tv.tv_sec;
int64_t tv_usec = tv.tv_usec;
return tv_sec * 1000000 + tv_usec;
}
SOCKET open_connection(const std::string &target, std::string &err) {
struct addrinfo *addrs = nullptr;
struct addrinfo *goodaddr = nullptr;
@@ -138,11 +117,39 @@ public:
return sock;
}
class Driver {
public:
enum ChanState {
CHAN_INACTIVE,
CHAN_CONNECTING,
CHAN_OPEN,
};
struct ChanInfo {
ChanState state;
SOCKET socket;
};
static const int MAX_CHAN = DrivenEngine::MAX_CHAN;
DrivenEngine *driven_;
ChanInfo channels_[MAX_CHAN];
bool short_sleep_;
std::map<int, SOCKET> listen_sockets_;
std::unique_ptr<char[]> chbuf;
int64_t basetime_;
int64_t get_now() {
struct timeval tv;
gettimeofday(&tv, nullptr);
int64_t tv_sec = tv.tv_sec;
int64_t tv_usec = tv.tv_usec;
return tv_sec * 1000000 + tv_usec;
}
void init(DrivenEngine *de) {
driven_ = de;
for (int i = 0; i < MAX_CHAN; i++) {
socket_[i] = INVALID_SOCKET;
connected_[i] = false;
channels_[i].state = CHAN_INACTIVE;
channels_[i].socket = INVALID_SOCKET;
}
short_sleep_ = false;
chbuf.reset(new char[65536]);
@@ -172,10 +179,10 @@ public:
void handle_new_closed_sockets() {
for (int chid = 1; chid < MAX_CHAN; chid++) {
if (driven_->drv_get_channel_released(chid)) {
if (socket_[chid] != INVALID_SOCKET) {
assert(close(socket_[chid]) == 0);
socket_[chid] = INVALID_SOCKET;
connected_[chid] = false;
if (channels_[chid].state != CHAN_INACTIVE) {
assert(close(channels_[chid].socket) == 0);
channels_[chid].state = CHAN_INACTIVE;
channels_[chid].socket = INVALID_SOCKET;
}
driven_->drv_notify_close(chid, "");
}
@@ -186,15 +193,15 @@ public:
std::set<int> chans;
driven_->drv_get_new_outgoing(chans);
for (int chid : chans) {
assert(socket_[chid] == INVALID_SOCKET);
assert(channels_[chid].state == CHAN_INACTIVE);
std::string err;
SOCKET sock = open_connection(driven_->drv_get_target(chid), err);
if (sock == INVALID_SOCKET) {
driven_->drv_notify_close(chid, err);
short_sleep_ = true;
} else {
socket_[chid] = sock;
connected_[chid] = false;
channels_[chid].state = CHAN_CONNECTING;
channels_[chid].socket = sock;
}
}
}
@@ -225,12 +232,12 @@ public:
driven_->drv_set_clock(double(now) / 1000000.0);
}
void close_socket(int chid, const std::string err) {
assert(socket_[chid] != INVALID_SOCKET);
assert(close(socket_[chid]) == 0);
void close_channel(int chid, const std::string err) {
assert(channels_[chid].state != CHAN_INACTIVE);
assert(close(channels_[chid].socket) == 0);
driven_->drv_notify_close(chid, err);
socket_[chid] = INVALID_SOCKET;
connected_[chid] = false;
channels_[chid].state = CHAN_INACTIVE;
channels_[chid].socket = INVALID_SOCKET;
short_sleep_ = true;
}
@@ -239,8 +246,9 @@ public:
SOCKET chsock = accept(sock, nullptr, nullptr);
if (chsock > 0) {
int chid = driven_->drv_notify_accept(port);
socket_[chid] = chsock;
connected_[chid] = true;
assert(channels_[chid].state == CHAN_INACTIVE);
channels_[chid].state = CHAN_OPEN;
channels_[chid].socket = chsock;
short_sleep_ = true;
continue;
}
@@ -270,7 +278,7 @@ public:
if (p.second > largest) largest = p.second;
}
for (int chid = 1; chid < MAX_CHAN; chid++) {
SOCKET sock = socket_[chid];
SOCKET sock = channels_[chid].socket;
if (sock == INVALID_SOCKET) continue;
FD_SET(sock, &rfds);
FD_SET(sock, &efds);
@@ -299,15 +307,16 @@ public:
}
for (int chid = 1; chid < MAX_CHAN; chid++) {
SOCKET sock = socket_[chid];
ChanInfo *chan = channels_ + chid;
SOCKET sock = chan->socket;
if (sock == INVALID_SOCKET) continue;
if (FD_ISSET(sock, &wfds)) {
connected_[chid] = true;
chan->state = CHAN_OPEN;
driven_->drv_peek_outgoing(chid, &nbytes, &bytes);
if (nbytes > 0) {
int wbytes = send(sock, bytes, nbytes, 0);
if (wbytes < 0) {
close_socket(chid, "send failure");
close_channel(chid, "send failure");
continue;
} else {
driven_->drv_sent_outgoing(chid, wbytes);
@@ -317,7 +326,7 @@ public:
if (FD_ISSET(sock, &rfds) || FD_ISSET(sock, &efds)) {
int nrecv = recv(sock, chbuf.get(), 65536, 0);
if (nrecv <= 0) {
close_socket(chid, "recv failure");
close_channel(chid, "recv failure");
continue;
} else {
driven_->drv_recv_incoming(chid, nrecv, chbuf.get());