/* * Copyright (c) 2020 Calvin Rose * * Permission is hereby granted, free of charge, to any person obtaining a copy * of this software and associated documentation files (the "Software"), to * deal in the Software without restriction, including without limitation the * rights to use, copy, modify, merge, publish, distribute, sublicense, and/or * sell copies of the Software, and to permit persons to whom the Software is * furnished to do so, subject to the following conditions: * * The above copyright notice and this permission notice shall be included in * all copies or substantial portions of the Software. * * THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR * IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, * FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE * AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER * LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING * FROM, OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS * IN THE SOFTWARE. */ #ifndef JANET_AMALG #include "features.h" #include #include "util.h" #include "gc.h" #include "state.h" #include "fiber.h" #endif #ifdef JANET_EV /* Includes */ #ifdef JANET_WINDOWS #include #include #include #else #include #include #include #include #include #include #include #include #include #include #include #ifdef JANET_EV_EPOLL #include #include #endif #endif /* Ring buffer for storing a list of fibers */ typedef struct { int32_t capacity; int32_t head; int32_t tail; void *data; } JanetQueue; #define JANET_MAX_Q_CAPACITY 0x7FFFFFF static void janet_q_init(JanetQueue *q) { q->data = NULL; q->head = 0; q->tail = 0; q->capacity = 0; } static void janet_q_deinit(JanetQueue *q) { free(q->data); } static int32_t janet_q_count(JanetQueue *q) { return (q->head > q->tail) ? (q->tail + q->capacity - q->head) : (q->tail - q->head); } static int janet_q_push(JanetQueue *q, void *item, size_t itemsize) { int32_t count = janet_q_count(q); /* Resize if needed */ if (count + 1 >= q->capacity) { if (count + 1 >= JANET_MAX_Q_CAPACITY) return 1; int32_t newcap = (count + 2) * 2; if (newcap > JANET_MAX_Q_CAPACITY) newcap = JANET_MAX_Q_CAPACITY; q->data = realloc(q->data, itemsize * newcap); if (NULL == q->data) { JANET_OUT_OF_MEMORY; } if (q->head > q->tail) { /* Two segments, fix 2nd seg. */ int32_t newhead = q->head + (newcap - q->capacity); size_t seg1 = (size_t)(q->capacity - q->head); if (seg1 > 0) { memmove((char *) q->data + (newhead * itemsize), (char *) q->data + (q->head * itemsize), seg1 * itemsize); } q->head = newhead; } q->capacity = newcap; } memcpy((char *) q->data + itemsize * q->tail, item, itemsize); q->tail = q->tail + 1 < q->capacity ? q->tail + 1 : 0; return 0; } static int janet_q_pop(JanetQueue *q, void *out, size_t itemsize) { if (q->head == q->tail) return 1; memcpy(out, (char *) q->data + itemsize * q->head, itemsize); q->head = q->head + 1 < q->capacity ? q->head + 1 : 0; return 0; } /* New fibers to spawn or resume */ typedef struct JanetTask JanetTask; struct JanetTask { JanetFiber *fiber; Janet value; JanetSignal sig; }; /* Min priority queue of timestamps for timeouts. */ typedef int64_t JanetTimestamp; typedef struct JanetTimeout JanetTimeout; struct JanetTimeout { JanetTimestamp when; JanetFiber *fiber; uint32_t sched_id; int is_error; }; /* Forward declaration */ static void janet_unlisten(JanetListenerState *state); /* Global data */ JANET_THREAD_LOCAL size_t janet_vm_active_listeners = 0; JANET_THREAD_LOCAL size_t janet_vm_tq_count = 0; JANET_THREAD_LOCAL size_t janet_vm_tq_capacity = 0; JANET_THREAD_LOCAL JanetQueue janet_vm_spawn; JANET_THREAD_LOCAL JanetTimeout *janet_vm_tq = NULL; JANET_THREAD_LOCAL JanetRNG janet_vm_ev_rng; /* Get current timestamp (millisecond precision) */ static JanetTimestamp ts_now(void); /* Get current timestamp + an interval (millisecond precision) */ static JanetTimestamp ts_delta(JanetTimestamp ts, double delta) { ts += (int64_t)round(delta * 1000); return ts; } /* Look at the next timeout value without * removing it. */ static int peek_timeout(JanetTimeout *out) { if (janet_vm_tq_count == 0) return 0; *out = janet_vm_tq[0]; return 1; } /* Remove the next timeout from the priority queue */ static void pop_timeout(size_t index) { if (janet_vm_tq_count <= index) return; janet_vm_tq[index] = janet_vm_tq[--janet_vm_tq_count]; for (;;) { size_t left = (index << 1) + 1; size_t right = left + 1; size_t smallest = index; if (left < janet_vm_tq_count && (janet_vm_tq[left].when < janet_vm_tq[smallest].when)) smallest = left; if (right < janet_vm_tq_count && (janet_vm_tq[right].when < janet_vm_tq[smallest].when)) smallest = right; if (smallest == index) return; JanetTimeout temp = janet_vm_tq[index]; janet_vm_tq[index] = janet_vm_tq[smallest]; janet_vm_tq[smallest] = temp; index = smallest; } } /* Add a timeout to the timeout min heap */ static void add_timeout(JanetTimeout to) { size_t oldcount = janet_vm_tq_count; size_t newcount = oldcount + 1; if (newcount > janet_vm_tq_capacity) { size_t newcap = 2 * newcount; JanetTimeout *tq = realloc(janet_vm_tq, newcap * sizeof(JanetTimeout)); if (NULL == tq) { JANET_OUT_OF_MEMORY; } janet_vm_tq = tq; janet_vm_tq_capacity = newcap; } /* Append */ janet_vm_tq_count = (int32_t) newcount; janet_vm_tq[oldcount] = to; /* Heapify */ size_t index = oldcount; while (index > 0) { size_t parent = (index - 1) >> 1; if (janet_vm_tq[parent].when <= janet_vm_tq[index].when) break; /* Swap */ JanetTimeout tmp = janet_vm_tq[index]; janet_vm_tq[index] = janet_vm_tq[parent]; janet_vm_tq[parent] = tmp; /* Next */ index = parent; } } /* Create a new event listener */ static JanetListenerState *janet_listen_impl(JanetStream *stream, JanetListener behavior, int mask, size_t size, void *user) { if (stream->_mask & mask) { janet_panic("cannot listen for duplicate event on stream"); } if (janet_vm_root_fiber->waiting != NULL) { janet_panic("current fiber is already waiting for event"); } if (size < sizeof(JanetListenerState)) size = sizeof(JanetListenerState); JanetListenerState *state = malloc(size); if (NULL == state) { JANET_OUT_OF_MEMORY; } state->machine = behavior; state->fiber = janet_vm_root_fiber; janet_vm_root_fiber->waiting = state; state->stream = stream; state->_mask = mask; state->_index = 0; stream->_mask |= mask; janet_vm_active_listeners++; /* Prepend to linked list */ state->_next = stream->state; stream->state = state; /* Emit INIT event for convenience */ state->event = user; state->machine(state, JANET_ASYNC_EVENT_INIT); return state; } /* Indicate we are no longer listening for an event. This * frees the memory of the state machine as well. */ static void janet_unlisten_impl(JanetListenerState *state) { state->machine(state, JANET_ASYNC_EVENT_DEINIT); /* Remove state machine from poll list */ JanetListenerState **iter = &(state->stream->state); while (*iter && *iter != state) iter = &((*iter)->_next); janet_assert(*iter, "failed to remove listener"); *iter = state->_next; janet_vm_active_listeners--; /* Remove mask */ state->stream->_mask &= ~(state->_mask); /* Ensure fiber does not reference this state */ JanetFiber *fiber = state->fiber; if (NULL != fiber && fiber->waiting == state) { fiber->waiting = NULL; } free(state); } static const JanetMethod ev_default_stream_methods[] = { {"close", janet_cfun_stream_close}, {"read", janet_cfun_stream_read}, {"chunk", janet_cfun_stream_chunk}, {"write", janet_cfun_stream_write}, {NULL, NULL} }; /* Create a stream*/ JanetStream *janet_stream(JanetHandle handle, uint32_t flags, const JanetMethod *methods) { JanetStream *stream = janet_abstract(&janet_stream_type, sizeof(JanetStream)); stream->handle = handle; stream->flags = flags; stream->state = NULL; stream->_mask = 0; if (methods == NULL) methods = ev_default_stream_methods; stream->methods = methods; #ifdef JANET_NET if (flags & JANET_STREAM_SOCKET) { #ifdef JANET_WINDOWS u_long iMode = 0; ioctlsocket(handle, FIONBIO, &iMode); #else #if !defined(SOCK_CLOEXEC) && defined(O_CLOEXEC) int extra = O_CLOEXEC; #else int extra = 0; #endif fcntl(handle, F_SETFL, fcntl(handle, F_GETFL, 0) | O_NONBLOCK | extra); #endif } #endif return stream; } /* Called to clean up a stream */ static int janet_stream_gc(void *p, size_t s) { (void) s; JanetStream *stream = (JanetStream *)p; janet_stream_close(stream); return 0; } /* Close a stream */ void janet_stream_close(JanetStream *stream) { if (stream->flags & JANET_STREAM_CLOSED) return; JanetListenerState *state = stream->state; while (NULL != state) { state->machine(state, JANET_ASYNC_EVENT_CLOSE); JanetListenerState *next_state = state->_next; janet_unlisten(state); state = next_state; } stream->state = NULL; stream->flags |= JANET_STREAM_CLOSED; #ifdef JANET_WINDOWS #ifdef JANET_NET if (stream->flags & JANET_STREAM_SOCKET) { closesocket((SOCKET) stream->handle); } else #endif { CloseHandle(stream->handle); } #else close(stream->handle); #endif } /* Mark a stream for GC */ static int janet_stream_mark(void *p, size_t s) { (void) s; JanetStream *stream = (JanetStream *) p; JanetListenerState *state = stream->state; while (NULL != state) { if (NULL != state->fiber) { janet_mark(janet_wrap_fiber(state->fiber)); } (state->machine)(state, JANET_ASYNC_EVENT_MARK); state = state->_next; } return 0; } static int janet_stream_getter(void *p, Janet key, Janet *out) { JanetStream *stream = (JanetStream *)p; if (!janet_checktype(key, JANET_KEYWORD)) return 0; const JanetMethod *stream_methods = stream->methods; return janet_getmethod(janet_unwrap_keyword(key), stream_methods, out); return 0; } const JanetAbstractType janet_stream_type = { "core/stream", janet_stream_gc, janet_stream_mark, janet_stream_getter, JANET_ATEND_GET }; /* Register a fiber to resume with value */ void janet_schedule_signal(JanetFiber *fiber, Janet value, JanetSignal sig) { if (fiber->flags & JANET_FIBER_FLAG_SCHEDULED) return; fiber->flags |= JANET_FIBER_FLAG_SCHEDULED; fiber->sched_id++; JanetTask t = { fiber, value, sig }; janet_q_push(&janet_vm_spawn, &t, sizeof(t)); } void janet_cancel(JanetFiber *fiber, Janet value) { janet_schedule_signal(fiber, value, JANET_SIGNAL_ERROR); } void janet_schedule(JanetFiber *fiber, Janet value) { janet_schedule_signal(fiber, value, JANET_SIGNAL_OK); } void janet_fiber_did_resume(JanetFiber *fiber) { /* Cancel any pending fibers */ if (fiber->waiting) janet_unlisten(fiber->waiting); } /* Mark all pending tasks */ void janet_ev_mark(void) { JanetTask *tasks = janet_vm_spawn.data; if (janet_vm_spawn.head <= janet_vm_spawn.tail) { for (int32_t i = janet_vm_spawn.head; i < janet_vm_spawn.tail; i++) { janet_mark(janet_wrap_fiber(tasks[i].fiber)); janet_mark(tasks[i].value); } } else { for (int32_t i = janet_vm_spawn.head; i < janet_vm_spawn.capacity; i++) { janet_mark(janet_wrap_fiber(tasks[i].fiber)); janet_mark(tasks[i].value); } for (int32_t i = 0; i < janet_vm_spawn.tail; i++) { janet_mark(janet_wrap_fiber(tasks[i].fiber)); janet_mark(tasks[i].value); } } for (size_t i = 0; i < janet_vm_tq_count; i++) { janet_mark(janet_wrap_fiber(janet_vm_tq[i].fiber)); } } /* Run a top level task */ static void run_one(JanetFiber *fiber, Janet value, JanetSignal sigin) { fiber->flags &= ~JANET_FIBER_FLAG_SCHEDULED; Janet res; JanetSignal sig = janet_continue_signal(fiber, value, &res, sigin); if (sig != JANET_SIGNAL_OK && sig != JANET_SIGNAL_EVENT) { janet_stacktrace(fiber, res); } } /* Common init code */ void janet_ev_init_common(void) { janet_q_init(&janet_vm_spawn); janet_vm_active_listeners = 0; janet_vm_tq = NULL; janet_vm_tq_count = 0; janet_vm_tq_capacity = 0; janet_rng_seed(&janet_vm_ev_rng, 0); } /* Common deinit code */ void janet_ev_deinit_common(void) { janet_q_deinit(&janet_vm_spawn); } /* Short hand to yield to event loop */ void janet_await(void) { janet_signalv(JANET_SIGNAL_EVENT, janet_wrap_nil()); } /* Set timeout for the current root fiber */ void janet_addtimeout(double sec) { JanetFiber *fiber = janet_vm_root_fiber; JanetTimeout to; to.when = ts_delta(ts_now(), sec); to.fiber = fiber; to.sched_id = fiber->sched_id; to.is_error = 1; add_timeout(to); } /* Channels */ typedef struct { JanetFiber *fiber; uint32_t sched_id; enum { JANET_CP_MODE_ITEM, JANET_CP_MODE_CHOICE_READ, JANET_CP_MODE_CHOICE_WRITE } mode; } JanetChannelPending; typedef struct { JanetQueue items; JanetQueue read_pending; JanetQueue write_pending; int32_t limit; } JanetChannel; #define JANET_MAX_CHANNEL_CAPACITY 0xFFFFFF static void janet_chan_init(JanetChannel *chan, int32_t limit) { chan->limit = limit; janet_q_init(&chan->items); janet_q_init(&chan->read_pending); janet_q_init(&chan->write_pending); } static void janet_chan_deinit(JanetChannel *chan) { janet_q_deinit(&chan->read_pending); janet_q_deinit(&chan->write_pending); janet_q_deinit(&chan->items); } /* * Janet Channel abstract type */ /*static int janet_chanat_get(void *p, Janet key, Janet *out);*/ static int janet_chanat_mark(void *p, size_t s); static int janet_chanat_gc(void *p, size_t s); static const JanetAbstractType ChannelAT = { "core/channel", janet_chanat_gc, janet_chanat_mark, NULL, /* janet_chanat_get */ JANET_ATEND_GET }; static int janet_chanat_gc(void *p, size_t s) { (void) s; JanetChannel *channel = p; janet_chan_deinit(channel); return 0; } static void janet_chanat_mark_fq(JanetQueue *fq) { JanetChannelPending *pending = fq->data; if (fq->head <= fq->tail) { for (int32_t i = fq->head; i < fq->tail; i++) janet_mark(janet_wrap_fiber(pending[i].fiber)); } else { for (int32_t i = fq->head; i < fq->capacity; i++) janet_mark(janet_wrap_fiber(pending[i].fiber)); for (int32_t i = 0; i < fq->tail; i++) janet_mark(janet_wrap_fiber(pending[i].fiber)); } } static int janet_chanat_mark(void *p, size_t s) { (void) s; JanetChannel *chan = p; janet_chanat_mark_fq(&chan->read_pending); janet_chanat_mark_fq(&chan->write_pending); JanetQueue *items = &chan->items; Janet *data = chan->items.data; if (items->head <= items->tail) { for (int32_t i = items->head; i < items->tail; i++) janet_mark(data[i]); } else { for (int32_t i = items->head; i < items->capacity; i++) janet_mark(data[i]); for (int32_t i = 0; i < items->tail; i++) janet_mark(data[i]); } return 0; } static Janet make_write_result(JanetChannel *channel) { Janet *tup = janet_tuple_begin(2); tup[0] = janet_ckeywordv("give"); tup[1] = janet_wrap_abstract(channel); return janet_wrap_tuple(janet_tuple_end(tup)); } static Janet make_read_result(JanetChannel *channel, Janet x) { Janet *tup = janet_tuple_begin(3); tup[0] = janet_ckeywordv("take"); tup[1] = janet_wrap_abstract(channel); tup[2] = x; return janet_wrap_tuple(janet_tuple_end(tup)); } /* Push a value to a channel, and return 1 if channel should block, zero otherwise. * If the push would block, will add to the write_pending queue in the channel. */ static int janet_channel_push(JanetChannel *channel, Janet x, int is_choice) { JanetChannelPending reader; int is_empty; do { is_empty = janet_q_pop(&channel->read_pending, &reader, sizeof(reader)); } while (!is_empty && (reader.sched_id != reader.fiber->sched_id)); if (is_empty) { /* No pending reader */ if (janet_q_push(&channel->items, &x, sizeof(Janet))) { janet_panicf("channel overflow: %v", x); } else if (janet_q_count(&channel->items) > channel->limit) { /* Pushed successfully, but should block. */ JanetChannelPending pending; pending.fiber = janet_vm_root_fiber, pending.sched_id = janet_vm_root_fiber->sched_id, pending.mode = is_choice ? JANET_CP_MODE_CHOICE_WRITE : JANET_CP_MODE_ITEM; janet_q_push(&channel->write_pending, &pending, sizeof(pending)); return 1; } } else { /* Pending reader */ if (reader.mode == JANET_CP_MODE_CHOICE_READ) { janet_schedule(reader.fiber, make_read_result(channel, x)); } else { janet_schedule(reader.fiber, x); } } return 0; } /* Pop from a channel - returns 1 if item was obtain, 0 otherwise. The item * is returned by reference. If the pop would block, will add to the read_pending * queue in the channel. */ static int janet_channel_pop(JanetChannel *channel, Janet *item, int is_choice) { JanetChannelPending writer; if (janet_q_pop(&channel->items, item, sizeof(Janet))) { /* Queue empty */ JanetChannelPending pending; pending.fiber = janet_vm_root_fiber, pending.sched_id = janet_vm_root_fiber->sched_id; pending.mode = is_choice ? JANET_CP_MODE_CHOICE_READ : JANET_CP_MODE_ITEM; janet_q_push(&channel->read_pending, &pending, sizeof(pending)); return 0; } if (!janet_q_pop(&channel->write_pending, &writer, sizeof(writer))) { /* pending writer */ if (writer.mode == JANET_CP_MODE_CHOICE_WRITE) { janet_schedule(writer.fiber, make_write_result(channel)); } else { janet_schedule(writer.fiber, janet_wrap_abstract(channel)); } } return 1; } /* Channel Methods */ static Janet cfun_channel_push(int32_t argc, Janet *argv) { janet_fixarity(argc, 2); JanetChannel *channel = janet_getabstract(argv, 0, &ChannelAT); if (janet_channel_push(channel, argv[1], 0)) { janet_await(); } return argv[0]; } static Janet cfun_channel_pop(int32_t argc, Janet *argv) { janet_fixarity(argc, 1); JanetChannel *channel = janet_getabstract(argv, 0, &ChannelAT); Janet item; if (janet_channel_pop(channel, &item, 0)) { janet_schedule(janet_vm_root_fiber, item); } janet_await(); } static Janet cfun_channel_choice(int32_t argc, Janet *argv) { janet_arity(argc, 1, -1); int32_t len; const Janet *data; /* Check channels for immediate reads and writes */ for (int32_t i = 0; i < argc; i++) { if (janet_indexed_view(argv[i], &data, &len) && len == 2) { /* Write */ JanetChannel *chan = janet_getabstract(data, 0, &ChannelAT); if (janet_q_count(&chan->items) < chan->limit) { janet_channel_push(chan, data[1], 1); return make_write_result(chan); } } else { /* Read */ JanetChannel *chan = janet_getabstract(argv, i, &ChannelAT); if (chan->items.head != chan->items.tail) { Janet item; janet_channel_pop(chan, &item, 1); return make_read_result(chan, item); } } } /* Wait for all readers or writers */ for (int32_t i = 0; i < argc; i++) { if (janet_indexed_view(argv[i], &data, &len) && len == 2) { /* Write */ JanetChannel *chan = janet_getabstract(data, 0, &ChannelAT); janet_channel_push(chan, data[1], 1); } else { /* Read */ Janet item; JanetChannel *chan = janet_getabstract(argv, i, &ChannelAT); janet_channel_pop(chan, &item, 1); } } janet_await(); } static Janet cfun_channel_full(int32_t argc, Janet *argv) { janet_fixarity(argc, 1); JanetChannel *channel = janet_getabstract(argv, 0, &ChannelAT); return janet_wrap_boolean(janet_q_count(&channel->items) >= channel->limit); } static Janet cfun_channel_capacity(int32_t argc, Janet *argv) { janet_fixarity(argc, 1); JanetChannel *channel = janet_getabstract(argv, 0, &ChannelAT); return janet_wrap_integer(channel->limit); } static Janet cfun_channel_count(int32_t argc, Janet *argv) { janet_fixarity(argc, 1); JanetChannel *channel = janet_getabstract(argv, 0, &ChannelAT); return janet_wrap_integer(janet_q_count(&channel->items)); } /* Fisher yates shuffle of arguments to get fairness */ static void fisher_yates_args(int32_t argc, Janet *argv) { for (int32_t i = argc; i > 1; i--) { int32_t swap_index = janet_rng_u32(&janet_vm_ev_rng) % i; Janet temp = argv[swap_index]; argv[swap_index] = argv[i - 1]; argv[i - 1] = temp; } } static Janet cfun_channel_rchoice(int32_t argc, Janet *argv) { fisher_yates_args(argc, argv); return cfun_channel_choice(argc, argv); } static Janet cfun_channel_new(int32_t argc, Janet *argv) { janet_arity(argc, 0, 1); int32_t limit = janet_optnat(argv, argc, 0, 0); JanetChannel *channel = janet_abstract(&ChannelAT, sizeof(JanetChannel)); janet_chan_init(channel, limit); return janet_wrap_abstract(channel); } /* Main event loop */ void janet_loop1_impl(int has_timeout, JanetTimestamp timeout); void janet_loop1(void) { /* Schedule expired timers */ JanetTimeout to; JanetTimestamp now = ts_now(); while (peek_timeout(&to) && to.when <= now) { pop_timeout(0); if (to.fiber->sched_id == to.sched_id) { if (to.is_error) { janet_cancel(to.fiber, janet_cstringv("timeout")); } else { janet_schedule(to.fiber, janet_wrap_nil()); } } } /* Run scheduled fibers */ while (janet_vm_spawn.head != janet_vm_spawn.tail) { JanetTask task = {NULL, janet_wrap_nil(), JANET_SIGNAL_OK}; janet_q_pop(&janet_vm_spawn, &task, sizeof(task)); run_one(task.fiber, task.value, task.sig); } /* Poll for events */ if (janet_vm_active_listeners || janet_vm_tq_count) { JanetTimeout to; memset(&to, 0, sizeof(to)); int has_timeout; /* Drop timeouts that are no longer needed */ while ((has_timeout = peek_timeout(&to)) && to.fiber->sched_id != to.sched_id) { pop_timeout(0); } /* Run polling implementation */ janet_loop1_impl(has_timeout, to.when); } } void janet_loop(void) { while (janet_vm_active_listeners || (janet_vm_spawn.head != janet_vm_spawn.tail) || janet_vm_tq_count) { janet_loop1(); } } #ifdef JANET_WINDOWS /* Epoll global data */ JANET_THREAD_LOCAL HANDLE janet_vm_iocp = NULL; static JanetTimestamp ts_now(void) { return (JanetTimestamp) GetTickCount64(); } 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(JanetStream *stream, JanetListener behavior, int mask, size_t size, void *user) { /* Add the handle to the io completion port if not already added */ JanetListenerState *state = janet_listen_impl(stream, behavior, mask, size, user); if (!(stream->flags & JANET_POLL_FLAG_IOCP)) { if (NULL == CreateIoCompletionPort(stream->handle, janet_vm_iocp, (ULONG_PTR) stream, 0)) { janet_panic("failed to listen for events"); } stream->flags |= JANET_POLL_FLAG_IOCP; } return state; } static void janet_unlisten(JanetListenerState *state) { janet_unlisten_impl(state); } void janet_loop1_impl(int has_timeout, JanetTimestamp 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) { waittime = 0; } else { waittime = (uint64_t)(to - now); } } else { waittime = INFINITE; } BOOL result = GetQueuedCompletionStatus(janet_vm_iocp, &num_bytes_transfered, &completionKey, &overlapped, (DWORD) waittime); if (!result) { if (!has_timeout) { /* queue emptied */ } } else { /* Normal event */ JanetStream *stream = (JanetStream *) completionKey; JanetListenerState *state = stream->state; while (state != NULL) { if (state->tag == overlapped) { state->event = overlapped; state->bytes = num_bytes_transfered; JanetAsyncStatus status = state->machine(state, JANET_ASYNC_EVENT_COMPLETE); if (status == JANET_ASYNC_STATUS_DONE) { janet_unlisten(state); } break; } else { state = state->_next; } } } } #elif defined(JANET_EV_POLL) /* * Start linux/epoll implementation */ static JanetTimestamp ts_now(void) { struct timespec now; janet_assert(-1 != clock_gettime(CLOCK_MONOTONIC, &now), "failed to get time"); uint64_t res = 1000 * now.tv_sec; res += now.tv_nsec / 1000000; return res; } /* Epoll global data */ JANET_THREAD_LOCAL int janet_vm_epoll = 0; JANET_THREAD_LOCAL int janet_vm_timerfd = 0; JANET_THREAD_LOCAL int janet_vm_timer_enabled = 0; static int make_epoll_events(int mask) { int events = EPOLLET; if (mask & JANET_ASYNC_LISTEN_READ) events |= EPOLLIN; if (mask & JANET_ASYNC_LISTEN_WRITE) events |= EPOLLOUT; return events; } /* Wait for the next event */ JanetListenerState *janet_listen(JanetStream *stream, JanetListener behavior, int mask, size_t size, void *user) { int is_first = !(stream->state); int op = is_first ? EPOLL_CTL_ADD : EPOLL_CTL_MOD; JanetListenerState *state = janet_listen_impl(stream, behavior, mask, size, user); struct epoll_event ev; ev.events = make_epoll_events(state->stream->_mask); ev.data.ptr = stream; int status; do { status = epoll_ctl(janet_vm_epoll, op, stream->handle, &ev); } while (status == -1 && errno == EINTR); if (status == -1) { janet_unlisten_impl(state); janet_panicf("failed to schedule event: %s", strerror(errno)); } return state; } /* Tell system we are done listening for a certain event */ static void janet_unlisten(JanetListenerState *state) { JanetStream *stream = state->stream; if (!(stream->flags & JANET_STREAM_CLOSED)) { int is_last = (state->_next == NULL && stream->state == state); int op = is_last ? EPOLL_CTL_DEL : EPOLL_CTL_MOD; struct epoll_event ev; ev.events = make_epoll_events(stream->_mask & ~state->_mask); ev.data.ptr = stream; int status; do { status = epoll_ctl(janet_vm_epoll, op, stream->handle, &ev); } while (status == -1 && errno == EINTR); if (status == -1) { janet_panicf("failed to unschedule event: %s", strerror(errno)); } } /* Destroy state machine and free memory */ janet_unlisten_impl(state); } #define JANET_EPOLL_MAX_EVENTS 64 void janet_loop1_impl(int has_timeout, JanetTimestamp timeout) { if (janet_vm_timer_enabled || has_timeout) { memset(&its, 0, sizeof(its)); if (has_timeout) { its.it_value.tv_sec = timeout / 1000; its.it_value.tv_nsec = (timeout % 1000) * 1000000; } timerfd_settime(janet_vm_timerfd, TFD_TIMER_ABSTIME, &its, NULL); } janet_vm_timer_enabled = has_timeout; /* Poll for events */ struct epoll_event events[JANET_EPOLL_MAX_EVENTS]; int ready; do { ready = epoll_wait(janet_vm_epoll, events, JANET_EPOLL_MAX_EVENTS, -1); } while (ready == -1 && errno == EINTR); if (ready == -1) { JANET_EXIT("failed to poll events"); } /* Step state machines */ for (int i = 0; i < ready; i++) { JanetStream *stream = events[i].data.ptr; if (NULL != stream) { /* If NULL, is a timeout */ int mask = events[i].events; JanetListenerState *state = stream->state; state->event = events + i; while (NULL != state) { JanetListenerState *next_state = state->_next; JanetAsyncStatus status1 = JANET_ASYNC_STATUS_NOT_DONE; JanetAsyncStatus status2 = JANET_ASYNC_STATUS_NOT_DONE; if (mask & EPOLLOUT) status1 = state->machine(state, JANET_ASYNC_EVENT_WRITE); if (mask & EPOLLIN) status2 = state->machine(state, JANET_ASYNC_EVENT_READ); if (mask & EPOLLERR) status3 = state->machine(state, JANET_ASYNC_EVENT_ERR); if (mask & EPOLLHUP) status4 = state->machine(state, JANET_ASYNC_EVENT_HUP); if (status1 == JANET_ASYNC_STATUS_DONE || status2 == JANET_ASYNC_STATUS_DONE || status3 == JANET_ASYNC_STATUS_DONE || status4 == JANET_ASYNC_STATUS_DONE) janet_unlisten(state); state = next_state; } } } } void janet_ev_init(void) { janet_ev_init_common(); janet_vm_epoll = epoll_create1(EPOLL_CLOEXEC); janet_vm_timerfd = timerfd_create(CLOCK_MONOTONIC, TFD_CLOEXEC | TFD_NONBLOCK); janet_vm_timer_enabled = 0; if (janet_vm_epoll == -1 || janet_vm_timerfd == -1) goto error; struct epoll_event ev; ev.events = EPOLLIN | EPOLLET; ev.data.ptr = NULL; if (-1 == epoll_ctl(janet_vm_epoll, EPOLL_CTL_ADD, janet_vm_timerfd, &ev)) goto error; return; error: JANET_EXIT("failed to initialize event loop"); } void janet_ev_deinit(void) { janet_ev_deinit_common(); close(janet_vm_epoll); close(janet_vm_timerfd); janet_vm_epoll = 0; } /* * End epoll implementation */ #else #include /* Poll implementation */ static JanetTimestamp ts_now(void) { struct timespec now; janet_assert(-1 != clock_gettime(CLOCK_REALTIME, &now), "failed to get time"); uint64_t res = 1000 * now.tv_sec; res += now.tv_nsec / 1000000; return res; } /* Epoll global data */ JANET_THREAD_LOCAL struct pollfd *janet_vm_fds = NULL; JANET_THREAD_LOCAL JanetListenerState **janet_vm_listener_map = NULL; JANET_THREAD_LOCAL size_t janet_vm_fdcap = 0; JANET_THREAD_LOCAL size_t janet_vm_fdcount = 0; static int make_poll_events(int mask) { int events = 0; if (mask & JANET_ASYNC_LISTEN_READ) events |= POLLIN; if (mask & JANET_ASYNC_LISTEN_WRITE) events |= POLLOUT; return events; } static void janet_push_pollfd(struct pollfd pfd) { if (janet_vm_fdcap == janet_vm_fdcount) { size_t newcap = janet_vm_fdcount ? janet_vm_fdcount * 2 : 16; janet_vm_fds = realloc(janet_vm_fds, newcap * sizeof(struct pollfd)); if (NULL == janet_vm_fds) { JANET_OUT_OF_MEMORY; } janet_vm_listener_map = realloc(janet_vm_listener_map, newcap * sizeof(JanetListenerState *)); if (NULL == janet_vm_listener_map) { JANET_OUT_OF_MEMORY; } janet_vm_fdcap = newcap; } janet_vm_fds[janet_vm_fdcount++] = pfd; } /* Wait for the next event */ JanetListenerState *janet_listen(JanetStream *stream, JanetListener behavior, int mask, size_t size, void *user) { JanetListenerState *state = janet_listen_impl(stream, behavior, mask, size, user); struct pollfd ev; ev.fd = stream->handle; ev.events = make_poll_events(state->stream->_mask); ev.revents = 0; state->_index = janet_vm_fdcount; janet_push_pollfd(ev); janet_vm_listener_map[state->_index] = state; return state; } /* Tell system we are done listening for a certain event */ static void janet_unlisten(JanetListenerState *state) { janet_vm_fds[state->_index] = janet_vm_fds[--janet_vm_fdcount]; JanetListenerState *replacer = janet_vm_listener_map[janet_vm_fdcount]; janet_vm_listener_map[state->_index] = replacer; /* Update pointers in replacer */ replacer->_index = state->_index; /* Destroy state machine and free memory */ janet_unlisten_impl(state); } void janet_loop1_impl(int has_timeout, JanetTimestamp timeout) { /* Poll for events */ int ready; do { if (has_timeout) { JanetTimestamp now = ts_now(); ready = poll(janet_vm_fds, janet_vm_fdcount, now > timeout ? 0 : (int)(timeout - now)); } else { ready = poll(janet_vm_fds, janet_vm_fdcount, -1); } } while (ready == -1 && errno == EINTR); if (ready == -1) { JANET_EXIT("failed to poll events"); } /* Step state machines */ for (size_t i = 0; i < janet_vm_fdcount; i++) { struct pollfd *pfd = janet_vm_fds + i; /* Skip fds where nothing interesting happened */ JanetListenerState *state = janet_vm_listener_map[i]; /* Normal event */ int mask = janet_vm_fds[i].revents; JanetAsyncStatus status1 = JANET_ASYNC_STATUS_NOT_DONE; JanetAsyncStatus status2 = JANET_ASYNC_STATUS_NOT_DONE; JanetAsyncStatus status3 = JANET_ASYNC_STATUS_NOT_DONE; JanetAsyncStatus status4 = JANET_ASYNC_STATUS_NOT_DONE; state->event = pfd; if (mask & POLLOUT) status1 = state->machine(state, JANET_ASYNC_EVENT_WRITE); if (mask & POLLIN) status2 = state->machine(state, JANET_ASYNC_EVENT_READ); if (mask & POLLERR) status2 = state->machine(state, JANET_ASYNC_EVENT_ERR); if (mask & POLLHUP) status2 = state->machine(state, JANET_ASYNC_EVENT_HUP); if (status1 == JANET_ASYNC_STATUS_DONE || status2 == JANET_ASYNC_STATUS_DONE || status3 == JANET_ASYNC_STATUS_DONE || status4 == JANET_ASYNC_STATUS_DONE) janet_unlisten(state); } } void janet_ev_init(void) { janet_ev_init_common(); janet_vm_fds = NULL; janet_vm_listener_map = NULL; janet_vm_fdcap = 0; janet_vm_fdcount = 0; return; } void janet_ev_deinit(void) { janet_ev_deinit_common(); free(janet_vm_fds); free(janet_vm_listener_map); janet_vm_fds = NULL; janet_vm_listener_map = NULL; janet_vm_fdcap = 0; janet_vm_fdcount = 0; } #endif /* C API helpers for reading and writing from streams. * There is some networking code in here as well as generic * reading and writing primitives. */ /* When there is an IO error, we need to be able to convert it to a Janet * string to raise a Janet error. */ #ifdef JANET_WINDOWS #define JANET_EV_CHUNKSIZE 4096 Janet janet_ev_lasterr(void) { int code = GetLastError(); char msgbuf[256]; msgbuf[0] = '\0'; FormatMessage(FORMAT_MESSAGE_FROM_SYSTEM | FORMAT_MESSAGE_IGNORE_INSERTS, NULL, code, MAKELANGID(LANG_NEUTRAL, SUBLANG_DEFAULT), msgbuf, sizeof(msgbuf), NULL); if (!*msgbuf) sprintf(msgbuf, "%d", code); char *c = msgbuf; while (*c) { if (*c == '\n' || *c == '\r') { *c = '\0'; break; } c++; } return janet_cstringv(msgbuf); } #else Janet janet_ev_lasterr(void) { return janet_cstringv(strerror(errno)); } #endif /* State machine for read/recv/recvfrom */ typedef enum { JANET_ASYNC_READMODE_READ, JANET_ASYNC_READMODE_RECV, JANET_ASYNC_READMODE_RECVFROM } JanetReadMode; typedef struct { JanetListenerState head; int32_t bytes_left; int32_t bytes_read; JanetBuffer *buf; int is_chunk; JanetReadMode mode; #ifdef JANET_WINDOWS OVERLAPPED overlapped; #ifdef JANET_NET WSABUF wbuf; DWORD flags; struct sockaddr from; int fromlen; #endif uint8_t chunk_buf[JANET_EV_CHUNKSIZE]; #else int flags; #endif } StateRead; JanetAsyncStatus ev_machine_read(JanetListenerState *s, JanetAsyncEvent event) { StateRead *state = (StateRead *) s; switch (event) { default: break; case JANET_ASYNC_EVENT_MARK: janet_mark(janet_wrap_buffer(state->buf)); break; case JANET_ASYNC_EVENT_CLOSE: janet_cancel(s->fiber, janet_cstringv("stream closed")); return JANET_ASYNC_STATUS_DONE; #ifdef JANET_WINDOWS case JANET_ASYNC_EVENT_COMPLETE: { /* Called when read finished */ state->bytes_read += s->bytes; if (state->bytes_read == 0 && (state->mode != JANET_ASYNC_READMODE_RECVFROM)) { janet_schedule(s->fiber, janet_wrap_nil()); return JANET_ASYNC_STATUS_DONE; } janet_buffer_push_bytes(state->buf, state->chunk_buf, s->bytes); state->bytes_left -= s->bytes; if (state->bytes_left <= 0 || !state->is_chunk || s->bytes == 0) { Janet resume_val; #ifdef JANET_NET if (state->mode == JANET_ASYNC_READMODE_RECVFROM) { void *abst = janet_abstract(&janet_address_type, state->fromlen); memcpy(abst, &state->from, state->fromlen); resume_val = janet_wrap_abstract(abst); } else #endif { resume_val = janet_wrap_buffer(state->buf); } janet_schedule(s->fiber, resume_val); return JANET_ASYNC_STATUS_DONE; } } /* fallthrough */ case JANET_ASYNC_EVENT_USER: { int32_t chunk_size = state->bytes_left > JANET_EV_CHUNKSIZE ? JANET_EV_CHUNKSIZE : state->bytes_left; s->tag = &state->overlapped; memset(&(state->overlapped), 0, sizeof(OVERLAPPED)); int status; #ifdef JANET_NET if (state->mode != JANET_ASYNC_READMODE_READ) { state->wbuf.len = (ULONG) chunk_size; state->wbuf.buf = state->chunk_buf; if (state->mode == JANET_ASYNC_READMODE_RECVFROM) { status = WSARecvFrom((SOCKET) s->stream->handle, &state->wbuf, 1, NULL, &state->flags, &state->from, &state->fromlen, &state->overlapped, NULL); } else { status = WSARecv((SOCKET) s->stream->handle, &state->wbuf, 1, NULL, &state->flags, &state->overlapped, NULL); } } if (status && (WSA_IO_PENDING != WSAGetLastError())) { janet_cancel(s->fiber, janet_ev_lasterr()); return JANET_ASYNC_STATUS_DONE; } } else #endif { status = ReadFile(s->stream->handle, state->chunk_buf, chunk_size, NULL, &state->overlapped); if (status && (ERROR_IO_PENDING != WSAGetLastError())) { janet_cancel(s->fiber, janet_ev_lasterr()); return JANET_ASYNC_STATUS_DONE; } } } break; #else case JANET_ASYNC_EVENT_ERR: case JANET_ASYNC_EVENT_HUP: { if (state->bytes_read) { janet_schedule(s->fiber, janet_wrap_buffer(state->buf)); } else { janet_schedule(s->fiber, janet_wrap_nil()); } return JANET_ASYNC_STATUS_DONE; } case JANET_ASYNC_EVENT_READ: /* Read in bytes */ { JanetBuffer *buffer = state->buf; int32_t bytes_left = state->bytes_left; janet_buffer_extra(buffer, bytes_left); ssize_t nread; #ifdef JANET_NET char saddr[256]; socklen_t socklen = sizeof(saddr); #endif do { #ifdef JANET_NET if (state->mode == JANET_ASYNC_READMODE_RECVFROM) { nread = recvfrom(s->stream->handle, buffer->data + buffer->count, bytes_left, state->flags, (struct sockaddr *)&saddr, &socklen); } else if (state->mode == JANET_ASYNC_READMODE_RECV) { nread = recv(s->stream->handle, buffer->data + buffer->count, bytes_left, state->flags); } else #endif { nread = read(s->stream->handle, buffer->data + buffer->count, bytes_left); } } while (nread == -1 && errno == EINTR); /* Check for errors - special case errors that can just be waited on to fix */ if (nread == -1) { if (errno == EAGAIN || errno == EWOULDBLOCK) break; janet_cancel(s->fiber, janet_ev_lasterr()); return JANET_ASYNC_STATUS_DONE; } /* Only allow 0-length packets in recv-from. In stream protocols, a zero length packet is EOS. */ state->bytes_read += nread; if (state->bytes_read == 0 && (state->mode != JANET_ASYNC_READMODE_RECVFROM)) { janet_schedule(s->fiber, janet_wrap_nil()); return JANET_ASYNC_STATUS_DONE; } /* Increment buffer counts */ buffer->count += nread; bytes_left -= nread; state->bytes_left = bytes_left; /* Resume if done */ if (!state->is_chunk || bytes_left == 0 || nread == 0) { Janet resume_val; #ifdef JANET_NET if (state->mode == JANET_ASYNC_READMODE_RECVFROM) { void *abst = janet_abstract(&janet_address_type, socklen); memcpy(abst, &saddr, socklen); resume_val = janet_wrap_abstract(abst); } else #endif { resume_val = janet_wrap_buffer(buffer); } janet_schedule(s->fiber, resume_val); return JANET_ASYNC_STATUS_DONE; } } break; #endif } return JANET_ASYNC_STATUS_NOT_DONE; } static void janet_ev_read_generic(JanetStream *stream, JanetBuffer *buf, int32_t nbytes, int is_chunked, JanetReadMode mode, int flags) { StateRead *state = (StateRead *) janet_listen(stream, ev_machine_read, JANET_ASYNC_LISTEN_READ, sizeof(StateRead), NULL); state->is_chunk = is_chunked; state->buf = buf; state->bytes_left = nbytes; state->bytes_read = 0; state->mode = mode; #ifdef JANET_WINDOWS ev_machine_read((JanetListenerState *) state, JANET_ASYNC_EVENT_USER); state->flags = (DWORD) flags; #else state->flags = flags; #endif } void janet_ev_read(JanetStream *stream, JanetBuffer *buf, int32_t nbytes) { janet_ev_read_generic(stream, buf, nbytes, 0, JANET_ASYNC_READMODE_READ, 0); } void janet_ev_readchunk(JanetStream *stream, JanetBuffer *buf, int32_t nbytes) { janet_ev_read_generic(stream, buf, nbytes, 1, JANET_ASYNC_READMODE_READ, 0); } #ifdef JANET_NET void janet_ev_recv(JanetStream *stream, JanetBuffer *buf, int32_t nbytes, int flags) { janet_ev_read_generic(stream, buf, nbytes, 0, JANET_ASYNC_READMODE_RECV, flags); } void janet_ev_recvchunk(JanetStream *stream, JanetBuffer *buf, int32_t nbytes, int flags) { janet_ev_read_generic(stream, buf, nbytes, 1, JANET_ASYNC_READMODE_RECV, flags); } void janet_ev_recvfrom(JanetStream *stream, JanetBuffer *buf, int32_t nbytes, int flags) { janet_ev_read_generic(stream, buf, nbytes, 0, JANET_ASYNC_READMODE_RECVFROM, flags); } #endif /* * State machine for write/send/send-to */ typedef enum { JANET_ASYNC_WRITEMODE_WRITE, JANET_ASYNC_WRITEMODE_SEND, JANET_ASYNC_WRITEMODE_SENDTO } JanetWriteMode; typedef struct { JanetListenerState head; union { JanetBuffer *buf; const uint8_t *str; } src; int is_buffer; JanetWriteMode mode; void *dest_abst; #ifdef JANET_WINDOWS OVERLAPPED overlapped; #ifdef JANET_NET WSABUF wbuf; DWORD flags; #endif #else int flags; int32_t start; #endif } StateWrite; JanetAsyncStatus ev_machine_write(JanetListenerState *s, JanetAsyncEvent event) { StateWrite *state = (StateWrite *) s; switch (event) { default: break; case JANET_ASYNC_EVENT_MARK: janet_mark(state->is_buffer ? janet_wrap_buffer(state->src.buf) : janet_wrap_string(state->src.str)); if (state->mode == JANET_ASYNC_WRITEMODE_SENDTO) { janet_mark(janet_wrap_abstract(state->dest_abst)); } break; case JANET_ASYNC_EVENT_CLOSE: janet_cancel(s->fiber, janet_cstringv("stream closed")); return JANET_ASYNC_STATUS_DONE; #ifdef JANET_WINDOWS case JANET_ASYNC_EVENT_COMPLETE: { /* Called when write finished */ if (s->bytes == 0 && (state->mode != JANET_ASYNC_WRITEMODE_SENDTO)) { janet_cancel(s->fiber, janet_cstringv("disconnect")); return JANET_ASYNC_STATUS_DONE; } janet_schedule(s->fiber, janet_wrap_nil()); return JANET_ASYNC_STATUS_DONE; } break; case JANET_ASYNC_EVENT_USER: { /* Begin write */ int32_t start, len; const uint8_t *bytes; if (state->is_buffer) { /* If buffer, convert to string. */ /* TODO - be more efficient about this */ JanetBuffer *buffer = state->src.buf; JanetString str = janet_string(buffer->data, buffer->count); bytes = str; len = buffer->count; state->is_buffer = 0; state->src.str = str; } else { bytes = state->src.str; len = janet_string_length(bytes); } s->tag = &state->overlapped; memset(&(state->overlapped), 0, sizeof(WSAOVERLAPPED)); int status; #ifdef JANET_NET if (state->mode != JANET_ASYNC_WRITEMODE_WRITE) { SOCKET sock = (SOCKET) s->stream->handle; state->wbuf.buf = (char *) bytes; state->wbuf.len = len; if (state->mode == JANET_ASYNC_WRITEMODE_SENDTO) { const struct sockaddr *to = state->dest_abst; int tolen = (int) janet_abstract_size((void *) to); status = WSASendTo(sock, &state->wbuf, 1, NULL, state->flags, to, tolen, &state->overlapped, NULL); } else { status = WSASend(sock, &state->wbuf, 1, NULL, state->flags, &state->overlapped, NULL); } if (status && (WSA_IO_PENDING != WSAGetLastError())) { janet_cancel(s->fiber, janet_cstringv("failed to write to stream")); return JANET_ASYNC_STATUS_DONE; } } else #endif { status = WriteFile(s->stream->handle, bytes, len, NULL, &state->overlapped); if (status && (ERROR_IO_PENDING != WSAGetLastError())) { janet_cancel(s->fiber, janet_cstringv("failed to write to stream")); return JANET_ASYNC_STATUS_DONE; } } } break; #else case JANET_ASYNC_EVENT_ERR: janet_cancel(s->fiber, janet_cstringv("stream err")); return JANET_ASYNC_STATUS_DONE; case JANET_ASYNC_EVENT_HUP: janet_cancel(s->fiber, janet_cstringv("stream hup")); return JANET_ASYNC_STATUS_DONE; case JANET_ASYNC_EVENT_WRITE: { int32_t start, len; const uint8_t *bytes; start = state->start; if (state->is_buffer) { JanetBuffer *buffer = state->src.buf; bytes = buffer->data; len = buffer->count; } else { bytes = state->src.str; len = janet_string_length(bytes); } ssize_t nwrote = 0; if (start < len) { int32_t nbytes = len - start; void *dest_abst = state->dest_abst; do { #ifdef JANET_NET if (state->mode == JANET_ASYNC_WRITEMODE_SENDTO) { nwrote = sendto(s->stream->handle, bytes + start, nbytes, state->flags, (struct sockaddr *) dest_abst, janet_abstract_size(dest_abst)); } else if (state->mode == JANET_ASYNC_WRITEMODE_SEND) { nwrote = send(s->stream->handle, bytes + start, nbytes, state->flags); } else #endif { nwrote = write(s->stream->handle, bytes + start, nbytes); } } while (nwrote == -1 && errno == EINTR); /* Handle write errors */ if (nwrote == -1) { if (errno == EAGAIN || errno == EWOULDBLOCK) break; janet_cancel(s->fiber, janet_ev_lasterr()); return JANET_ASYNC_STATUS_DONE; } /* Unless using datagrams, empty message is a disconnect */ if (nwrote == 0 && !dest_abst) { janet_cancel(s->fiber, janet_cstringv("disconnect")); return JANET_ASYNC_STATUS_DONE; } if (nwrote > 0) { start += nwrote; } else { start = len; } } state->start = start; if (start >= len) { janet_schedule(s->fiber, janet_wrap_nil()); return JANET_ASYNC_STATUS_DONE; } break; } break; #endif } return JANET_ASYNC_STATUS_NOT_DONE; } static void janet_ev_write_generic(JanetStream *stream, void *buf, void *dest_abst, JanetWriteMode mode, int is_buffer, int flags) { StateWrite *state = (StateWrite *) janet_listen(stream, ev_machine_write, JANET_ASYNC_LISTEN_WRITE, sizeof(StateWrite), NULL); state->is_buffer = is_buffer; state->src.buf = buf; state->dest_abst = dest_abst; state->mode = mode; #ifdef JANET_WINDOWS state->flags = (DWORD) flags; net_machine_write((JanetListenerState *) state, JANET_ASYNC_EVENT_USER); #else state->start = 0; state->flags = flags; #endif } void janet_ev_write_buffer(JanetStream *stream, JanetBuffer *buf) { janet_ev_write_generic(stream, buf, NULL, JANET_ASYNC_WRITEMODE_WRITE, 1, 0); } void janet_ev_write_string(JanetStream *stream, JanetString str) { janet_ev_write_generic(stream, (void *) str, NULL, JANET_ASYNC_WRITEMODE_WRITE, 0, 0); } #ifdef JANET_NET void janet_ev_send_buffer(JanetStream *stream, JanetBuffer *buf, int flags) { janet_ev_write_generic(stream, buf, NULL, JANET_ASYNC_WRITEMODE_SEND, 1, flags); } void janet_ev_send_string(JanetStream *stream, JanetString str, int flags) { janet_ev_write_generic(stream, (void *) str, NULL, JANET_ASYNC_WRITEMODE_SEND, 0, flags); } void janet_ev_sendto_buffer(JanetStream *stream, JanetBuffer *buf, void *dest, int flags) { janet_ev_write_generic(stream, buf, dest, JANET_ASYNC_WRITEMODE_SENDTO, 1, flags); } void janet_ev_sendto_string(JanetStream *stream, JanetString str, void *dest, int flags) { janet_ev_write_generic(stream, (void *) str, dest, JANET_ASYNC_WRITEMODE_SENDTO, 0, flags); } #endif /* C functions */ static Janet cfun_ev_go(int32_t argc, Janet *argv) { janet_arity(argc, 1, 2); JanetFiber *fiber = janet_getfiber(argv, 0); Janet value = argc == 2 ? argv[1] : janet_wrap_nil(); janet_schedule(fiber, value); return argv[0]; } static Janet cfun_ev_call(int32_t argc, Janet *argv) { janet_arity(argc, 1, -1); JanetFunction *fn = janet_getfunction(argv, 0); JanetFiber *fiber = janet_fiber(fn, 64, argc - 1, argv + 1); janet_schedule(fiber, janet_wrap_nil()); return janet_wrap_fiber(fiber); } static Janet cfun_ev_sleep(int32_t argc, Janet *argv) { janet_fixarity(argc, 1); double sec = janet_getnumber(argv, 0); JanetTimeout to; to.when = ts_delta(ts_now(), sec); to.fiber = janet_vm_root_fiber; to.is_error = 0; to.sched_id = to.fiber->sched_id; add_timeout(to); janet_await(); } static Janet cfun_ev_cancel(int32_t argc, Janet *argv) { janet_fixarity(argc, 2); JanetFiber *fiber = janet_getfiber(argv, 0); Janet err = argv[1]; janet_cancel(fiber, err); return argv[0]; } Janet janet_cfun_stream_close(int32_t argc, Janet *argv) { janet_fixarity(argc, 1); JanetStream *stream = janet_getabstract(argv, 0, &janet_stream_type); janet_stream_close(stream); return argv[0]; } Janet janet_cfun_stream_read(int32_t argc, Janet *argv) { janet_arity(argc, 2, 4); JanetStream *stream = janet_getabstract(argv, 0, &janet_stream_type); janet_stream_flags(stream, JANET_STREAM_READABLE); int32_t n = janet_getnat(argv, 1); JanetBuffer *buffer = janet_optbuffer(argv, argc, 2, 10); double to = janet_optnumber(argv, argc, 3, INFINITY); if (to != INFINITY) janet_addtimeout(to); janet_ev_read(stream, buffer, n); janet_await(); } Janet janet_cfun_stream_chunk(int32_t argc, Janet *argv) { janet_arity(argc, 2, 4); JanetStream *stream = janet_getabstract(argv, 0, &janet_stream_type); janet_stream_flags(stream, JANET_STREAM_READABLE); int32_t n = janet_getnat(argv, 1); JanetBuffer *buffer = janet_optbuffer(argv, argc, 2, 10); double to = janet_optnumber(argv, argc, 3, INFINITY); if (to != INFINITY) janet_addtimeout(to); janet_ev_readchunk(stream, buffer, n); janet_await(); } Janet janet_cfun_stream_write(int32_t argc, Janet *argv) { janet_arity(argc, 2, 3); JanetStream *stream = janet_getabstract(argv, 0, &janet_stream_type); janet_stream_flags(stream, JANET_STREAM_WRITABLE); double to = janet_optnumber(argv, argc, 2, INFINITY); if (janet_checktype(argv[1], JANET_BUFFER)) { if (to != INFINITY) janet_addtimeout(to); janet_ev_write_buffer(stream, janet_getbuffer(argv, 1)); } else { JanetByteView bytes = janet_getbytes(argv, 1); if (to != INFINITY) janet_addtimeout(to); janet_ev_write_string(stream, bytes.bytes); } janet_await(); } static Janet cfun_ev_pipe(int32_t argc, Janet *argv) { (void) argv; janet_fixarity(argc, 0); #ifdef JANET_WINDOWS JanetHandle rhandle, whandle; if (!CreatePipe(&rhandle, &whandle, NULL, 0)) janet_panicv(janet_ev_lasterr()); JanetStream *reader = janet_stream(rhandle, JANET_STREAM_READABLE, NULL); JanetStream *writer = janet_stream(whandle, JANET_STREAM_WRITABLE, NULL); #else int fds[2]; if (pipe(fds)) janet_panicv(janet_ev_lasterr()); JanetStream *reader = janet_stream(fds[0], JANET_STREAM_READABLE, NULL); JanetStream *writer = janet_stream(fds[1], JANET_STREAM_WRITABLE, NULL); #endif Janet tup[2] = {janet_wrap_abstract(reader), janet_wrap_abstract(writer)}; return janet_wrap_tuple(janet_tuple_n(tup, 2)); } static const JanetReg ev_cfuns[] = { { "ev/call", cfun_ev_call, JDOC("(ev/call fn & args)\n\n" "Call a function asynchronously. Returns a fiber that is scheduled to " "run the function.") }, { "ev/go", cfun_ev_go, JDOC("(ev/go fiber &opt value)\n\n" "Put a fiber on the event loop to be resumed later. Optionally pass " "a value to resume with, otherwise resumes with nil.") }, { "ev/sleep", cfun_ev_sleep, JDOC("(ev/sleep sec)\n\n" "Suspend the current fiber for sec seconds without blocking the event loop.") }, { "ev/chan", cfun_channel_new, JDOC("(ev/chan &opt capacity)\n\n" "Create a new channel. capacity is the number of values to queue before " "blocking writers, defaults to 0 if not provided. Returns a new channel.") }, { "ev/give", cfun_channel_push, JDOC("(ev/give channel value)\n\n" "Write a value to a channel, suspending the current fiber if the channel is full.") }, { "ev/take", cfun_channel_pop, JDOC("(ev/take channel)\n\n" "Read from a channel, suspending the current fiber if no value is available.") }, { "ev/full", cfun_channel_full, JDOC("(ev/full channel)\n\n" "Check if a channel is full or not.") }, { "ev/capacity", cfun_channel_capacity, JDOC("(ev/capacity channel)\n\n" "Get the number of items a channel will store before blocking writers.") }, { "ev/count", cfun_channel_count, JDOC("(ev/count channel)\n\n" "Get the number of items currently waiting in a channel.") }, { "ev/cancel", cfun_ev_cancel, JDOC("(ev/cancel fiber err)\n\n" "Cancel a suspended fiber in the event loop. Differs from cancel in that it returns the canceled fiber immediately") }, { "ev/select", cfun_channel_choice, JDOC("(ev/select & clauses)\n\n" "Block until the first of several channel operations occur. Returns a tuple of the form [:give chan] or [:take chan x], where " "a :give tuple is the result of a write and :take tuple is the result of a write. Each clause must be either a channel (for " "a channel take operation) or a tuple [channel x] for a channel give operation. Operations are tried in order, such that the first " "clauses will take precedence over later clauses.") }, { "ev/rselect", cfun_channel_rchoice, JDOC("(ev/rselect & clauses)\n\n" "Similar to ev/choice, but will try clauses in a random order for fairness.") }, { "ev/close", janet_cfun_stream_close, JDOC("(ev/close stream)\n\n" "Close a stream. This should be the same as calling (:close stream) for all streams.") }, { "ev/read", janet_cfun_stream_read, JDOC("(ev/read stream n &opt buffer timeout)\n\n" "Read up to n bytes into a buffer asynchronously from a stream. " "Optionally provide a buffer to write into " "as well as a timeout in seconds after which to cancel the operation and raise an error. " "Returns the buffer if the read was successful or nil if end-of-stream reached. Will raise an " "error if there are problems with the IO operation.") }, { "ev/chunk", janet_cfun_stream_chunk, JDOC("(ev/chunk stream n &opt buffer timeout)\n\n" "Same as ev/read, but will not return early if less than n bytes are available. If an end of " "stream is reached, will also return early with the collected bytes.") }, { "ev/write", janet_cfun_stream_write, JDOC("(ev/write stream data &opt timeout)\n\n" "Write data to a stream, suspending the current fiber until the write " "completes. Takes an optional timeout in seconds, after which will return nil. " "Returns nil, or raises an error if the write failed.") }, { "ev/pipe", cfun_ev_pipe, JDOC("(ev/pipe)\n\n" "Create a readable stream and a writable stream that are connected. Returns a two element " "tuple where the first element is a readable stream and the second element is the writable " "stream.") }, {NULL, NULL, NULL} }; void janet_lib_ev(JanetTable *env) { janet_core_cfuns(env, NULL, ev_cfuns); } #endif