mirrors/janet
1
0
Fork 0
mirror of https://github.com/janet-lang/janet synced 2024-11-29 03:19:54 +00:00
Code Issues 1 Releases Wiki Activity

Remove direct references to file descriptors.

Browse Source 
If a descriptor is freed by the Janet code, other
uses of that descriptor, say in the event loop, need
to know that it has been closed.
This commit is contained in:
Calvin Rose 2020-02-11 08:57:44 -06:00
parent 135aff9e17
commit 79bb9e54d5
2 changed files with 138 additions and 111 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

79
src/core/net.c
View File

@ -36,10 +36,12 @@
#include <netdb.h> #include <netdb.h>
/* /*
* Event loops * Streams
*/ */
#define JANET_STREAM_CLOSED 1 #define JANET_STREAM_CLOSED 1
#define JANET_STREAM_READABLE 2
#define JANET_STREAM_WRITABLE 4
typedef struct { typedef struct {
int fd; int fd;
@ -48,6 +50,8 @@ typedef struct {
static int janet_stream_close(void *p, size_t s); static int janet_stream_close(void *p, size_t s);
static int janet_stream_getter(void *p, size_t, Janet key);
static const JanetAbstractType StreamAT = { static const JanetAbstractType StreamAT = {
"core/stream", "core/stream",
janet_stream_close, janet_stream_close,
@ -64,22 +68,25 @@ static int janet_stream_close(void *p, size_t s) {
return 0; return 0;
} }
static JanetStream *make_stream(int fd) { static JanetStream *make_stream(int fd, int flags) {
JanetStream *stream = janet_abstract(&StreamAT, sizeof(JanetStream)); JanetStream *stream = janet_abstract(&StreamAT, sizeof(JanetStream));
int flags = fcntl(fd, F_GETFL, 0); fcntl(fd, F_SETFL, fcntl(fd, F_GETFL, 0) | O_NONBLOCK);
fcntl(fd, F_SETFL, flags | O_NONBLOCK);
stream->fd = fd; stream->fd = fd;
stream->flags = 0; stream->flags = flags;
return stream; return stream;
} }
/*
* Event loop
*/
/* This large struct describes a waiting file descriptor, as well /* This large struct describes a waiting file descriptor, as well
* as what to do when we get an event for it. It is a variant type, where * as what to do when we get an event for it. It is a variant type, where
* each variant implements a simple state machine. */ * each variant implements a simple state machine. */
typedef struct { typedef struct {
/* File descriptor to listen for events on. */ /* File descriptor to listen for events on. */
int fd; JanetStream *stream;
/* Fiber to resume when event finishes. Can be NULL, in which case, /* Fiber to resume when event finishes. Can be NULL, in which case,
* no fiber is resumed when event completes. */ * no fiber is resumed when event completes. */
@ -137,6 +144,7 @@ JANET_THREAD_LOCAL int janet_vm_loop_count;
void janet_net_markloop(void) { void janet_net_markloop(void) {
for (int i = 0; i < janet_vm_loop_count; i++) { for (int i = 0; i < janet_vm_loop_count; i++) {
JanetLoopFD lfd = janet_vm_loopfds[i]; JanetLoopFD lfd = janet_vm_loopfds[i];
janet_mark(janet_wrap_abstract(lfd.stream));
switch (lfd.event_type) { switch (lfd.event_type) {
default: default:
break; break;
@ -175,9 +183,15 @@ static int janet_loop_schedule(JanetLoopFD lfd) {
* we can separate out the polling logic */ * we can separate out the polling logic */
static size_t janet_loop_event(size_t index) { static size_t janet_loop_event(size_t index) {
JanetLoopFD *jlfd = janet_vm_loopfds + index; JanetLoopFD *jlfd = janet_vm_loopfds + index;
JanetStream *stream = jlfd->stream;
int fd = stream->fd;
int ret = 1; int ret = 1;
int should_resume = 0; int should_resume = 0;
Janet resumeval = janet_wrap_nil(); Janet resumeval = janet_wrap_nil();
if (stream->flags & JANET_STREAM_CLOSED) {
should_resume = 1;
ret = 0;
} else {
switch (jlfd->event_type) { switch (jlfd->event_type) {
case JLE_READ_CHUNK: case JLE_READ_CHUNK:
case JLE_READ_SOME: { case JLE_READ_SOME: {
@ -186,8 +200,13 @@ static size_t janet_loop_event(size_t index) {
janet_buffer_extra(buffer, bytes_left); janet_buffer_extra(buffer, bytes_left);
ssize_t nread; ssize_t nread;
errno = 0; errno = 0;
if (!(stream->flags & JANET_STREAM_READABLE)) {
should_resume = 1;
ret = 0;
break;
}
do { do {
nread = read(jlfd->fd, buffer->data + buffer->count, bytes_left); nread = read(fd, buffer->data + buffer->count, bytes_left);
} while (errno == EINTR); } while (errno == EINTR);
if (errno == EAGAIN || errno == EWOULDBLOCK) { if (errno == EAGAIN || errno == EWOULDBLOCK) {
ret = 1; ret = 1;
@ -214,10 +233,10 @@ static size_t janet_loop_event(size_t index) {
case JLE_READ_ACCEPT: { case JLE_READ_ACCEPT: {
char addr[256] = {0}; /* Just make sure it is large enough for largest address type */ char addr[256] = {0}; /* Just make sure it is large enough for largest address type */
socklen_t len = 0; socklen_t len = 0;
int connfd = accept(jlfd->fd, (void *) &addr, &len); int connfd = accept(fd, (void *) &addr, &len);
if (connfd >= 0) { if (connfd >= 0) {
/* Made a new connection socket */ /* Made a new connection socket */
JanetStream *stream = make_stream(connfd); JanetStream *stream = make_stream(connfd, JANET_STREAM_READABLE | JANET_STREAM_WRITABLE);
Janet streamv = janet_wrap_abstract(stream); Janet streamv = janet_wrap_abstract(stream);
JanetFunction *handler = jlfd->data.read_accept.handler; JanetFunction *handler = jlfd->data.read_accept.handler;
Janet out; Janet out;
@ -235,6 +254,11 @@ static size_t janet_loop_event(size_t index) {
case JLE_WRITE_FROM_STRINGLIKE: { case JLE_WRITE_FROM_STRINGLIKE: {
int32_t start, len; int32_t start, len;
const uint8_t *bytes; const uint8_t *bytes;
if (!(stream->flags & JANET_STREAM_WRITABLE)) {
should_resume = 1;
ret = 0;
break;
}
if (jlfd->event_type == JLE_WRITE_FROM_BUFFER) { if (jlfd->event_type == JLE_WRITE_FROM_BUFFER) {
JanetBuffer *buffer = jlfd->data.write_from_buffer.buf; JanetBuffer *buffer = jlfd->data.write_from_buffer.buf;
bytes = buffer->data; bytes = buffer->data;
@ -249,7 +273,7 @@ static size_t janet_loop_event(size_t index) {
int32_t nbytes = len - start; int32_t nbytes = len - start;
ssize_t nwrote; ssize_t nwrote;
do { do {
nwrote = write(jlfd->fd, bytes + start, nbytes); nwrote = write(fd, bytes + start, nbytes);
} while (nwrote == EINTR); } while (nwrote == EINTR);
if (nwrote > 0) { if (nwrote > 0) {
start += nwrote; start += nwrote;
@ -275,6 +299,7 @@ static size_t janet_loop_event(size_t index) {
break; break;
} }
} }
}
/* Resume a fiber for some events */ /* Resume a fiber for some events */
if (NULL != jlfd->fiber && should_resume) { if (NULL != jlfd->fiber && should_resume) {
@ -303,9 +328,10 @@ static void janet_loop1(void) {
int fd_max = 0; int fd_max = 0;
for (int i = 0; i < janet_vm_loop_count; i++) { for (int i = 0; i < janet_vm_loop_count; i++) {
JanetLoopFD *jlfd = janet_vm_loopfds + i; JanetLoopFD *jlfd = janet_vm_loopfds + i;
if (jlfd->fd > fd_max) fd_max = jlfd->fd; int fd = jlfd->stream->fd;
if (fd > fd_max) fd_max = fd;
fd_set *set = (jlfd->event_type <= JLE_READ_ACCEPT) ? &readfds : &writefds; fd_set *set = (jlfd->event_type <= JLE_READ_ACCEPT) ? &readfds : &writefds;
FD_SET(jlfd->fd, set); FD_SET(fd, set);
} }
/* Blocking call - we should add timeout functionality */ /* Blocking call - we should add timeout functionality */
@ -314,8 +340,9 @@ static void janet_loop1(void) {
/* Now handle all events */ /* Now handle all events */
for (int i = 0; i < janet_vm_loop_count;) { for (int i = 0; i < janet_vm_loop_count;) {
JanetLoopFD *jlfd = janet_vm_loopfds + i; JanetLoopFD *jlfd = janet_vm_loopfds + i;
int fd = jlfd->stream->fd;
fd_set *set = (jlfd->event_type <= JLE_READ_ACCEPT) ? &readfds : &writefds; fd_set *set = (jlfd->event_type <= JLE_READ_ACCEPT) ? &readfds : &writefds;
if (FD_ISSET(jlfd->fd, set)) { if (FD_ISSET(fd, set)) {
size_t delta = janet_loop_event(i); size_t delta = janet_loop_event(i);
i += delta; i += delta;
} else { } else {
@ -336,9 +363,9 @@ void janet_loop(void) {
#define JANET_SCHED_FSOME 1 #define JANET_SCHED_FSOME 1
JANET_NO_RETURN static void janet_sched_read(int fd, JanetBuffer *buf, int32_t nbytes, int flags) { JANET_NO_RETURN static void janet_sched_read(JanetStream *stream, JanetBuffer *buf, int32_t nbytes, int flags) {
JanetLoopFD lfd = {0}; JanetLoopFD lfd = {0};
lfd.fd = fd; lfd.stream = stream;
lfd.fiber = janet_root_fiber(); lfd.fiber = janet_root_fiber();
lfd.event_type = (flags & JANET_SCHED_FSOME) ? JLE_READ_SOME : JLE_READ_CHUNK; lfd.event_type = (flags & JANET_SCHED_FSOME) ? JLE_READ_SOME : JLE_READ_CHUNK;
lfd.data.read_chunk.buf = buf; lfd.data.read_chunk.buf = buf;
@ -347,9 +374,9 @@ JANET_NO_RETURN static void janet_sched_read(int fd, JanetBuffer *buf, int32_t n
janet_signalv(JANET_SIGNAL_USER9, janet_wrap_nil()); janet_signalv(JANET_SIGNAL_USER9, janet_wrap_nil());
} }
JANET_NO_RETURN static void janet_sched_write_buffer(int fd, JanetBuffer *buf) { JANET_NO_RETURN static void janet_sched_write_buffer(JanetStream *stream, JanetBuffer *buf) {
JanetLoopFD lfd = {0}; JanetLoopFD lfd = {0};
lfd.fd = fd; lfd.stream = stream;
lfd.fiber = janet_root_fiber(); lfd.fiber = janet_root_fiber();
lfd.event_type = JLE_WRITE_FROM_BUFFER; lfd.event_type = JLE_WRITE_FROM_BUFFER;
lfd.data.write_from_buffer.buf = buf; lfd.data.write_from_buffer.buf = buf;
@ -358,9 +385,9 @@ JANET_NO_RETURN static void janet_sched_write_buffer(int fd, JanetBuffer *buf) {
janet_signalv(JANET_SIGNAL_USER9, janet_wrap_nil()); janet_signalv(JANET_SIGNAL_USER9, janet_wrap_nil());
} }
JANET_NO_RETURN static void janet_sched_write_stringlike(int fd, const uint8_t *str) { JANET_NO_RETURN static void janet_sched_write_stringlike(JanetStream *stream, const uint8_t *str) {
JanetLoopFD lfd = {0}; JanetLoopFD lfd = {0};
lfd.fd = fd; lfd.stream = stream;
lfd.fiber = janet_root_fiber(); lfd.fiber = janet_root_fiber();
lfd.event_type = JLE_WRITE_FROM_STRINGLIKE; lfd.event_type = JLE_WRITE_FROM_STRINGLIKE;
lfd.data.write_from_stringlike.str = str; lfd.data.write_from_stringlike.str = str;
@ -413,7 +440,7 @@ static Janet cfun_net_connect(int32_t argc, Janet *argv) {
} }
/* Wrap socket in abstract type JanetStream */ /* Wrap socket in abstract type JanetStream */
JanetStream *stream = make_stream(sock); JanetStream *stream = make_stream(sock, JANET_STREAM_READABLE | JANET_STREAM_WRITABLE);
return janet_wrap_abstract(stream); return janet_wrap_abstract(stream);
} }
@ -454,12 +481,12 @@ static Janet cfun_net_server(int32_t argc, Janet *argv) {
/* Put sfd on our loop */ /* Put sfd on our loop */
JanetLoopFD lfd = {0}; JanetLoopFD lfd = {0};
lfd.fd = sfd; lfd.stream = make_stream(sfd, 0);
lfd.event_type = JLE_READ_ACCEPT; lfd.event_type = JLE_READ_ACCEPT;
lfd.data.read_accept.handler = fun; lfd.data.read_accept.handler = fun;
janet_loop_schedule(lfd); janet_loop_schedule(lfd);
return janet_wrap_nil(); return janet_wrap_abstract(lfd.stream);
} }
static Janet cfun_stream_read(int32_t argc, Janet *argv) { static Janet cfun_stream_read(int32_t argc, Janet *argv) {
@ -467,7 +494,7 @@ static Janet cfun_stream_read(int32_t argc, Janet *argv) {
JanetStream *stream = janet_getabstract(argv, 0, &StreamAT); JanetStream *stream = janet_getabstract(argv, 0, &StreamAT);
int32_t n = janet_getnat(argv, 1); int32_t n = janet_getnat(argv, 1);
JanetBuffer *buffer = janet_optbuffer(argv, argc, 2, 10); JanetBuffer *buffer = janet_optbuffer(argv, argc, 2, 10);
janet_sched_read(stream->fd, buffer, n, JANET_SCHED_FSOME); janet_sched_read(stream, buffer, n, JANET_SCHED_FSOME);
} }
static Janet cfun_stream_chunk(int32_t argc, Janet *argv) { static Janet cfun_stream_chunk(int32_t argc, Janet *argv) {
@ -475,7 +502,7 @@ static Janet cfun_stream_chunk(int32_t argc, Janet *argv) {
JanetStream *stream = janet_getabstract(argv, 0, &StreamAT); JanetStream *stream = janet_getabstract(argv, 0, &StreamAT);
int32_t n = janet_getnat(argv, 1); int32_t n = janet_getnat(argv, 1);
JanetBuffer *buffer = janet_optbuffer(argv, argc, 2, 10); JanetBuffer *buffer = janet_optbuffer(argv, argc, 2, 10);
janet_sched_read(stream->fd, buffer, n, 0); janet_sched_read(stream, buffer, n, 0);
} }
static Janet cfun_stream_close(int32_t argc, Janet *argv) { static Janet cfun_stream_close(int32_t argc, Janet *argv) {
@ -489,10 +516,10 @@ static Janet cfun_stream_write(int32_t argc, Janet *argv) {
janet_fixarity(argc, 2); janet_fixarity(argc, 2);
JanetStream *stream = janet_getabstract(argv, 0, &StreamAT); JanetStream *stream = janet_getabstract(argv, 0, &StreamAT);
if (janet_checktype(argv[1], JANET_BUFFER)) { if (janet_checktype(argv[1], JANET_BUFFER)) {
janet_sched_write_buffer(stream->fd, janet_getbuffer(argv, 1)); janet_sched_write_buffer(stream, janet_getbuffer(argv, 1));
} else { } else {
JanetByteView bytes = janet_getbytes(argv, 1); JanetByteView bytes = janet_getbytes(argv, 1);
janet_sched_write_stringlike(stream->fd, bytes.bytes); janet_sched_write_stringlike(stream, bytes.bytes);
} }
} }