KittyPatch/kitty
1
0
Fork 0
Code Issues Pull Requests Projects Releases Wiki Activity

Rewrite the talk loop for interaction with peers

Browse Source 
The new code is simpler and hopefully more robust. Supports
multiple requests/responses. Fixes #2852
This commit is contained in:
Kovid Goyal
2020-07-15 14:31:45 +05:30
parent c0ff39c59d
commit cec4371b51
2 changed files with 212 additions and 259 deletions
Download Patch File Download Diff File Expand all files Collapse all files

3
docs/changelog.rst
View File

@@ -15,6 +15,9 @@ To update |kitty|, :doc:`follow the instructions <binary>`.
- Improve anti-aliasing of triangular box drawing characters, noticeable on - Improve anti-aliasing of triangular box drawing characters, noticeable on
low-resolution screens (:iss:`2844`) low-resolution screens (:iss:`2844`)
- Fix ``kitty @ send-text`` not working reliably when using a docket for remote
control (:iss:`2852`)
- Implement support for box drawing rounded-corners characters (:iss:`2240`) - Implement support for box drawing rounded-corners characters (:iss:`2240`)
- When a character from the Unicode Dingbat block is followed by a space, use - When a character from the Unicode Dingbat block is followed by a space, use

468
kitty/child-monitor.c
View File

