mirrors/janet
1
0
Fork 0
mirror of https://github.com/janet-lang/janet synced 2025-02-02 18:29:10 +00:00
Code Issues 1 Releases Wiki Activity

Work on getting ev support on windows.

Browse Source
This commit is contained in:
Calvin Rose 2020-10-04 12:46:15 -05:00
parent 307c7e00e2
commit db67538311
4 changed files with 191 additions and 28 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

12
examples/evsleep.janet Normal file
View File

@ -0,0 +1,12 @@
(defn worker
"Run for a number of iterations."
[name iterations]
(for i 0 iterations
(ev/sleep 1)
(print "worker " name " iteration " i)))
(ev/call worker :a 10)
(ev/sleep 0.2)
(ev/call worker :b 5)
(ev/sleep 0.3)
(ev/call worker :c 12)

117
src/core/ev.c
View File

@ -33,6 +33,12 @@
/* Includes */ /* Includes */
#ifdef JANET_WINDOWS
#include <windows.h>
#else
#include <limits.h> #include <limits.h>
#include <errno.h> #include <errno.h>
#include <unistd.h> #include <unistd.h>
@ -50,6 +56,8 @@
#include <sys/timerfd.h> #include <sys/timerfd.h>
#endif #endif
#endif
/* General queue */ /* General queue */
/* Ring buffer for storing a list of fibers */ /* Ring buffer for storing a list of fibers */
@ -95,22 +103,22 @@ static int janet_q_push(JanetQueue *q, void *item, size_t itemsize) {
int32_t newhead = q->head + (newcap - q->capacity); int32_t newhead = q->head + (newcap - q->capacity);
size_t seg1 = (size_t)(q->capacity - q->head); size_t seg1 = (size_t)(q->capacity - q->head);
if (seg1 > 0) { if (seg1 > 0) {
memmove(q->data + (newhead * itemsize), memmove((char *) q->data + (newhead * itemsize),
q->data + (q->head * itemsize), (char *) q->data + (q->head * itemsize),
seg1 * itemsize); seg1 * itemsize);
} }
q->head = newhead; q->head = newhead;
} }
q->capacity = newcap; q->capacity = newcap;
} }
memcpy(q->data + itemsize * q->tail, item, itemsize); memcpy((char *) q->data + itemsize * q->tail, item, itemsize);
q->tail = q->tail + 1 < q->capacity ? q->tail + 1 : 0; q->tail = q->tail + 1 < q->capacity ? q->tail + 1 : 0;
return 0; return 0;
} }
static int janet_q_pop(JanetQueue *q, void *out, size_t itemsize) { static int janet_q_pop(JanetQueue *q, void *out, size_t itemsize) {
if (q->head == q->tail) return 1; if (q->head == q->tail) return 1;
memcpy(out, q->data + itemsize * q->head, itemsize); memcpy(out, (char *) q->data + itemsize * q->head, itemsize);
q->head = q->head + 1 < q->capacity ? q->head + 1 : 0; q->head = q->head + 1 < q->capacity ? q->head + 1 : 0;
return 0; return 0;
} }
@ -165,7 +173,7 @@ static void pop_timeout(size_t index) {
if (janet_vm_tq_count <= index) return; if (janet_vm_tq_count <= index) return;
janet_vm_tq[index].fiber->timeout_index = -1; janet_vm_tq[index].fiber->timeout_index = -1;
janet_vm_tq[index] = janet_vm_tq[--janet_vm_tq_count]; janet_vm_tq[index] = janet_vm_tq[--janet_vm_tq_count];
janet_vm_tq[index].fiber->timeout_index = index; janet_vm_tq[index].fiber->timeout_index = (int32_t) index;
for (;;) { for (;;) {
size_t left = (index << 1) + 1; size_t left = (index << 1) + 1;
size_t right = left + 1; size_t right = left + 1;
@ -180,8 +188,8 @@ static void pop_timeout(size_t index) {
JanetTimeout temp = janet_vm_tq[index]; JanetTimeout temp = janet_vm_tq[index];
janet_vm_tq[index] = janet_vm_tq[smallest]; janet_vm_tq[index] = janet_vm_tq[smallest];
janet_vm_tq[smallest] = temp; janet_vm_tq[smallest] = temp;
janet_vm_tq[index].fiber->timeout_index = index; janet_vm_tq[index].fiber->timeout_index = (int32_t) index;
janet_vm_tq[smallest].fiber->timeout_index = smallest; janet_vm_tq[smallest].fiber->timeout_index = (int32_t) smallest;
index = smallest; index = smallest;
} }
} }
@ -200,14 +208,14 @@ static void add_timeout(JanetTimeout to) {
janet_vm_tq_capacity = newcap; janet_vm_tq_capacity = newcap;
} }
/* Append */ /* Append */
janet_vm_tq_count = newcount; janet_vm_tq_count = (int32_t) newcount;
janet_vm_tq[oldcount] = to; janet_vm_tq[oldcount] = to;
/* Heapify */ /* Heapify */
size_t index = oldcount; size_t index = oldcount;
if (to.fiber->timeout_index >= 0) { if (to.fiber->timeout_index >= 0) {
pop_timeout(to.fiber->timeout_index); pop_timeout(to.fiber->timeout_index);
} }
to.fiber->timeout_index = index; to.fiber->timeout_index = (int32_t) index;
while (index > 0) { while (index > 0) {
size_t parent = (index - 1) >> 1; size_t parent = (index - 1) >> 1;
if (janet_vm_tq[parent].when <= janet_vm_tq[index].when) break; if (janet_vm_tq[parent].when <= janet_vm_tq[index].when) break;
@ -277,7 +285,7 @@ static void janet_unlisten_impl(JanetListenerState *state) {
} }
/* Call after creating a pollable */ /* Call after creating a pollable */
void janet_pollable_init(JanetPollable *pollable, JanetPollType handle) { void janet_pollable_init(JanetPollable *pollable, JanetHandle handle) {
pollable->handle = handle; pollable->handle = handle;
pollable->flags = 0; pollable->flags = 0;
pollable->state = NULL; pollable->state = NULL;
@ -625,7 +633,92 @@ void janet_loop(void) {
} }
} }
#ifdef JANET_EV_EPOLL #ifdef JANET_WINDOWS
/* Epoll global data */
JANET_THREAD_LOCAL HANDLE janet_vm_iocp = NULL;
static JanetTimestamp ts_now(void) {
FILETIME ftime;
GetSystemTimeAsFileTime(&ftime);
return (uint64_t)(ftime.dwLowDateTime) | ((uint64_t)(ftime.dwHighDateTime) << 32);
}
void janet_ev_init(void) {
janet_ev_init_common();
janet_vm_iocp = CreateIoCompletionPort(INVALID_HANDLE_VALUE, NULL, 0, 0);
if (NULL == janet_vm_iocp) janet_panic("could not create io completion port");
}
void janet_ev_deinit(void) {
janet_ev_deinit_common();
CloseHandle(janet_vm_iocp);
}
JanetListenerState *janet_listen(JanetPollable *pollable, JanetListener behavior, int mask, size_t size) {
JanetListenerState *state = janet_listen_impl(pollable, behavior, mask, size);
/* TODO - associate IO operation with listener state somehow
* maybe we could require encoding the operation in a mask. */
/* on windows, janet_listen does not actually start any listening behavior. */
return state;
}
static void janet_unlisten(JanetListenerState *state) {
/* We don't necessarily want to cancel all io on this pollable */
janet_unlisten_impl(state);
}
void janet_loop1_impl(void) {
JanetTimeout to;
memset(&to, 0, sizeof(to));
int has_timeout = peek_timeout(&to);
ULONG_PTR completionKey = 0;
DWORD num_bytes_transfered = 0;
LPOVERLAPPED overlapped;
/* Calculate how long to wait before timeout */
uint64_t waittime;
if (has_timeout) {
JanetTimestamp now = ts_now();
if (now > to.when) {
waittime = 0;
} else {
waittime = (uint64_t) (to.when - now);
}
} else {
waittime = INFINITE;
}
BOOL result = GetQueuedCompletionStatus(janet_vm_iocp, &num_bytes_transfered, &completionKey, &overlapped, (DWORD) waittime);
if (!result) {
/* timeout ? */
if (has_timeout) {
/* Timer event */
pop_timeout(0);
/* Cancel waiters for this fiber */
if (to.is_error) {
janet_cancel(to.fiber, janet_cstringv("timeout"));
} else {
janet_schedule(to.fiber, janet_wrap_nil());
}
} else {
JANET_EXIT("failed to get iocp GetQueuedCompletionStatus");
}
} else {
JANET_EXIT("Unexpected event");
/* Normal event */
JanetListenerState *state = (JanetListenerState *) completionKey;
state->event = overlapped;
JanetAsyncStatus status = state->machine(state, JANET_ASYNC_EVENT_COMPLETE);
if (status == JANET_ASYNC_STATUS_DONE)
janet_unlisten(state);
}
}
#elif defined(JANET_EV_POLL)
/* /*
* Start linux/epoll implementation * Start linux/epoll implementation
@ -734,6 +827,7 @@ void janet_loop1_impl(void) {
/* Normal event */ /* Normal event */
int mask = events[i].events; int mask = events[i].events;
JanetListenerState *state = pollable->state; JanetListenerState *state = pollable->state;
state->event = events + i;
while (NULL != state) { while (NULL != state) {
JanetListenerState *next_state = state->_next; JanetListenerState *next_state = state->_next;
JanetAsyncStatus status1 = JANET_ASYNC_STATUS_NOT_DONE; JanetAsyncStatus status1 = JANET_ASYNC_STATUS_NOT_DONE;
@ -877,6 +971,7 @@ void janet_loop1_impl(void) {
int mask = janet_vm_fds[i].revents; int mask = janet_vm_fds[i].revents;
JanetAsyncStatus status1 = JANET_ASYNC_STATUS_NOT_DONE; JanetAsyncStatus status1 = JANET_ASYNC_STATUS_NOT_DONE;
JanetAsyncStatus status2 = JANET_ASYNC_STATUS_NOT_DONE; JanetAsyncStatus status2 = JANET_ASYNC_STATUS_NOT_DONE;
state->event = pfd;
if (mask & POLLOUT) if (mask & POLLOUT)
status1 = state->machine(state, JANET_ASYNC_EVENT_WRITE); status1 = state->machine(state, JANET_ASYNC_EVENT_WRITE);
if (mask & POLLIN) if (mask & POLLIN)

71
src/core/net.c
View File

@ -75,7 +75,7 @@ static const JanetAbstractType AddressAT = {
}; };
#ifdef JANET_WINDOWS #ifdef JANET_WINDOWS
#define JSOCKCLOSE(x) closesocket(x) #define JSOCKCLOSE(x) closesocket((SOCKET) x)
#define JSOCKDEFAULT INVALID_SOCKET #define JSOCKDEFAULT INVALID_SOCKET
#define JLASTERR WSAGetLastError() #define JLASTERR WSAGetLastError()
#define JSOCKVALID(x) ((x) != INVALID_SOCKET) #define JSOCKVALID(x) ((x) != INVALID_SOCKET)
@ -89,7 +89,7 @@ static const JanetAbstractType AddressAT = {
static JanetStream *make_stream(SOCKET fd, uint32_t flags) { static JanetStream *make_stream(SOCKET fd, uint32_t flags) {
u_long iMode = 0; u_long iMode = 0;
JanetStream *stream = janet_abstract(&StreamAT, sizeof(JanetStream)); JanetStream *stream = janet_abstract(&StreamAT, sizeof(JanetStream));
janet_pollable_init(stream, fd); janet_pollable_init(stream, (JanetHandle) fd);
ioctlsocket(fd, FIONBIO, &iMode); ioctlsocket(fd, FIONBIO, &iMode);
stream->flags = flags; stream->flags = flags;
return stream; return stream;
@ -180,6 +180,16 @@ JanetAsyncStatus net_machine_read(JanetListenerState *s, JanetAsyncEvent event)
case JANET_ASYNC_EVENT_CLOSE: case JANET_ASYNC_EVENT_CLOSE:
janet_cancel(s->fiber, janet_cstringv("stream closed")); janet_cancel(s->fiber, janet_cstringv("stream closed"));
return JANET_ASYNC_STATUS_DONE; return JANET_ASYNC_STATUS_DONE;
#ifdef JANET_WINDOWS
case JANET_ASYNC_EVENT_INIT: {
/* Begin read */
}
break;
case JANET_ASYNC_EVENT_COMPLETE: {
/* Called when read finished */
}
break;
#else
case JANET_ASYNC_EVENT_READ: case JANET_ASYNC_EVENT_READ:
/* Read in bytes */ /* Read in bytes */
{ {
@ -233,6 +243,7 @@ JanetAsyncStatus net_machine_read(JanetListenerState *s, JanetAsyncEvent event)
} }
} }
break; break;
#endif
} }
return JANET_ASYNC_STATUS_NOT_DONE; return JANET_ASYNC_STATUS_NOT_DONE;
} }
@ -298,6 +309,17 @@ JanetAsyncStatus net_machine_write(JanetListenerState *s, JanetAsyncEvent event)
case JANET_ASYNC_EVENT_CLOSE: case JANET_ASYNC_EVENT_CLOSE:
janet_cancel(s->fiber, janet_cstringv("stream closed")); janet_cancel(s->fiber, janet_cstringv("stream closed"));
return JANET_ASYNC_STATUS_DONE; return JANET_ASYNC_STATUS_DONE;
#ifdef JANET_WINDOWS
case JANET_ASYNC_EVENT_INIT: {
/* Begin write */
}
break;
case JANET_ASYNC_EVENT_COMPLETE: {
/* Called when write finished */
}
break;
#else
case JANET_ASYNC_EVENT_WRITE: { case JANET_ASYNC_EVENT_WRITE: {
int32_t start, len; int32_t start, len;
const uint8_t *bytes; const uint8_t *bytes;
@ -342,6 +364,7 @@ JanetAsyncStatus net_machine_write(JanetListenerState *s, JanetAsyncEvent event)
break; break;
} }
break; break;
#endif
} }
return JANET_ASYNC_STATUS_NOT_DONE; return JANET_ASYNC_STATUS_NOT_DONE;
} }
@ -384,6 +407,9 @@ JanetAsyncStatus net_machine_simple_server(JanetListenerState *s, JanetAsyncEven
case JANET_ASYNC_EVENT_INIT: case JANET_ASYNC_EVENT_INIT:
/* We know the pollable will be a stream */ /* We know the pollable will be a stream */
janet_gcroot(janet_wrap_abstract(s->pollable)); janet_gcroot(janet_wrap_abstract(s->pollable));
#ifdef JANET_WINDOWS
/* requires some more setup code */
#endif
break; break;
case JANET_ASYNC_EVENT_MARK: case JANET_ASYNC_EVENT_MARK:
janet_mark(janet_wrap_function(state->function)); janet_mark(janet_wrap_function(state->function));
@ -391,6 +417,12 @@ JanetAsyncStatus net_machine_simple_server(JanetListenerState *s, JanetAsyncEven
case JANET_ASYNC_EVENT_CLOSE: case JANET_ASYNC_EVENT_CLOSE:
janet_gcunroot(janet_wrap_abstract(s->pollable)); janet_gcunroot(janet_wrap_abstract(s->pollable));
return JANET_ASYNC_STATUS_DONE; return JANET_ASYNC_STATUS_DONE;
#ifdef JANET_WINDOWS
case JANET_ASYNC_EVENT_COMPLETE: {
/* Called when ever we get an IOCP event */
}
break;
#else
case JANET_ASYNC_EVENT_READ: { case JANET_ASYNC_EVENT_READ: {
JSock connfd = accept(s->pollable->handle, NULL, NULL); JSock connfd = accept(s->pollable->handle, NULL, NULL);
if (JSOCKVALID(connfd)) { if (JSOCKVALID(connfd)) {
@ -403,6 +435,7 @@ JanetAsyncStatus net_machine_simple_server(JanetListenerState *s, JanetAsyncEven
} }
break; break;
} }
#endif
} }
return JANET_ASYNC_STATUS_NOT_DONE; return JANET_ASYNC_STATUS_NOT_DONE;
} }
@ -420,6 +453,16 @@ JanetAsyncStatus net_machine_accept(JanetListenerState *s, JanetAsyncEvent event
case JANET_ASYNC_EVENT_CLOSE: case JANET_ASYNC_EVENT_CLOSE:
janet_cancel(s->fiber, janet_cstringv("stream closed")); janet_cancel(s->fiber, janet_cstringv("stream closed"));
return JANET_ASYNC_STATUS_DONE; return JANET_ASYNC_STATUS_DONE;
#ifdef JANET_WINDOWS
case JANET_ASYNC_EVENT_INIT: {
}
break;
case JANET_ASYNC_EVENT_COMPLETE: {
}
break;
#else
case JANET_ASYNC_EVENT_READ: { case JANET_ASYNC_EVENT_READ: {
JSock connfd = accept(s->pollable->handle, NULL, NULL); JSock connfd = accept(s->pollable->handle, NULL, NULL);
if (JSOCKVALID(connfd)) { if (JSOCKVALID(connfd)) {
@ -431,6 +474,7 @@ JanetAsyncStatus net_machine_accept(JanetListenerState *s, JanetAsyncEvent event
} }
break; break;
} }
#endif
} }
return JANET_ASYNC_STATUS_NOT_DONE; return JANET_ASYNC_STATUS_NOT_DONE;
} }
@ -456,7 +500,8 @@ static int janet_get_sockettype(Janet *argv, int32_t argc, int32_t n) {
/* Needs argc >= offset + 2 */ /* Needs argc >= offset + 2 */
/* For unix paths, just rertuns a single sockaddr and sets *is_unix to 1, otherwise 0 */ /* For unix paths, just rertuns a single sockaddr and sets *is_unix to 1, otherwise 0 */
static struct addrinfo *janet_get_addrinfo(Janet *argv, int32_t offset, int socktype, int passive, int *is_unix) { static struct addrinfo *janet_get_addrinfo(Janet *argv, int32_t offset, int socktype, int passive, int *is_unix) {
/* Unix socket support */ /* Unix socket support - not yet supported on windows. */
#ifndef JANET_WINDOWS
if (janet_keyeq(argv[offset], "unix")) { if (janet_keyeq(argv[offset], "unix")) {
const char *path = janet_getcstring(argv, offset + 1); const char *path = janet_getcstring(argv, offset + 1);
struct sockaddr_un *saddr = malloc(sizeof(struct sockaddr_un)); struct sockaddr_un *saddr = malloc(sizeof(struct sockaddr_un));
@ -468,6 +513,7 @@ static struct addrinfo *janet_get_addrinfo(Janet *argv, int32_t offset, int sock
*is_unix = 1; *is_unix = 1;
return (struct addrinfo *) saddr; return (struct addrinfo *) saddr;
} }
#endif
/* Get host and port */ /* Get host and port */
const char *host = janet_getcstring(argv, offset); const char *host = janet_getcstring(argv, offset);
const char *port; const char *port;
@ -501,12 +547,15 @@ static Janet cfun_net_sockaddr(int32_t argc, Janet *argv) {
int is_unix = 0; int is_unix = 0;
int make_arr = (argc >= 3 && janet_truthy(argv[3])); int make_arr = (argc >= 3 && janet_truthy(argv[3]));
struct addrinfo *ai = janet_get_addrinfo(argv, 0, socktype, 0, &is_unix); struct addrinfo *ai = janet_get_addrinfo(argv, 0, socktype, 0, &is_unix);
#ifndef JANET_WINDOWS
/* no unix domain socket support on windows yet */
if (is_unix) { if (is_unix) {
void *abst = janet_abstract(&AddressAT, sizeof(struct sockaddr_un)); void *abst = janet_abstract(&AddressAT, sizeof(struct sockaddr_un));
memcpy(abst, ai, sizeof(struct sockaddr_un)); memcpy(abst, ai, sizeof(struct sockaddr_un));
Janet ret = janet_wrap_abstract(abst); Janet ret = janet_wrap_abstract(abst);
return make_arr ? janet_wrap_array(janet_array_n(&ret, 1)) : ret; return make_arr ? janet_wrap_array(janet_array_n(&ret, 1)) : ret;
} }
#endif
if (make_arr) { if (make_arr) {
/* Select all */ /* Select all */
JanetArray *arr = janet_array(10); JanetArray *arr = janet_array(10);
@ -542,6 +591,7 @@ static Janet cfun_net_connect(int32_t argc, Janet *argv) {
JSock sock = JSOCKDEFAULT; JSock sock = JSOCKDEFAULT;
void *addr = NULL; void *addr = NULL;
socklen_t addrlen; socklen_t addrlen;
#ifndef JANET_WINDOWS
if (is_unix) { if (is_unix) {
sock = socket(AF_UNIX, socktype | JSOCKFLAGS, 0); sock = socket(AF_UNIX, socktype | JSOCKFLAGS, 0);
if (!JSOCKVALID(sock)) { if (!JSOCKVALID(sock)) {
@ -549,13 +599,15 @@ static Janet cfun_net_connect(int32_t argc, Janet *argv) {
} }
addr = (void *) ai; addr = (void *) ai;
addrlen = sizeof(struct sockaddr_un); addrlen = sizeof(struct sockaddr_un);
} else { } else
#endif
{
struct addrinfo *rp = NULL; struct addrinfo *rp = NULL;
for (rp = ai; rp != NULL; rp = rp->ai_next) { for (rp = ai; rp != NULL; rp = rp->ai_next) {
sock = socket(rp->ai_family, rp->ai_socktype | JSOCKFLAGS, rp->ai_protocol); sock = socket(rp->ai_family, rp->ai_socktype | JSOCKFLAGS, rp->ai_protocol);
if (JSOCKVALID(sock)) { if (JSOCKVALID(sock)) {
addr = rp->ai_addr; addr = rp->ai_addr;
addrlen = rp->ai_addrlen; addrlen = (socklen_t) rp->ai_addrlen;
break; break;
} }
} }
@ -610,6 +662,7 @@ static Janet cfun_net_server(int32_t argc, Janet *argv) {
struct addrinfo *ai = janet_get_addrinfo(argv, 0, socktype, 1, &is_unix); struct addrinfo *ai = janet_get_addrinfo(argv, 0, socktype, 1, &is_unix);
JSock sfd = JSOCKDEFAULT; JSock sfd = JSOCKDEFAULT;
#ifndef JANET_WINDOWS
if (is_unix) { if (is_unix) {
sfd = socket(AF_UNIX, socktype | JSOCKFLAGS, 0); sfd = socket(AF_UNIX, socktype | JSOCKFLAGS, 0);
if (!JSOCKVALID(sfd)) { if (!JSOCKVALID(sfd)) {
@ -623,7 +676,9 @@ static Janet cfun_net_server(int32_t argc, Janet *argv) {
janet_panic(err ? err : "could not bind socket"); janet_panic(err ? err : "could not bind socket");
} }
free(ai); free(ai);
} else { } else
#endif
{
/* Check all addrinfos in a loop for the first that we can bind to. */ /* Check all addrinfos in a loop for the first that we can bind to. */
struct addrinfo *rp = NULL; struct addrinfo *rp = NULL;
for (rp = ai; rp != NULL; rp = rp->ai_next) { for (rp = ai; rp != NULL; rp = rp->ai_next) {
@ -779,9 +834,9 @@ static Janet cfun_stream_flush(int32_t argc, Janet *argv) {
check_stream_flag(stream, JANET_STREAM_WRITABLE); check_stream_flag(stream, JANET_STREAM_WRITABLE);
/* Toggle no delay flag */ /* Toggle no delay flag */
int flag = 1; int flag = 1;
setsockopt(stream->handle, IPPROTO_TCP, TCP_NODELAY, (char *) &flag, sizeof(int)); setsockopt((JSock) stream->handle, IPPROTO_TCP, TCP_NODELAY, (char *) &flag, sizeof(int));
flag = 0; flag = 0;
setsockopt(stream->handle, IPPROTO_TCP, TCP_NODELAY, (char *) &flag, sizeof(int)); setsockopt((JSock) stream->handle, IPPROTO_TCP, TCP_NODELAY, (char *) &flag, sizeof(int));
return argv[0]; return argv[0];
} }

19
src/include/janet.h
View File

@ -509,7 +509,8 @@ typedef enum {
JANET_ASYNC_EVENT_CLOSE, JANET_ASYNC_EVENT_CLOSE,
JANET_ASYNC_EVENT_READ, JANET_ASYNC_EVENT_READ,
JANET_ASYNC_EVENT_WRITE, JANET_ASYNC_EVENT_WRITE,
JANET_ASYNC_EVENT_TIMEOUT JANET_ASYNC_EVENT_TIMEOUT,
JANET_ASYNC_EVENT_COMPLETE /* Used on windows for IOCP */
} JanetAsyncEvent; } JanetAsyncEvent;
#define JANET_ASYNC_LISTEN_READ (1 << JANET_ASYNC_EVENT_READ) #define JANET_ASYNC_LISTEN_READ (1 << JANET_ASYNC_EVENT_READ)
@ -522,21 +523,19 @@ typedef enum {
} JanetAsyncStatus; } JanetAsyncStatus;
/* Typedefs */ /* Typedefs */
#ifdef JANET_WINDOWS
typedef HANDLE JanetPollType;
#else
typedef int JanetPollType;
#endif
typedef struct JanetListenerState JanetListenerState; typedef struct JanetListenerState JanetListenerState;
typedef struct JanetPollable JanetPollable; typedef struct JanetPollable JanetPollable;
typedef JanetAsyncStatus(*JanetListener)(JanetListenerState *state, JanetAsyncEvent event); typedef JanetAsyncStatus(*JanetListener)(JanetListenerState *state, JanetAsyncEvent event);
/* Wrapper around file descriptors and HANDLEs that can be polled. */ /* Wrapper around file descriptors and HANDLEs that can be polled. */
struct JanetPollable { struct JanetPollable {
JanetPollType handle; JanetHandle handle;
uint32_t flags; uint32_t flags;
/* Linked list of all in-flight IO routines for this pollable */
JanetListenerState *state; JanetListenerState *state;
/* internal */ /* internal - used to disallow multiple concurrent reads / writes on the same pollable.
* this constraint may be lifted later but allowing such would require more internal book keeping
* for some implementations. You can read and write at the same time on the same pollable, though. */
int _mask; int _mask;
}; };
@ -545,6 +544,8 @@ struct JanetListenerState {
JanetListener machine; JanetListener machine;
JanetFiber *fiber; JanetFiber *fiber;
JanetPollable *pollable; JanetPollable *pollable;
void *event; /* Used to pass data from asynchronous IO event. Contents depend on both
implementation of the event loop and the particular event. */
/* internal */ /* internal */
int _index; /* not used in all implementations */ int _index; /* not used in all implementations */
int _mask; int _mask;
@ -1246,7 +1247,7 @@ extern enum JanetInstructionType janet_instructions[JOP_INSTRUCTION_COUNT];
JANET_API void janet_loop(void); JANET_API void janet_loop(void);
/* Wrapper around pollables */ /* Wrapper around pollables */
JANET_API void janet_pollable_init(JanetPollable *pollable, JanetPollType handle); JANET_API void janet_pollable_init(JanetPollable *pollable, JanetHandle handle);
JANET_API void janet_pollable_mark(JanetPollable *pollable); JANET_API void janet_pollable_mark(JanetPollable *pollable);
JANET_API void janet_pollable_deinit(JanetPollable *pollable); JANET_API void janet_pollable_deinit(JanetPollable *pollable);