@@ -40,7 +40,7 @@ static void (*parse_func)(Screen*, PyObject*, monotonic_t);
typedef struct { typedef struct {
char *data; char *data;
size_t sz; size_t sz;
int fd; id_type peer_id;
} Message; } Message;
typedef struct { typedef struct {
@@ -71,9 +71,8 @@ static const Child EMPTY_CHILD = {0};
pthread_mutex_##op(&screen->which##_buf_lock); pthread_mutex_##op(&screen->which##_buf_lock);
#define children_mutex(op) \ #define children_mutex(op) \
pthread_mutex_##op(&children_lock); pthread_mutex_##op(&children_lock);
#define peer_mutex(op) \ #define talk_mutex(op) \
pthread_mutex_##op(&talk_data.peer_lock); pthread_mutex_##op(&talk_lock);
static Child children[MAX_CHILDREN] = {{0}}; static Child children[MAX_CHILDREN] = {{0}};
@@ -82,7 +81,7 @@ static Child add_queue[MAX_CHILDREN] = {{0}}, remove_queue[MAX_CHILDREN] = {{0}}
static unsigned long remove_notify[MAX_CHILDREN] = {0}; static unsigned long remove_notify[MAX_CHILDREN] = {0};
static size_t add_queue_count = 0, remove_queue_count = 0; static size_t add_queue_count = 0, remove_queue_count = 0;
static struct pollfd fds[MAX_CHILDREN + EXTRA_FDS] = {{0}}; static struct pollfd fds[MAX_CHILDREN + EXTRA_FDS] = {{0}};
static pthread_mutex_t children_lock; static pthread_mutex_t children_lock, talk_lock;
static bool kill_signal_received = false; static bool kill_signal_received = false;
static ChildMonitor *the_monitor = NULL; static ChildMonitor *the_monitor = NULL;
static uint8_t drain_buf[1024]; static uint8_t drain_buf[1024];
@@ -131,6 +130,10 @@ new(PyTypeObject *type, PyObject *args, PyObject UNUSED *kwds) {
PyErr_Format(PyExc_RuntimeError, "Failed to create children_lock mutex: %s", strerror(ret)); PyErr_Format(PyExc_RuntimeError, "Failed to create children_lock mutex: %s", strerror(ret));
return NULL; return NULL;
} }
if ((ret = pthread_mutex_init(&talk_lock, NULL)) != 0) {
PyErr_Format(PyExc_RuntimeError, "Failed to create talk_lock mutex: %s", strerror(ret));
return NULL;
}
self = (ChildMonitor *)type->tp_alloc(type, 0); self = (ChildMonitor *)type->tp_alloc(type, 0);
if (!init_loop_data(&self->io_loop_data)) return PyErr_SetFromErrno(PyExc_OSError); if (!init_loop_data(&self->io_loop_data)) return PyErr_SetFromErrno(PyExc_OSError);
if (!install_signal_handlers(&self->io_loop_data)) return PyErr_SetFromErrno(PyExc_OSError); if (!install_signal_handlers(&self->io_loop_data)) return PyErr_SetFromErrno(PyExc_OSError);
@@ -153,6 +156,7 @@ new(PyTypeObject *type, PyObject *args, PyObject UNUSED *kwds) {
static void static void
dealloc(ChildMonitor* self) { dealloc(ChildMonitor* self) {
pthread_mutex_destroy(&children_lock); pthread_mutex_destroy(&children_lock);
pthread_mutex_destroy(&talk_lock);
Py_CLEAR(self->dump_callback); Py_CLEAR(self->dump_callback);
Py_CLEAR(self->death_notify); Py_CLEAR(self->death_notify);
Py_TYPE(self)->tp_free((PyObject*)self); Py_TYPE(self)->tp_free((PyObject*)self);
@@ -174,7 +178,7 @@ wakeup_io_loop(ChildMonitor *self, bool in_signal_handler) {
static void* io_loop(void *data); static void* io_loop(void *data);
static void* talk_loop(void *data); static void* talk_loop(void *data);
static void send_response(int fd, const char *msg, size_t msg_sz); static void send_response(id_type peer_id, const char *msg, size_t msg_sz);
static void wakeup_talk_loop(bool); static void wakeup_talk_loop(bool);
static bool talk_thread_started = false; static bool talk_thread_started = false;
@@ -324,7 +328,6 @@ parse_input(ChildMonitor *self) {
size_t count = 0, remove_count = 0; size_t count = 0, remove_count = 0;
bool input_read = false; bool input_read = false;
monotonic_t now = monotonic(); monotonic_t now = monotonic();
PyObject *msg = NULL;
children_mutex(lock); children_mutex(lock);
while (remove_queue_count) { while (remove_queue_count) {
remove_queue_count--; remove_queue_count--;
@@ -333,18 +336,6 @@ parse_input(ChildMonitor *self) {
FREE_CHILD(remove_queue[remove_queue_count]); FREE_CHILD(remove_queue[remove_queue_count]);
} }
if (UNLIKELY(self->messages_count)) {
msg = PyTuple_New(self->messages_count);
if (msg) {
for (size_t i = 0; i < self->messages_count; i++) {
Message *m = self->messages + i;
PyTuple_SET_ITEM(msg, i, Py_BuildValue("y#i", m->data, (int)m->sz, m->fd));
free(m->data); m->data = NULL; m->sz = 0;
}
self->messages_count = 0;
} else fatal("Out of memory");
}
if (UNLIKELY(kill_signal_received)) { if (UNLIKELY(kill_signal_received)) {
global_state.quit_request = IMPERATIVE_CLOSE_REQUESTED; global_state.quit_request = IMPERATIVE_CLOSE_REQUESTED;
global_state.has_pending_closes = true; global_state.has_pending_closes = true;
@@ -358,15 +349,34 @@ parse_input(ChildMonitor *self) {
} }
} }
children_mutex(unlock); children_mutex(unlock);
if (msg) {
for (Py_ssize_t i = 0; i < PyTuple_GET_SIZE(msg); i++) { Message *msgs = NULL;
PyObject *resp = PyObject_CallMethod(global_state.boss, "peer_message_received", "O", PyTuple_GET_ITEM(PyTuple_GET_ITEM(msg, i), 0)); size_t msgs_count = 0;
int peer_fd = (int)PyLong_AsLong(PyTuple_GET_ITEM(PyTuple_GET_ITEM(msg, i), 1)); talk_mutex(lock);
if (resp && PyBytes_Check(resp)) send_response(peer_fd, PyBytes_AS_STRING(resp), PyBytes_GET_SIZE(resp)); if (UNLIKELY(self->messages_count)) {
else { send_response(peer_fd, NULL, 0); if (!resp) PyErr_Print(); } msgs = malloc(sizeof(Message) * self->messages_count);
if (msgs) {
memcpy(msgs, self->messages, sizeof(Message) * self->messages_count);
msgs_count = self->messages_count;
}
self->messages_count = 0;
}
talk_mutex(unlock);
if (msgs_count) {
for (size_t i = 0; i < msgs_count; i++) {
Message *msg = msgs + i;
PyObject *resp = NULL;
if (msg->data) {
resp = PyObject_CallMethod(global_state.boss, "peer_message_received", "y#", msg->data, (int)msg->sz);
free(msg->data);
if (!resp) PyErr_Print();
}
if (resp && PyBytes_Check(resp)) send_response(msg->peer_id, PyBytes_AS_STRING(resp), PyBytes_GET_SIZE(resp));
else send_response(msg->peer_id, NULL, 0);
Py_CLEAR(resp); Py_CLEAR(resp);
} }
Py_CLEAR(msg); free(msgs); msgs = NULL;
} }
while(remove_count) { while(remove_count) {
@@ -1279,33 +1289,30 @@ io_loop(void *data) {
// {{{ Talk thread functions // {{{ Talk thread functions
typedef struct { typedef struct {
char *data; id_type id;
size_t capacity, used; size_t num_of_unresponded_messages_sent_to_main_thread;
bool finished_reading;
int fd; int fd;
bool finished, close_socket, is_peer_command; struct {
} PeerReadData; char *data;
static PeerReadData empty_prd = {.fd = -1, 0}; size_t capacity, used, command_end;
bool finished;
} read;
struct {
char *data;
size_t capacity, used;
bool failed;
} write;
} Peer;
static id_type peer_id_counter = 0;
typedef struct { typedef struct {
char *data; size_t num_peers, peers_capacity;
size_t sz, pos; Peer *peers;
int fd;
bool finished;
} PeerWriteData;
static PeerWriteData empty_pwd = {.fd = -1, 0};
typedef struct {
size_t num_listen_fds, num_talk_fds, num_reads, num_writes, num_queued_writes;
size_t fds_capacity, reads_capacity, writes_capacity, queued_writes_capacity;
struct pollfd *fds;
PeerReadData *reads;
PeerWriteData *writes;
PeerWriteData *queued_writes;
LoopData loop_data; LoopData loop_data;
pthread_mutex_t peer_lock;
} TalkData; } TalkData;
static TalkData talk_data = {0}; static TalkData talk_data = {0};
typedef struct pollfd PollFD; typedef struct pollfd PollFD;
#define PEER_LIMIT 256 #define PEER_LIMIT 256
#define nuke_socket(s) { shutdown(s, SHUT_RDWR); safe_close(s, __FILE__, __LINE__); } #define nuke_socket(s) { shutdown(s, SHUT_RDWR); safe_close(s, __FILE__, __LINE__); }
@@ -1318,13 +1325,12 @@ accept_peer(int listen_fd, bool shutting_down) {
if (!shutting_down) perror("accept() on talk socket failed!"); if (!shutting_down) perror("accept() on talk socket failed!");
return false; return false;
} }
size_t fd_idx = talk_data.num_listen_fds + talk_data.num_talk_fds; if (talk_data.num_peers < PEER_LIMIT) {
if (fd_idx < PEER_LIMIT && talk_data.reads_capacity < PEER_LIMIT) { ensure_space_for(&talk_data, peers, Peer, talk_data.num_peers + 8, peers_capacity, 8, false);
ensure_space_for(&talk_data, fds, PollFD, fd_idx + 1, fds_capacity, 8, false); Peer *p = talk_data.peers + talk_data.num_peers++;
talk_data.fds[fd_idx].fd = peer; talk_data.fds[fd_idx].events = POLLIN; memset(p, 0, sizeof(Peer));
ensure_space_for(&talk_data, reads, PeerReadData, talk_data.num_reads + 1, reads_capacity, 8, false); p->fd = peer; p->id = ++peer_id_counter;
talk_data.reads[talk_data.num_reads] = empty_prd; talk_data.reads[talk_data.num_reads++].fd = peer; if (!p->id) p->id = ++peer_id_counter;
talk_data.num_talk_fds++;
} else { } else {
log_error("Too many peers want to talk, ignoring one."); log_error("Too many peers want to talk, ignoring one.");
nuke_socket(peer); nuke_socket(peer);
@@ -1332,168 +1338,102 @@ accept_peer(int listen_fd, bool shutting_down) {
return true; return true;
} }
static inline void
free_peer(Peer *peer) {
free(peer->read.data); peer->read.data = NULL;
free(peer->write.data); peer->write.data = NULL;
if (peer->fd > -1) { nuke_socket(peer->fd); peer->fd = -1; }
}
#define KITTY_CMD_PREFIX "\x1bP@kitty-cmd{" #define KITTY_CMD_PREFIX "\x1bP@kitty-cmd{"
static inline void static inline void
queue_peer_message(ChildMonitor *self, char *buf, size_t sz, int fd) { queue_peer_message(ChildMonitor *self, Peer *peer) {
children_mutex(lock); talk_mutex(lock);
ensure_space_for(self, messages, Message, self->messages_count + 1, messages_capacity, 16, true); ensure_space_for(self, messages, Message, self->messages_count + 16, messages_capacity, 16, true);
Message *m = self->messages + self->messages_count++; Message *m = self->messages + self->messages_count++;
m->data = buf; m->sz = sz; m->fd = fd; memset(m, 0, sizeof(Message));
children_mutex(unlock); if (peer->read.used) {
m->data = malloc(peer->read.used);
if (m->data) {
memcpy(m->data, peer->read.data, peer->read.used);
m->sz = peer->read.used;
}
}
m->peer_id = peer->id;
peer->num_of_unresponded_messages_sent_to_main_thread++;
talk_mutex(unlock);
wakeup_main_loop(); wakeup_main_loop();
} }
static inline void static inline bool
dispatch_peer_command(ChildMonitor *self, PeerReadData *rd, int fd) { has_complete_peer_command(Peer *peer) {
size_t end = 0; peer->read.command_end = 0;
for (size_t i = 0; i < rd->used - 1; i++) { if (peer->read.used > sizeof(KITTY_CMD_PREFIX) && memcmp(peer->read.data, KITTY_CMD_PREFIX, sizeof(KITTY_CMD_PREFIX)-1) == 0) {
if (rd->data[i] == 0x1b && rd->data[i+1] == '\\') { for (size_t i = sizeof(KITTY_CMD_PREFIX)-1; i < peer->read.used - 1; i++) {
end = i + 2; if (peer->read.data[i] == 0x1b && peer->read.data[i+1] == '\\') {
break; peer->read.command_end = i + 2;
break;
}
} }
} }
if (!end) return; return peer->read.command_end ? true : false;
char *buf = malloc(end + 8);
if (buf) {
memcpy(buf, rd->data, end);
queue_peer_message(self, buf, end, fd);
}
rd->is_peer_command = false;
if (rd->used > end) {
rd->used -= end;
memmove(rd->data, rd->data + end, rd->used);
if (rd->used >= sizeof(KITTY_CMD_PREFIX) - 1 && memcmp(rd->data, KITTY_CMD_PREFIX, sizeof(KITTY_CMD_PREFIX)-1) == 0) rd->is_peer_command = true;
} else rd->used = 0;
} }
static inline bool
read_from_peer(ChildMonitor *self, int s) { static inline void
bool read_finished = false; dispatch_peer_command(ChildMonitor *self, Peer *peer) {
for (size_t i = 0; i < talk_data.num_reads; i++) { if (peer->read.command_end) {
PeerReadData *rd = talk_data.reads + i; size_t used = peer->read.used;
#define failed(msg) { read_finished = true; log_error("%s", msg); rd->finished = true; rd->close_socket = true; break; } peer->read.used = peer->read.command_end;
if (rd->fd == s) { queue_peer_message(self, peer);
if (rd->used >= rd->capacity) { peer->read.used = used;
if (rd->capacity >= 64 * 1024) failed("Ignoring too large message from peer"); if (peer->read.used > peer->read.command_end) {
rd->capacity = MAX(8192u, rd->capacity * 2); peer->read.used -= peer->read.command_end;
rd->data = realloc(rd->data, rd->capacity); memmove(peer->read.data, peer->read.data + peer->read.command_end, peer->read.used);
if (!rd->data) failed("Out of memory"); } else peer->read.used = 0;
} peer->read.command_end = 0;
ssize_t n = recv(s, rd->data + rd->used, rd->capacity - rd->used, 0); }
if (n == 0) { }
while (rd->is_peer_command) dispatch_peer_command(self, rd, s);
read_finished = true; rd->finished = true; static inline void
if (rd->used) queue_peer_message(self, rd->data, rd->used, s); read_from_peer(ChildMonitor *self, Peer *peer) {
else free(rd->data); #define failed(msg) { log_error("Reading from peer failed: %s", msg); shutdown(peer->fd, SHUT_RD); peer->read.finished = true; return; }
rd->data = NULL; if (peer->read.used >= peer->read.capacity) {
} else if (n < 0) { if (peer->read.capacity >= 64 * 1024) failed("Ignoring too large message from peer");
if (errno != EINTR) { peer->read.capacity = MAX(8192u, peer->read.capacity * 2);
perror("Error reading from talk peer"); peer->read.data = realloc(peer->read.data, peer->read.capacity);
failed("Closing connection to peer"); if (!peer->read.data) failed("Out of memory");
} }
} else { ssize_t n = recv(peer->fd, peer->read.data + peer->read.used, peer->read.capacity - peer->read.used, 0);
if (!rd->used && memcmp(rd->data, KITTY_CMD_PREFIX, sizeof(KITTY_CMD_PREFIX)-1) == 0) rd->is_peer_command = true; if (n == 0) {
rd->used += n; peer->read.finished = true;
while (rd->is_peer_command) dispatch_peer_command(self, rd, s); shutdown(peer->fd, SHUT_RD);
} while (has_complete_peer_command(peer)) dispatch_peer_command(self, peer);
break; queue_peer_message(self, peer);
} free(peer->read.data); peer->read.data = NULL;
peer->read.used = 0; peer->read.capacity = 0;
} else if (n < 0) {
if (errno != EINTR) failed(strerror(errno));
} else {
peer->read.used += n;
while (has_complete_peer_command(peer)) dispatch_peer_command(self, peer);
} }
#undef failed #undef failed
return read_finished;
}
static inline bool
write_to_peer(int fd) {
bool write_finished = false;
for (size_t i = 0; i < talk_data.num_writes; i++) {
PeerWriteData *wd = talk_data.writes + i;
#define failed(msg) { write_finished = true; log_error("%s", msg); wd->finished = true; break; }
if (wd->fd == fd) {
ssize_t n = send(fd, wd->data + wd->pos, wd->sz - wd->pos, MSG_NOSIGNAL);
if (n == 0) { failed("send() to peer failed to send any data"); }
else if (n < 0) {
if (errno != EINTR) { perror("write() to peer socket failed with error"); failed(""); }
} else {
wd->pos += n;
if (wd->pos >= wd->sz) { write_finished = true; wd->finished = true; }
}
break;
}
}
#undef failed
return write_finished;
} }
static inline void static inline void
remove_poll_fd(int fd) { write_to_peer(Peer *peer) {
size_t count = talk_data.num_talk_fds + talk_data.num_listen_fds; talk_mutex(lock);
for (size_t i = talk_data.num_listen_fds; i < count; i++) { ssize_t n = send(peer->fd, peer->write.data, peer->write.used, MSG_NOSIGNAL);
struct pollfd *pfd = talk_data.fds + i; if (n == 0) { log_error("send() to peer failed to send any data"); peer->write.used = 0; peer->write.failed = true; }
if (pfd->fd == fd) { else if (n < 0) {
size_t num_to_right = count - 1 - i; if (errno != EINTR) { log_error("write() to peer socket failed with error: %s", strerror(errno)); peer->write.used = 0; peer->write.failed = true; }
if (num_to_right) memmove(talk_data.fds + i, talk_data.fds + i + 1, num_to_right * sizeof(struct pollfd)); } else {
talk_data.num_talk_fds--; if ((size_t)n > peer->write.used) memmove(peer->write.data, peer->write.data + n, peer->write.used - n);
break; peer->write.used -= n;
}
}
}
static inline void
prune_finished_reads(void) {
if (!talk_data.num_reads) return;
for (ssize_t i = talk_data.num_reads - 1; i >= 0; i--) {
PeerReadData *rd = talk_data.reads + i;
if (rd->finished) {
remove_poll_fd(rd->fd);
if (rd->close_socket) { nuke_socket(rd->fd); }
else shutdown(rd->fd, SHUT_RD);
free(rd->data);
ssize_t num_to_right = talk_data.num_reads - 1 - i;
if (num_to_right > 0) memmove(talk_data.reads + i, talk_data.reads + i + 1, num_to_right * sizeof(PeerReadData));
else talk_data.reads[i] = empty_prd;
talk_data.num_reads--;
}
}
}
static inline void
prune_finished_writes(void) {
if (!talk_data.num_writes) return;
for (ssize_t i = talk_data.num_writes - 1; i >= 0; i--) {
PeerWriteData *wd = talk_data.writes + i;
if (wd->finished) {
remove_poll_fd(wd->fd);
shutdown(wd->fd, SHUT_WR); safe_close(wd->fd, __FILE__, __LINE__);
free(wd->data);
ssize_t num_to_right = talk_data.num_writes - 1 - i;
if (num_to_right > 0) memmove(talk_data.writes + i, talk_data.writes + i + 1, num_to_right * sizeof(PeerWriteData));
else talk_data.writes[i] = empty_pwd;
talk_data.num_writes--;
}
}
}
static inline void
prune_invalid_fd(int fd) {
log_error("Pruning invalid peer fd: %d", fd);
for (ssize_t i = talk_data.num_reads - 1; i >= 0; i--) {
PeerReadData *rd = talk_data.reads + i;
if (rd->fd == fd) {
rd->finished = true;
rd->close_socket = true;
return;
}
}
for (ssize_t i = talk_data.num_writes - 1; i >= 0; i--) {
PeerWriteData *wd = talk_data.writes + i;
if (wd->fd == fd) {
wd->finished = true;
return;
}
} }
talk_mutex(unlock);
} }
static void static void
@@ -1501,95 +1441,105 @@ wakeup_talk_loop(bool in_signal_handler) {
if (talk_thread_started) wakeup_loop(&talk_data.loop_data, in_signal_handler, "talk_loop"); if (talk_thread_started) wakeup_loop(&talk_data.loop_data, in_signal_handler, "talk_loop");
} }
static inline void static inline void
move_queued_writes(void) { prune_peers(void) {
while (talk_data.num_queued_writes) { for (size_t i = 0; i < talk_data.num_peers; i++) {
PeerWriteData *src = talk_data.queued_writes + --talk_data.num_queued_writes; size_t idx = talk_data.num_peers - 1 - i;
size_t fd_idx = talk_data.num_listen_fds + talk_data.num_talk_fds; Peer *p = talk_data.peers + idx;
if (fd_idx < PEER_LIMIT && talk_data.num_writes < PEER_LIMIT) { if (p->read.finished && !p->num_of_unresponded_messages_sent_to_main_thread && !p->write.used) {
ensure_space_for(&talk_data, fds, PollFD, fd_idx + 1, fds_capacity, 8, false); free_peer(p);
talk_data.fds[fd_idx].fd = src->fd; talk_data.fds[fd_idx].events = POLLOUT; remove_i_from_array(talk_data.peers, idx, talk_data.num_peers);
ensure_space_for(&talk_data, writes, PeerWriteData, talk_data.num_writes + 1, writes_capacity, 8, false);
talk_data.writes[talk_data.num_writes++] = *src;
talk_data.num_talk_fds++;
} else {
log_error("Cannot send response to peer, too many peers");
free(src->data); nuke_socket(src->fd);
} }
*src = empty_pwd;
} }
} }
static void* static void*
talk_loop(void *data) { talk_loop(void *data) {
// The talk thread loop // The talk thread loop
ChildMonitor *self = (ChildMonitor*)data; ChildMonitor *self = (ChildMonitor*)data;
set_thread_name("KittyPeerMon"); set_thread_name("KittyPeerMon");
if ((pthread_mutex_init(&talk_data.peer_lock, NULL)) != 0) { perror("Failed to create peer mutex"); return 0; }
if (!init_loop_data(&talk_data.loop_data)) { log_error("Failed to create wakeup fd for talk thread with error: %s", strerror(errno)); } if (!init_loop_data(&talk_data.loop_data)) { log_error("Failed to create wakeup fd for talk thread with error: %s", strerror(errno)); }
ensure_space_for(&talk_data, fds, PollFD, 8, fds_capacity, 8, false); PollFD fds[PEER_LIMIT + 8] = {0};
size_t num_listen_fds = 0, num_peer_fds = 0;
#define add_listener(which) \ #define add_listener(which) \
if (self->which > -1) { \ if (self->which > -1) { \
talk_data.fds[talk_data.num_listen_fds].fd = self->which; talk_data.fds[talk_data.num_listen_fds++].events = POLLIN; \ fds[num_listen_fds].fd = self->which; fds[num_listen_fds++].events = POLLIN; \
} }
add_listener(talk_fd); add_listener(listen_fd); add_listener(talk_fd); add_listener(listen_fd);
#undef add_listener #undef add_listener
talk_data.fds[talk_data.num_listen_fds].fd = talk_data.loop_data.wakeup_read_fd; talk_data.fds[talk_data.num_listen_fds++].events = POLLIN; fds[num_listen_fds].fd = talk_data.loop_data.wakeup_read_fd; fds[num_listen_fds++].events = POLLIN;
while (LIKELY(!self->shutting_down)) { while (LIKELY(!self->shutting_down)) {
for (size_t i = 0; i < talk_data.num_listen_fds + talk_data.num_talk_fds; i++) { talk_data.fds[i].revents = 0; } num_peer_fds = 0;
int ret = poll(talk_data.fds, talk_data.num_listen_fds + talk_data.num_talk_fds, -1); if (talk_data.num_peers > 0) {
talk_mutex(unlock);
prune_peers();
for (size_t i = 0; i < talk_data.num_peers; i++) {
Peer *p = talk_data.peers + i;
if (!p->read.finished || p->write.used) {
fds[num_listen_fds + num_peer_fds].fd = p->fd;
fds[num_listen_fds + num_peer_fds].revents = 0;
int flags = 0;
if (!p->read.finished) flags |= POLLIN;
if (p->write.used) flags |= POLLOUT;
fds[num_listen_fds + num_peer_fds++].events = flags;
}
}
talk_mutex(unlock);
}
for (size_t i = 0; i < num_listen_fds; i++) fds[i].revents = 0;
int ret = poll(fds, num_listen_fds + num_peer_fds, -1);
if (ret > 0) { if (ret > 0) {
bool has_finished_reads = false, has_finished_writes = false; for (size_t i = 0; i < num_listen_fds - 1; i++) {
int invalid_fd = -1; if (fds[i].revents & POLLIN) {
for (size_t i = 0; i < talk_data.num_listen_fds - 1; i++) { if (!accept_peer(fds[i].fd, self->shutting_down)) goto end;
if (talk_data.fds[i].revents & POLLIN) {if (!accept_peer(talk_data.fds[i].fd, self->shutting_down)) goto end; } }
} }
if (talk_data.fds[talk_data.num_listen_fds - 1].revents & POLLIN) drain_fd(talk_data.fds[talk_data.num_listen_fds - 1].fd); // wakeup if (fds[num_listen_fds - 1].revents & POLLIN) {
for (size_t i = talk_data.num_listen_fds; i < talk_data.num_talk_fds + talk_data.num_listen_fds; i++) { drain_fd(fds[num_listen_fds - 1].fd); // wakeup
if (talk_data.fds[i].revents & (POLLIN | POLLHUP)) { if (read_from_peer(self, talk_data.fds[i].fd)) has_finished_reads = true; } }
if (talk_data.fds[i].revents & POLLOUT) { if (write_to_peer(talk_data.fds[i].fd)) has_finished_writes = true; } for (size_t i = num_listen_fds, k = 0; i < num_peer_fds + num_listen_fds; i++, k++) {
if (talk_data.fds[i].revents & POLLNVAL) invalid_fd = talk_data.fds[i].fd; Peer *p = talk_data.peers + k;
if (fds[i].revents & (POLLIN | POLLHUP)) read_from_peer(self, p);
if (fds[i].revents & POLLOUT) write_to_peer(p);
if (fds[i].revents & POLLNVAL) {
p->read.finished = true;
p->write.failed = true; p->write.used = 0;
}
} }
if (invalid_fd > -1) { prune_invalid_fd(invalid_fd); has_finished_reads = true; has_finished_writes = true; }
if (has_finished_reads) prune_finished_reads();
if (has_finished_writes) prune_finished_writes();
peer_mutex(lock);
if (talk_data.num_queued_writes) move_queued_writes();
peer_mutex(unlock);
} else if (ret < 0) { if (errno != EAGAIN && errno != EINTR) perror("poll() on talk fds failed"); } } else if (ret < 0) { if (errno != EAGAIN && errno != EINTR) perror("poll() on talk fds failed"); }
} }
end: end:
free_loop_data(&talk_data.loop_data); free_loop_data(&talk_data.loop_data);
free(talk_data.fds); free(talk_data.reads); free(talk_data.writes); free(talk_data.queued_writes); for (size_t i = 0; i < talk_data.num_peers; i++) free_peer(talk_data.peers + i);
free(talk_data.peers);
return 0; return 0;
} }
static inline bool
add_peer_writer(int fd, const char* msg, size_t msg_sz) {
bool ok = false;
peer_mutex(lock);
if (talk_data.num_queued_writes < PEER_LIMIT) {
ensure_space_for(&talk_data, queued_writes, PeerWriteData, talk_data.num_queued_writes + 1, queued_writes_capacity, 8, false);
talk_data.queued_writes[talk_data.num_queued_writes] = empty_pwd;
talk_data.queued_writes[talk_data.num_queued_writes].data = malloc(msg_sz);
if (talk_data.queued_writes[talk_data.num_queued_writes].data) {
memcpy(talk_data.queued_writes[talk_data.num_queued_writes].data, msg, msg_sz);
talk_data.queued_writes[talk_data.num_queued_writes].sz = msg_sz;
talk_data.queued_writes[talk_data.num_queued_writes++].fd = fd;
ok = true;
}
} else log_error("Cannot send response to peer, too many peers");
peer_mutex(unlock);
return ok;
}
static void static void
send_response(int fd, const char *msg, size_t msg_sz) { send_response(id_type peer_id, const char *msg, size_t msg_sz) {
if (msg == NULL) { shutdown(fd, SHUT_WR); safe_close(fd, __FILE__, __LINE__); return; } talk_mutex(lock);
if (!add_peer_writer(fd, msg, msg_sz)) { shutdown(fd, SHUT_WR); safe_close(fd, __FILE__, __LINE__); } for (size_t i = 0; i < talk_data.num_peers; i++) {
else wakeup_talk_loop(false); Peer *peer = talk_data.peers + i;
if (peer->id == peer_id) {
if (peer->num_of_unresponded_messages_sent_to_main_thread) peer->num_of_unresponded_messages_sent_to_main_thread--;
if (!peer->write.failed) {
if (peer->write.capacity - peer->write.used < msg_sz) {
void *data = realloc(peer->write.data, peer->write.capacity + msg_sz);
if (data) {
peer->write.data = data;
peer->write.capacity += msg_sz;
} else fatal("Out of memory");
}
memcpy(peer->write.data + peer->write.used, msg, msg_sz);
peer->write.used += msg_sz;
}
wakeup_talk_loop(false);
break;
}
}
talk_mutex(unlock);
} }
// }}} // }}}