Fix some valgrind errors.

A null pointer dereference and a memory leak with the line/col mapping.

.builds/epoll.yml

@@ -0,0 +1,15 @@
+image: archlinux
+sources:
+- https://git.sr.ht/~bakpakin/janet
+packages:
+- meson
+tasks:
+- build: |
+    cd janet
+    meson setup build --buildtype=release
+    cd build
+    meson configure -Depoll=false
+    ninja
+    ninja test
+    sudo ninja install
+    sudo jpm --verbose install circlet

CHANGELOG.md

@@ -2,7 +2,26 @@
 All notable changes to this project will be documented in this file.
 
 ## Unreleased - ???
+- Backtick delimited strings and buffers are now reindented based on the column of the
+  opening delimiter. WHitespace in columns to the left of the starting column is ignored unless
+  there are non-space/non-newline characters in that region, in which case the old behavior is preserved.
+- Argument to `(error)` combinator in PEGs is now optional.
+- Add `(line)` and `(column)` combinators to PEGs to capture source line and column.
+  This should make error reporting a bit easier.
+- Add `merge-module` to core.
+- During installation and release, merge janetconf.h into janet.h for easier install.
+- Add `upscope` special form.
+- `os/execute` and `os/spawn` can take streams for redirecting IO.
+- Add `;parser` and `:read` parameters to `run-context`.
+- Add `os/open` if ev is enabled.
+- Add `os/pipe` if ev is enabled.
+- Add `janet_thread_current(void)` to C API
 - Add integer parsing forms to pegs. This makes parsing many binary protocols easier.
+- Lots of updates to networking code - now can use epoll (or poll) on linux and IOCP on windows.
+- Add `ev/` module. This exposes a fiber scheduler, queues, timeouts, and other functionality to users
+  for single threaded cooperative scheduling and asynchornous IO.
+- Add `net/accept-loop` and `net/listen`. These functions break down `net/server` into it's essential parts
+  and are more flexible. They also allow furter improvements to these utility functions.
 
 ## 1.12.2 - 2020-09-20
 - Add janet\_try and janet\_restore to C API.

Makefile

@@ -67,7 +67,7 @@ ifeq ($(UNAME), Haiku)
 endif
 
 $(shell mkdir -p build/core build/mainclient build/webclient build/boot)
-all: $(JANET_TARGET) $(JANET_LIBRARY) $(JANET_STATIC_LIBRARY)
+all: $(JANET_TARGET) $(JANET_LIBRARY) $(JANET_STATIC_LIBRARY) build/janet.h
 
 ######################
 ##### Name Files #####
@@ -162,16 +162,16 @@ SONAME=libjanet.so.1.12
 build/shell.c: src/mainclient/shell.c
 	cp $< $@
 
-build/janet.h: src/include/janet.h
-	cp $< $@
+build/janet.h: $(JANET_TARGET) src/include/janet.h src/conf/janetconf.h
+	./$(JANET_TARGET) tools/patch-header.janet src/include/janet.h src/conf/janetconf.h $@
 
 build/janetconf.h: src/conf/janetconf.h
 	cp $< $@
 
-build/janet.o: build/janet.c build/janet.h build/janetconf.h
+build/janet.o: build/janet.c src/include/janet.h src/conf/janetconf.h
 	$(HOSTCC) $(BUILD_CFLAGS) -c $< -o $@ -I build
 
-build/shell.o: build/shell.c build/janet.h build/janetconf.h
+build/shell.o: build/shell.c src/include/janet.h src/conf/janetconf.h
 	$(HOSTCC) $(BUILD_CFLAGS) -c $< -o $@ -I build
 
 $(JANET_TARGET): build/janet.o build/shell.o
@@ -222,7 +222,7 @@ callgrind: $(JANET_TARGET)
 dist: build/janet-dist.tar.gz
 
 build/janet-%.tar.gz: $(JANET_TARGET) \
-	src/include/janet.h src/conf/janetconf.h \
+	build/janet.h \
 	jpm.1 janet.1 LICENSE CONTRIBUTING.md $(JANET_LIBRARY) $(JANET_STATIC_LIBRARY) \
 	build/doc.html README.md build/janet.c build/shell.c jpm
 	$(eval JANET_DIST_DIR = "janet-$(shell basename $*)")
@@ -262,11 +262,11 @@ build/janet.pc: $(JANET_TARGET)
 	echo 'Libs: -L$${libdir} -ljanet' >> $@
 	echo 'Libs.private: $(CLIBS)' >> $@
 
-install: $(JANET_TARGET) $(JANET_LIBRARY) $(JANET_STATIC_LIBRARY) build/janet.pc build/jpm
+install: $(JANET_TARGET) $(JANET_LIBRARY) $(JANET_STATIC_LIBRARY) build/janet.pc build/jpm build/janet.h
 	mkdir -p '$(DESTDIR)$(BINDIR)'
 	cp $(JANET_TARGET) '$(DESTDIR)$(BINDIR)/janet'
 	mkdir -p '$(DESTDIR)$(INCLUDEDIR)/janet'
-	cp -rf $(JANET_HEADERS) '$(DESTDIR)$(INCLUDEDIR)/janet'
+	cp -r build/janet.h '$(DESTDIR)$(INCLUDEDIR)/janet'
 	mkdir -p '$(DESTDIR)$(JANET_PATH)'
 	mkdir -p '$(DESTDIR)$(LIBDIR)'
 	cp $(JANET_LIBRARY) '$(DESTDIR)$(LIBDIR)/libjanet.so.$(shell $(JANET_TARGET) -e '(print janet/version)')'

appveyor.yml

@@ -35,9 +35,6 @@ artifacts:
     - name: janet.h
       path: dist\janet.h
       type: File
-    - name: janetconf.h
-      path: dist\janetconf.h
-      type: File
     - name: shell.c
       path: dist\shell.c
       type: File

build_win.bat

@@ -28,10 +28,10 @@ if not "%JANET_BUILD%" == "" (
     @set JANET_COMPILE=%JANET_COMPILE% /DJANET_BUILD="\"%JANET_BUILD%\""
 )
 
-mkdir build
-mkdir build\core
-mkdir build\mainclient
-mkdir build\boot
+if not exist build mkdir build
+if not exist build\core mkdir build\core
+if not exist build\mainclient mkdir build\mainclient
+if not exist build\boot mkdir build\boot
 
 @rem Build the bootstrap interpreter
 for %%f in (src\core\*.c) do (
@@ -113,8 +113,8 @@ copy README.md dist\README.md
 copy janet.lib dist\janet.lib
 copy janet.exp dist\janet.exp
 
-copy src\include\janet.h dist\janet.h
-copy src\conf\janetconf.h dist\janetconf.h
+janet.exe tools\patch-header.janet src\include\janet.h src\conf\janetconf.h build\janet.h
+copy build\janet.h dist\janet.h
 copy build\libjanet.lib dist\libjanet.lib
 
 copy .\jpm dist\jpm

meson.build

@@ -73,6 +73,7 @@ conf.set('JANET_NO_UMASK', not get_option('umask'))
 conf.set('JANET_NO_REALPATH', not get_option('realpath'))
 conf.set('JANET_NO_PROCESSES', not get_option('processes'))
 conf.set('JANET_SIMPLE_GETLINE', get_option('simple_getline'))
+conf.set('JANET_EV_EPOLL', get_option('epoll'))
 if get_option('os_name') != ''
   conf.set('JANET_OS_NAME', get_option('os_name'))
 endif
@@ -247,8 +248,14 @@ pkg.generate(libjanet,
 
 # Installation
 install_man('janet.1')
-install_headers(['src/include/janet.h', jconf], subdir: 'janet')
 install_data(sources : ['tools/.keep'], install_dir : join_paths(get_option('libdir'), 'janet'))
+patched_janet = custom_target('patched-janeth',
+  input : ['tools/patch-header.janet', 'src/include/janet.h', jconf],
+  install : true,
+  install_dir : join_paths(get_option('includedir'), 'janet'),
+  build_by_default : true,
+  output : ['janet.h'],
+  command : [janet_nativeclient, '@INPUT@', '@OUTPUT@'])
 if get_option('peg') and not get_option('reduced_os') and get_option('processes')
   install_man('jpm.1')
   patched_jpm = custom_target('patched-jpm',

meson_options.txt

@@ -17,6 +17,7 @@ option('processes', type : 'boolean', value : true)
 option('umask', type : 'boolean', value : true)
 option('realpath', type : 'boolean', value : true)
 option('simple_getline', type : 'boolean', value : false)
+option('epoll', type : 'boolean', value : false)
 
 option('recursion_guard', type : 'integer', min : 10, max : 8000, value : 1024)
 option('max_proto_depth', type : 'integer', min : 10, max : 8000, value : 200)

src/conf/janetconf.h

@@ -1,27 +1,4 @@
-/*
-* 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.
-*/
-
-/* This is an example janetconf.h file. This will be usually generated
- * by the build system. */
+/* This will be generated by the build system if this file is not used */
 
 #ifndef JANETCONF_H
 #define JANETCONF_H

src/core/fiber.c

@@ -81,7 +81,9 @@ JanetFiber *janet_fiber_reset(JanetFiber *fiber, JanetFunction *callee, int32_t
     }
     if (janet_fiber_funcframe(fiber, callee)) return NULL;
     janet_fiber_frame(fiber)->flags |= JANET_STACKFRAME_ENTRANCE;
+#ifdef JANET_EV
     fiber->waiting = NULL;
+#endif
     return fiber;
 }
 

src/core/gc.c

@@ -404,6 +404,7 @@ void janet_collect(void) {
 #ifdef JANET_EV
     janet_ev_mark();
 #endif
+    janet_mark_fiber(janet_vm_root_fiber);
     for (i = 0; i < orig_rootcount; i++)
         janet_mark(janet_vm_roots[i]);
     while (orig_rootcount < janet_vm_root_count) {

src/core/net.c

@@ -28,6 +28,7 @@
 
 #ifdef JANET_NET
 
+#include <math.h>
 #ifdef JANET_WINDOWS
 #include <winsock2.h>
 #include <windows.h>
@@ -47,127 +48,38 @@
 #include <netinet/tcp.h>
 #include <netdb.h>
 #include <fcntl.h>
-#include <math.h>
 #endif
 
-/*
- * Streams - simple abstract type that wraps a pollable + extra flags
- */
-
-#define JANET_STREAM_READABLE 0x200
-#define JANET_STREAM_WRITABLE 0x400
-#define JANET_STREAM_ACCEPTABLE 0x800
-#define JANET_STREAM_UDPSERVER 0x1000
-
-static int janet_stream_close(void *p, size_t s);
-static int janet_stream_mark(void *p, size_t s);
-static int janet_stream_getter(void *p, Janet key, Janet *out);
-static const JanetAbstractType StreamAT = {
-    "core/stream",
-    janet_stream_close,
-    janet_stream_mark,
-    janet_stream_getter,
-    JANET_ATEND_GET
-};
-
-typedef JanetPollable JanetStream;
-
-static const JanetAbstractType AddressAT = {
+const JanetAbstractType janet_address_type = {
     "core/socket-address",
     JANET_ATEND_NAME
 };
 
 #ifdef JANET_WINDOWS
-#define JANET_NET_CHUNKSIZE 4096
 #define JSOCKCLOSE(x) closesocket((SOCKET) x)
 #define JSOCKDEFAULT INVALID_SOCKET
-#define JLASTERR WSAGetLastError()
 #define JSOCKVALID(x) ((x) != INVALID_SOCKET)
-#define JEINTR WSAEINTR
-#define JEWOULDBLOCK WSAEWOULDBLOCK
-#define JEAGAIN WSAEWOULDBLOCK
-#define JPOLL WSAPoll
 #define JSock SOCKET
-#define JReadInt long
 #define JSOCKFLAGS 0
-static JanetStream *make_stream(SOCKET fd, uint32_t flags) {
-    u_long iMode = 0;
-    JanetStream *stream = janet_abstract(&StreamAT, sizeof(JanetStream));
-    janet_pollable_init(stream, (JanetHandle) fd);
-    ioctlsocket(fd, FIONBIO, &iMode);
-    stream->flags = flags;
-    return stream;
-}
-static Janet net_lasterr(void) {
-    int code = WSAGetLastError();
-    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
 #define JSOCKCLOSE(x) close(x)
 #define JSOCKDEFAULT 0
-#define JLASTERR errno
 #define JSOCKVALID(x) ((x) >= 0)
-#define JEINTR EINTR
-#define JEWOULDBLOCK EWOULDBLOCK
-#define JEAGAIN EAGAIN
-#define JPOLL poll
 #define JSock int
-#define JReadInt ssize_t
 #ifdef SOCK_CLOEXEC
 #define JSOCKFLAGS SOCK_CLOEXEC
 #else
 #define JSOCKFLAGS 0
 #endif
-static JanetStream *make_stream(int fd, uint32_t flags) {
-    JanetStream *stream = janet_abstract(&StreamAT, sizeof(JanetStream));
-    janet_pollable_init(stream, fd);
-#if !defined(SOCK_CLOEXEC) && defined(O_CLOEXEC)
-    int extra = O_CLOEXEC;
-#else
-    int extra = 0;
-#endif
-    fcntl(fd, F_SETFL, fcntl(fd, F_GETFL, 0) | O_NONBLOCK | extra);
-    stream->flags = flags;
-    return stream;
-}
-static Janet net_lasterr(void) {
-    return janet_cstringv(strerror(errno));
-}
 #endif
 
+static JanetStream *make_stream(JSock handle, uint32_t flags);
+
 /* We pass this flag to all send calls to prevent sigpipe */
 #ifndef MSG_NOSIGNAL
 #define MSG_NOSIGNAL 0
 #endif
 
-static int janet_stream_close(void *p, size_t s) {
-    (void) s;
-    JanetStream *stream = p;
-    if (!(stream->flags & JANET_POLL_FLAG_CLOSED)) {
-        JSOCKCLOSE(stream->handle);
-        janet_pollable_deinit(stream);
-    }
-    return 0;
-}
-
 static void nosigpipe(JSock s) {
 #ifdef SO_NOSIGPIPE
     int enable = 1;
@@ -180,358 +92,6 @@ static void nosigpipe(JSock s) {
 #endif
 }
 
-static int janet_stream_mark(void *p, size_t s) {
-    (void) s;
-    janet_pollable_mark((JanetPollable *) p);
-    return 0;
-}
-
-/*
- * State machine for read
- */
-
-typedef struct {
-    JanetListenerState head;
-    int32_t bytes_left;
-    JanetBuffer *buf;
-    int is_chunk;
-    int is_recv_from;
-#ifdef JANET_WINDOWS
-    WSAOVERLAPPED overlapped;
-    WSABUF wbuf;
-    DWORD flags;
-    int32_t chunk_size;
-    struct sockaddr from;
-    int fromlen;
-    uint8_t chunk_buf[JANET_NET_CHUNKSIZE];
-#endif
-} NetStateRead;
-
-JanetAsyncStatus net_machine_read(JanetListenerState *s, JanetAsyncEvent event) {
-    NetStateRead *state = (NetStateRead *) 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 */
-            if (s->bytes == 0 && !state->is_recv_from) {
-                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) {
-                Janet resume_val;
-                if (state->is_recv_from) {
-                    void *abst = janet_abstract(&AddressAT, state->fromlen);
-                    memcpy(abst, &state->from, state->fromlen);
-                    resume_val = janet_wrap_abstract(abst);
-                } else {
-                    resume_val = janet_wrap_buffer(state->buf);
-                }
-                janet_schedule(s->fiber, resume_val);
-                return JANET_ASYNC_STATUS_DONE;
-            }
-        }
-
-        /* fallthrough */
-        case JANET_ASYNC_EVENT_USER:
-        {
-            state->flags = 0;
-            int32_t chunk_size = state->bytes_left > JANET_NET_CHUNKSIZE ? JANET_NET_CHUNKSIZE : state->bytes_left;
-            state->wbuf.len = (ULONG) chunk_size;
-            state->wbuf.buf = state->chunk_buf;
-            state->chunk_size = chunk_size;
-            s->tag = &state->overlapped;
-            memset(&(state->overlapped), 0, sizeof(WSAOVERLAPPED));
-            int status;
-            if (state->is_recv_from) {
-                status = WSARecvFrom((SOCKET) s->pollable->handle, &state->wbuf, 1, NULL, &state->flags, &state->from, &state->fromlen, &state->overlapped, NULL);
-            } else {
-                status = WSARecv((SOCKET) s->pollable->handle, &state->wbuf, 1, NULL, &state->flags, &state->overlapped, NULL);
-            }
-            if (status && WSA_IO_PENDING != WSAGetLastError()) {
-                janet_cancel(s->fiber, net_lasterr());
-                return JANET_ASYNC_STATUS_DONE;
-            }
-        }
-        break;
-#else
-        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);
-            JReadInt nread;
-            char saddr[256];
-            socklen_t socklen = sizeof(saddr);
-            do {
-                if (state->is_recv_from) {
-                    nread = recvfrom(s->pollable->handle, buffer->data + buffer->count, bytes_left, 0,
-                                     (struct sockaddr *)&saddr, &socklen);
-                } else {
-                    nread = recv(s->pollable->handle, buffer->data + buffer->count, bytes_left, 0);
-                }
-            } while (nread == -1 && JLASTERR == JEINTR);
-
-            /* Check for errors - special case errors that can just be waited on to fix */
-            if (nread == -1) {
-                if (JLASTERR == JEAGAIN || JLASTERR == JEWOULDBLOCK) break;
-                janet_cancel(s->fiber, net_lasterr());
-                return JANET_ASYNC_STATUS_DONE;
-            }
-
-            /* Only allow 0-length packets in recv-from. In stream protocols, a zero length packet is EOS. */
-            if (nread == 0 && !state->is_recv_from) {
-                janet_schedule(s->fiber, janet_wrap_nil());
-                return JANET_ASYNC_STATUS_DONE;
-            }
-
-            /* Increment buffer counts */
-            if (nread > 0) {
-                buffer->count += nread;
-                bytes_left -= nread;
-            } else {
-                bytes_left = 0;
-            }
-            state->bytes_left = bytes_left;
-
-            /* Resume if done */
-            if (!state->is_chunk || bytes_left == 0) {
-                Janet resume_val;
-                if (state->is_recv_from) {
-                    void *abst = janet_abstract(&AddressAT, socklen);
-                    memcpy(abst, &saddr, socklen);
-                    resume_val = janet_wrap_abstract(abst);
-                } else {
-                    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;
-}
-
-JANET_NO_RETURN static void janet_sched_read(JanetStream *stream, JanetBuffer *buf, int32_t nbytes) {
-    NetStateRead *state = (NetStateRead *) janet_listen(stream, net_machine_read,
-                          JANET_ASYNC_LISTEN_READ, sizeof(NetStateRead), NULL);
-    state->is_chunk = 0;
-    state->buf = buf;
-    state->bytes_left = nbytes;
-    state->is_recv_from = 0;
-    net_machine_read((JanetListenerState *) state, JANET_ASYNC_EVENT_USER);
-    janet_await();
-}
-
-JANET_NO_RETURN static void janet_sched_chunk(JanetStream *stream, JanetBuffer *buf, int32_t nbytes) {
-    NetStateRead *state = (NetStateRead *) janet_listen(stream, net_machine_read,
-                          JANET_ASYNC_LISTEN_READ, sizeof(NetStateRead), NULL);
-    state->is_chunk = 1;
-    state->buf = buf;
-    state->bytes_left = nbytes;
-    state->is_recv_from = 0;
-    net_machine_read((JanetListenerState *) state, JANET_ASYNC_EVENT_USER);
-    janet_await();
-}
-
-JANET_NO_RETURN static void janet_sched_recv_from(JanetStream *stream, JanetBuffer *buf, int32_t nbytes) {
-    NetStateRead *state = (NetStateRead *) janet_listen(stream, net_machine_read,
-                          JANET_ASYNC_LISTEN_READ, sizeof(NetStateRead), NULL);
-    state->is_chunk = 0;
-    state->buf = buf;
-    state->bytes_left = nbytes;
-    state->is_recv_from = 1;
-    net_machine_read((JanetListenerState *) state, JANET_ASYNC_EVENT_USER);
-    janet_await();
-}
-
-/*
- * State machine for write/send-to
- */
-
-typedef struct {
-    JanetListenerState head;
-    union {
-        JanetBuffer *buf;
-        const uint8_t *str;
-    } src;
-    int32_t start;
-    int is_buffer;
-    void *dest_abst;
-#ifdef JANET_WINDOWS
-    WSAOVERLAPPED overlapped;
-    WSABUF wbuf;
-    DWORD flags;
-#endif
-} NetStateWrite;
-
-JanetAsyncStatus net_machine_write(JanetListenerState *s, JanetAsyncEvent event) {
-    NetStateWrite *state = (NetStateWrite *) 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->dest_abst != NULL) {
-                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->dest_abst) {
-                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;
-            start = state->start;
-            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);
-            }
-            state->wbuf.buf = (char *) bytes;
-            state->wbuf.len = len;
-            state->flags = 0;
-            s->tag = &state->overlapped;
-            memset(&(state->overlapped), 0, sizeof(WSAOVERLAPPED));
-
-            int status;
-            SOCKET sock = (SOCKET) s->pollable->handle;
-            if (state->dest_abst) {
-                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;
-            }
-        }
-        break;
-#else
-        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);
-            }
-            JReadInt nwrote = 0;
-            if (start < len) {
-                int32_t nbytes = len - start;
-                void *dest_abst = state->dest_abst;
-                do {
-                    if (dest_abst) {
-                        nwrote = sendto(s->pollable->handle, bytes + start, nbytes, 0,
-                                        (struct sockaddr *) dest_abst, janet_abstract_size(dest_abst));
-                    } else {
-                        nwrote = send(s->pollable->handle, bytes + start, nbytes, MSG_NOSIGNAL);
-                    }
-                } while (nwrote == -1 && JLASTERR == JEINTR);
-
-                /* Handle write errors */
-                if (nwrote == -1) {
-                    if (JLASTERR == JEAGAIN || JLASTERR == JEWOULDBLOCK) break;
-                    janet_cancel(s->fiber, net_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;
-}
-
-JANET_NO_RETURN static void janet_sched_write_buffer(JanetStream *stream, JanetBuffer *buf, void *dest_abst) {
-    NetStateWrite *state = (NetStateWrite *) janet_listen(stream, net_machine_write,
-                           JANET_ASYNC_LISTEN_WRITE, sizeof(NetStateWrite), NULL);
-    state->is_buffer = 1;
-    state->start = 0;
-    state->src.buf = buf;
-    state->dest_abst = dest_abst;
-    net_machine_write((JanetListenerState *) state, JANET_ASYNC_EVENT_USER);
-    janet_await();
-}
-
-
-JANET_NO_RETURN static void janet_sched_write_stringlike(JanetStream *stream, const uint8_t *str, void *dest_abst) {
-    NetStateWrite *state = (NetStateWrite *) janet_listen(stream, net_machine_write,
-                           JANET_ASYNC_LISTEN_WRITE, sizeof(NetStateWrite), NULL);
-    state->is_buffer = 0;
-    state->start = 0;
-    state->src.str = str;
-    state->dest_abst = dest_abst;
-    net_machine_write((JanetListenerState *) state, JANET_ASYNC_EVENT_USER);
-    janet_await();
-}
-
-/*
- * State machine for simple server
- */
-
 /* State machine for accepting connections. */
 
 #ifdef JANET_WINDOWS
@@ -557,7 +117,7 @@ JanetAsyncStatus net_machine_accept(JanetListenerState *s, JanetAsyncEvent event
             if (state->astream) janet_mark(janet_wrap_abstract(state->astream));
             if (state->function) janet_mark(janet_wrap_abstract(state->function));
             break;
-        } 
+        }
         case JANET_ASYNC_EVENT_CLOSE:
             janet_schedule(s->fiber, janet_wrap_nil());
             return JANET_ASYNC_STATUS_DONE;
@@ -565,16 +125,16 @@ JanetAsyncStatus net_machine_accept(JanetListenerState *s, JanetAsyncEvent event
             int seconds;
             int bytes = sizeof(seconds);
             if (NO_ERROR != getsockopt((SOCKET) state->astream->handle, SOL_SOCKET, SO_CONNECT_TIME,
-                    (char *)&seconds, &bytes)) {
+                                       (char *)&seconds, &bytes)) {
                 janet_cancel(s->fiber, janet_cstringv("failed to accept connection"));
                 return JANET_ASYNC_STATUS_DONE;
             }
-            if (NO_ERROR != setsockopt((SOCKET) state->astream->handle, SOL_SOCKET, SO_UPDATE_ACCEPT_CONTEXT, 
-                (char *)&(state->lstream->handle), sizeof(SOCKET))) {
+            if (NO_ERROR != setsockopt((SOCKET) state->astream->handle, SOL_SOCKET, SO_UPDATE_ACCEPT_CONTEXT,
+                                       (char *) & (state->lstream->handle), sizeof(SOCKET))) {
                 janet_cancel(s->fiber, janet_cstringv("failed to accept connection"));
                 return JANET_ASYNC_STATUS_DONE;
             }
-   
+
             Janet streamv = janet_wrap_abstract(state->astream);
             if (state->function) {
                 /* Schedule worker */
@@ -587,7 +147,7 @@ JanetAsyncStatus net_machine_accept(JanetListenerState *s, JanetAsyncEvent event
                     return JANET_ASYNC_STATUS_DONE;
                 }
             } else {
-                janet_schedule(s->fiber, streamv); 
+                janet_schedule(s->fiber, streamv);
                 return JANET_ASYNC_STATUS_DONE;
             }
         }
@@ -613,7 +173,7 @@ static int net_sched_accept_impl(NetStateAccept *state, Janet *err) {
     SOCKET lsock = (SOCKET) state->lstream->handle;
     SOCKET asock = WSASocketW(AF_INET, SOCK_STREAM, IPPROTO_TCP, NULL, 0, WSA_FLAG_OVERLAPPED);
     if (asock == INVALID_SOCKET) {
-        *err = net_lasterr();
+        *err = janet_ev_lasterr();
         return 1;
     }
     JanetStream *astream = make_stream(asock, JANET_STREAM_READABLE | JANET_STREAM_WRITABLE);
@@ -622,7 +182,7 @@ static int net_sched_accept_impl(NetStateAccept *state, Janet *err) {
     if (FALSE == AcceptEx(lsock, asock, state->buf, 0, socksize, socksize, NULL, &state->overlapped)) {
         int code = WSAGetLastError();
         if (code == WSA_IO_PENDING) return 0; /* indicates io is happening async */
-        *err = net_lasterr();
+        *err = janet_ev_lasterr();
         return 1;
     }
     return 0;
@@ -648,7 +208,7 @@ JanetAsyncStatus net_machine_accept(JanetListenerState *s, JanetAsyncEvent event
             janet_schedule(s->fiber, janet_wrap_nil());
             return JANET_ASYNC_STATUS_DONE;
         case JANET_ASYNC_EVENT_READ: {
-            JSock connfd = accept(s->pollable->handle, NULL, NULL);
+            JSock connfd = accept(s->stream->handle, NULL, NULL);
             if (JSOCKVALID(connfd)) {
                 nosigpipe(connfd);
                 JanetStream *stream = make_stream(connfd, JANET_STREAM_READABLE | JANET_STREAM_WRITABLE);
@@ -751,7 +311,7 @@ static Janet cfun_net_sockaddr(int32_t argc, Janet *argv) {
 #ifndef JANET_WINDOWS
     /* no unix domain socket support on windows yet */
     if (is_unix) {
-        void *abst = janet_abstract(&AddressAT, sizeof(struct sockaddr_un));
+        void *abst = janet_abstract(&janet_address_type, sizeof(struct sockaddr_un));
         memcpy(abst, ai, sizeof(struct sockaddr_un));
         Janet ret = janet_wrap_abstract(abst);
         return make_arr ? janet_wrap_array(janet_array_n(&ret, 1)) : ret;
@@ -762,7 +322,7 @@ static Janet cfun_net_sockaddr(int32_t argc, Janet *argv) {
         JanetArray *arr = janet_array(10);
         struct addrinfo *iter = ai;
         while (NULL != iter) {
-            void *abst = janet_abstract(&AddressAT, iter->ai_addrlen);
+            void *abst = janet_abstract(&janet_address_type, iter->ai_addrlen);
             memcpy(abst, iter->ai_addr, iter->ai_addrlen);
             janet_array_push(arr, janet_wrap_abstract(abst));
             iter = iter->ai_next;
@@ -774,7 +334,7 @@ static Janet cfun_net_sockaddr(int32_t argc, Janet *argv) {
         if (NULL == ai) {
             janet_panic("no data for given address");
         }
-        void *abst = janet_abstract(&AddressAT, ai->ai_addrlen);
+        void *abst = janet_abstract(&janet_address_type, ai->ai_addrlen);
         memcpy(abst, ai->ai_addr, ai->ai_addrlen);
         freeaddrinfo(ai);
         return janet_wrap_abstract(abst);
@@ -928,29 +488,18 @@ static Janet cfun_net_listen(int32_t argc, Janet *argv) {
     }
 }
 
-static void check_stream_flag(JanetStream *stream, int flag) {
-    if (!(stream->flags & flag) || (stream->flags & JANET_POLL_FLAG_CLOSED)) {
-        const char *msg = "";
-        if (flag == JANET_STREAM_READABLE) msg = "readable";
-        if (flag == JANET_STREAM_WRITABLE) msg = "writable";
-        if (flag == JANET_STREAM_ACCEPTABLE) msg = "server";
-        if (flag == JANET_STREAM_UDPSERVER) msg = "datagram server";
-        janet_panicf("bad stream, expected %s stream", msg);
-    }
-}
-
 static Janet cfun_stream_accept_loop(int32_t argc, Janet *argv) {
     janet_fixarity(argc, 2);
-    JanetStream *stream = janet_getabstract(argv, 0, &StreamAT);
-    check_stream_flag(stream, JANET_STREAM_ACCEPTABLE);
+    JanetStream *stream = janet_getabstract(argv, 0, &janet_stream_type);
+    janet_stream_flags(stream, JANET_STREAM_ACCEPTABLE | JANET_STREAM_SOCKET);
     JanetFunction *fun = janet_getfunction(argv, 1);
     janet_sched_accept(stream, fun);
 }
 
 static Janet cfun_stream_accept(int32_t argc, Janet *argv) {
     janet_arity(argc, 1, 2);
-    JanetStream *stream = janet_getabstract(argv, 0, &StreamAT);
-    check_stream_flag(stream, JANET_STREAM_ACCEPTABLE);
+    JanetStream *stream = janet_getabstract(argv, 0, &janet_stream_type);
+    janet_stream_flags(stream, JANET_STREAM_ACCEPTABLE | JANET_STREAM_SOCKET);
     double to = janet_optnumber(argv, argc, 1, INFINITY);
     if (to != INFINITY) janet_addtimeout(to);
     janet_sched_accept(stream, NULL);
@@ -958,85 +507,77 @@ static Janet cfun_stream_accept(int32_t argc, Janet *argv) {
 
 static Janet cfun_stream_read(int32_t argc, Janet *argv) {
     janet_arity(argc, 2, 4);
-    JanetStream *stream = janet_getabstract(argv, 0, &StreamAT);
-    check_stream_flag(stream, JANET_STREAM_READABLE);
+    JanetStream *stream = janet_getabstract(argv, 0, &janet_stream_type);
+    janet_stream_flags(stream, JANET_STREAM_READABLE | JANET_STREAM_SOCKET);
     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_sched_read(stream, buffer, n);
+    janet_ev_recv(stream, buffer, n, MSG_NOSIGNAL);
+    janet_await();
 }
 
 static Janet cfun_stream_chunk(int32_t argc, Janet *argv) {
     janet_arity(argc, 2, 4);
-    JanetStream *stream = janet_getabstract(argv, 0, &StreamAT);
-    check_stream_flag(stream, JANET_STREAM_READABLE);
+    JanetStream *stream = janet_getabstract(argv, 0, &janet_stream_type);
+    janet_stream_flags(stream, JANET_STREAM_READABLE | JANET_STREAM_SOCKET);
     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_sched_chunk(stream, buffer, n);
+    janet_ev_recvchunk(stream, buffer, n, MSG_NOSIGNAL);
+    janet_await();
 }
 
 static Janet cfun_stream_recv_from(int32_t argc, Janet *argv) {
     janet_arity(argc, 3, 4);
-    JanetStream *stream = janet_getabstract(argv, 0, &StreamAT);
-    check_stream_flag(stream, JANET_STREAM_UDPSERVER);
+    JanetStream *stream = janet_getabstract(argv, 0, &janet_stream_type);
+    janet_stream_flags(stream, JANET_STREAM_UDPSERVER | JANET_STREAM_SOCKET);
     int32_t n = janet_getnat(argv, 1);
     JanetBuffer *buffer = janet_getbuffer(argv, 2);
     double to = janet_optnumber(argv, argc, 3, INFINITY);
     if (to != INFINITY) janet_addtimeout(to);
-    janet_sched_recv_from(stream, buffer, n);
-}
-
-static Janet cfun_stream_close(int32_t argc, Janet *argv) {
-    janet_fixarity(argc, 1);
-    JanetStream *stream = janet_getabstract(argv, 0, &StreamAT);
-    janet_stream_close(stream, 0);
-    return janet_wrap_nil();
-}
-
-static Janet cfun_stream_closed(int32_t argc, Janet *argv) {
-    janet_fixarity(argc, 1);
-    JanetStream *stream = janet_getabstract(argv, 0, &StreamAT);
-    return janet_wrap_boolean(stream->flags & JANET_POLL_FLAG_CLOSED);
+    janet_ev_recvfrom(stream, buffer, n, MSG_NOSIGNAL);
+    janet_await();
 }
 
 static Janet cfun_stream_write(int32_t argc, Janet *argv) {
     janet_arity(argc, 2, 3);
-    JanetStream *stream = janet_getabstract(argv, 0, &StreamAT);
-    check_stream_flag(stream, JANET_STREAM_WRITABLE);
+    JanetStream *stream = janet_getabstract(argv, 0, &janet_stream_type);
+    janet_stream_flags(stream, JANET_STREAM_WRITABLE | JANET_STREAM_SOCKET);
     double to = janet_optnumber(argv, argc, 2, INFINITY);
     if (janet_checktype(argv[1], JANET_BUFFER)) {
         if (to != INFINITY) janet_addtimeout(to);
-        janet_sched_write_buffer(stream, janet_getbuffer(argv, 1), NULL);
+        janet_ev_send_buffer(stream, janet_getbuffer(argv, 1), MSG_NOSIGNAL);
     } else {
         JanetByteView bytes = janet_getbytes(argv, 1);
         if (to != INFINITY) janet_addtimeout(to);
-        janet_sched_write_stringlike(stream, bytes.bytes, NULL);
+        janet_ev_send_string(stream, bytes.bytes, MSG_NOSIGNAL);
     }
+    janet_await();
 }
 
 static Janet cfun_stream_send_to(int32_t argc, Janet *argv) {
     janet_arity(argc, 3, 4);
-    JanetStream *stream = janet_getabstract(argv, 0, &StreamAT);
-    check_stream_flag(stream, JANET_STREAM_UDPSERVER);
-    void *dest = janet_getabstract(argv, 1, &AddressAT);
+    JanetStream *stream = janet_getabstract(argv, 0, &janet_stream_type);
+    janet_stream_flags(stream, JANET_STREAM_UDPSERVER | JANET_STREAM_SOCKET);
+    void *dest = janet_getabstract(argv, 1, &janet_address_type);
     double to = janet_optnumber(argv, argc, 3, INFINITY);
     if (janet_checktype(argv[2], JANET_BUFFER)) {
         if (to != INFINITY) janet_addtimeout(to);
-        janet_sched_write_buffer(stream, janet_getbuffer(argv, 2), dest);
+        janet_ev_sendto_buffer(stream, janet_getbuffer(argv, 2), dest, MSG_NOSIGNAL);
     } else {
         JanetByteView bytes = janet_getbytes(argv, 2);
         if (to != INFINITY) janet_addtimeout(to);
-        janet_sched_write_stringlike(stream, bytes.bytes, dest);
+        janet_ev_sendto_string(stream, bytes.bytes, dest, MSG_NOSIGNAL);
     }
+    janet_await();
 }
 
 static Janet cfun_stream_flush(int32_t argc, Janet *argv) {
     janet_fixarity(argc, 2);
-    JanetStream *stream = janet_getabstract(argv, 0, &StreamAT);
-    check_stream_flag(stream, JANET_STREAM_WRITABLE);
+    JanetStream *stream = janet_getabstract(argv, 0, &janet_stream_type);
+    janet_stream_flags(stream, JANET_STREAM_WRITABLE | JANET_STREAM_SOCKET);
     /* Toggle no delay flag */
     int flag = 1;
     setsockopt((JSock) stream->handle, IPPROTO_TCP, TCP_NODELAY, (char *) &flag, sizeof(int));
@@ -1045,10 +586,9 @@ static Janet cfun_stream_flush(int32_t argc, Janet *argv) {
     return argv[0];
 }
 
-static const JanetMethod stream_methods[] = {
+static const JanetMethod net_stream_methods[] = {
     {"chunk", cfun_stream_chunk},
-    {"close", cfun_stream_close},
-    {"closed?", cfun_stream_closed},
+    {"close", janet_cfun_stream_close},
     {"read", cfun_stream_read},
     {"write", cfun_stream_write},
     {"flush", cfun_stream_flush},
@@ -1056,13 +596,15 @@ static const JanetMethod stream_methods[] = {
     {"accept-loop", cfun_stream_accept_loop},
     {"send-to", cfun_stream_send_to},
     {"recv-from", cfun_stream_recv_from},
+    {"recv-from", cfun_stream_recv_from},
+    {"evread", janet_cfun_stream_read},
+    {"evchunk", janet_cfun_stream_chunk},
+    {"evwrite", janet_cfun_stream_write},
     {NULL, NULL}
 };
 
-static int janet_stream_getter(void *p, Janet key, Janet *out) {
-    (void) p;
-    if (!janet_checktype(key, JANET_KEYWORD)) return 0;
-    return janet_getmethod(janet_unwrap_keyword(key), stream_methods, out);
+static JanetStream *make_stream(JSock handle, uint32_t flags) {
+    return janet_stream((JanetHandle) handle, flags | JANET_STREAM_SOCKET, net_stream_methods);
 }
 
 static const JanetReg net_cfuns[] = {
@@ -1071,7 +613,7 @@ static const JanetReg net_cfuns[] = {
         JDOC("(net/address host port &opt type)\n\n"
              "Look up the connection information for a given hostname, port, and connection type. Returns "
              "a handle that can be used to send datagrams over network without establishing a connection. "
-             "On Posix platforms, you can use :unix for host to connet to a unix domain socket, where the name is "
+             "On Posix platforms, you can use :unix for host to connect to a unix domain socket, where the name is "
              "given in the port argument. On Linux, abstract "
              "unix domain sockets are specified with a leading '@' character in port.")
     },
@@ -1079,7 +621,7 @@ static const JanetReg net_cfuns[] = {
         "net/listen", cfun_net_listen,
         JDOC("(net/listen host port &opt type)\n\n"
              "Creates a server. Returns a new stream that is neither readable nor "
-             "writeable. Use net/accept or net/accept-loop be to handle connections and start the server."
+             "writeable. Use net/accept or net/accept-loop be to handle connections and start the server. "
              "The type parameter specifies the type of network connection, either "
              "a :stream (usually tcp), or :datagram (usually udp). If not specified, the default is "
              ":stream. The host and port arguments are the same as in net/address.")
@@ -1137,19 +679,9 @@ static const JanetReg net_cfuns[] = {
              "Make sure that a stream is not buffering any data. This temporarily disables Nagle's algorithm. "
              "Use this to make sure data is sent without delay. Returns stream.")
     },
-    {
-        "net/close", cfun_stream_close,
-        JDOC("(net/close stream)\n\n"
-             "Close a stream so that no further communication can occur.")
-    },
-    {
-        "net/closed?", cfun_stream_closed,
-        JDOC("(net/closed? stream)\n\n"
-             "Check if a stream is closed.")
-    },
     {
         "net/connect", cfun_net_connect,
-        JDOC("(net/connect host porti &opt type)\n\n"
+        JDOC("(net/connect host port &opt type)\n\n"
              "Open a connection to communicate with a server. Returns a duplex stream "
              "that can be used to communicate with the server. Type is an optional keyword "
              "to specify a connection type, either :stream or :datagram. The default is :stream. ")

src/core/os.c

@@ -328,9 +328,15 @@ typedef struct {
     int pid;
 #endif
     int return_code;
+#ifdef JANET_EV
+    JanetStream *in;
+    JanetStream *out;
+    JanetStream *err;
+#else
     JanetFile *in;
     JanetFile *out;
     JanetFile *err;
+#endif
 } JanetProc;
 
 static int janet_proc_gc(void *p, size_t s) {
@@ -434,48 +440,28 @@ static void close_handle(JanetHandle handle) {
    up by the calling function. If everything goes well, *handle is owned by the calling function,
    (if it is set) and the returned JanetFile owns the other end of the pipe, which will be closed
    on GC or fclose. */
-static JanetFile *make_pipes(JanetHandle *handle, int reverse, int *errflag) {
+static JanetHandle make_pipes(JanetHandle *handle, int reverse, int *errflag) {
     JanetHandle handles[2];
 #ifdef JANET_WINDOWS
     SECURITY_ATTRIBUTES saAttr;
     memset(&saAttr, 0, sizeof(saAttr));
     saAttr.nLength = sizeof(saAttr);
     saAttr.bInheritHandle = TRUE;
-    if (!CreatePipe(handles, handles + 1, &saAttr, 0)) goto error_pipe;
+    if (!CreatePipe(handles, handles + 1, &saAttr, 0)) goto error;
     if (reverse) swap_handles(handles);
     /* Don't inherit the side of the pipe owned by this process */
-    if (!SetHandleInformation(handles[0], HANDLE_FLAG_INHERIT, 0)) goto error_set_handle_info;
+    if (!SetHandleInformation(handles[0], HANDLE_FLAG_INHERIT, 0)) goto error;
     *handle = handles[1];
-    int fd = _open_osfhandle((intptr_t) handles[0], reverse ? _O_WRONLY : _O_RDONLY);
-    if (fd == -1) goto error_open_osfhandle;
-    FILE *f = _fdopen(fd, reverse ? "w" : "r");
-    if (NULL == f) goto error_fdopen;
-    return janet_makejfile(f, reverse ? JANET_FILE_WRITE : JANET_FILE_READ);
-error_fdopen:
-    _close(fd); /* we need to close the fake file descriptor instead of the handle, as ownership has been transfered. */
-    *errflag = 1;
-    return NULL;
-error_set_handle_info:
-error_open_osfhandle:
-    close_handle(handles[0]);
-    /* fallthrough */
-error_pipe:
-    *errflag = 1;
-    return NULL;
+    return handles[0];
 #else
-    if (pipe(handles)) goto error_pipe;
+    if (pipe(handles)) goto error;
     if (reverse) swap_handles(handles);
     *handle = handles[1];
-    FILE *f = fdopen(handles[0], reverse ? "w" : "r");
-    if (NULL == f) goto error_fdopen;
-    return janet_makejfile(f, reverse ? JANET_FILE_WRITE : JANET_FILE_READ);
-error_fdopen:
-    close_handle(handles[0]);
-    /* fallthrough */
-error_pipe:
-    *errflag = 1;
-    return NULL;
+    return handles[0];
 #endif
+error:
+    *errflag = 1;
+    return JANET_HANDLE_NONE;
 }
 
 static const JanetMethod proc_methods[] = {
@@ -514,6 +500,81 @@ static const JanetAbstractType ProcAT = {
     JANET_ATEND_GET
 };
 
+static JanetHandle janet_getjstream(Janet *argv, int32_t n, void **orig) {
+#ifdef JANET_EV
+    JanetStream *stream = janet_checkabstract(argv[0], &janet_stream_type);
+    if (stream != NULL) {
+        if (stream->flags & JANET_STREAM_CLOSED)
+            janet_panic("stream is closed");
+        *orig = stream;
+        return stream->handle;
+    }
+#endif
+    JanetFile *f = janet_checkabstract(argv[0], &janet_file_type);
+    if (f != NULL) {
+        if (f->flags & JANET_FILE_CLOSED) {
+            janet_panic("file is closed");
+        }
+        *orig = f;
+#ifdef JANET_WINDOWS
+        return (HANDLE) _get_osfhandle(_fileno(f->file));
+#else
+        return fileno(f->file);
+#endif
+    }
+    janet_panicf("expected file|stream, got %v", argv[n]);
+}
+
+#ifdef JANET_EV
+static JanetStream *get_stdio_for_handle(JanetHandle handle, void *orig, int iswrite) {
+    if (orig == NULL) {
+        return janet_stream(handle, iswrite ? JANET_STREAM_WRITABLE : JANET_STREAM_READABLE, NULL);
+    } else if (janet_abstract_type(orig) == &janet_file_type) {
+        JanetFile *jf = (JanetFile *)orig;
+        uint32_t flags = 0;
+        if (jf->flags & JANET_FILE_WRITE) {
+            flags |= JANET_STREAM_WRITABLE;
+        }
+        if (jf->flags & JANET_FILE_READ) {
+            flags |= JANET_STREAM_READABLE;
+        }
+        /* duplicate handle when converting file to stream */
+#ifdef JANET_WINDOWS
+        HANDLE prochandle = GetCurrentProcess();
+        HANDLE newHandle = INVALID_HANDLE_VALUE;
+        if (!DuplicateHandle(prochandle, handle, prochandle, &newHandle, 0, FALSE, DUPLICATE_SAME_ACCESS)) {
+            return NULL;
+        }
+#else
+        int newHandle = dup(handle);
+        if (newHandle < 0) {
+            return NULL;
+        }
+#endif
+        return janet_stream(newHandle, flags, NULL);
+    } else {
+        return orig;
+    }
+}
+#else
+static JanetFile *get_stdio_for_handle(JanetHandle handle, void *orig, int iswrite) {
+    if (NULL != orig) return (JanetFile *) orig;
+#ifdef JANET_WINDOWS
+    int fd = _open_osfhandle((intptr_t) handle, iswrite ? _O_WRONLY : _O_RDONLY);
+    if (-1 == fd) return NULL;
+    FILE *f = _fdopen(fd, iswrite ? "w" : "r");
+    if (NULL == f) {
+        _close(fd);
+        return NULL;
+    }
+#else
+    FILE *f = fdopen(handle, iswrite ? "w" : "r");
+    if (NULL == f) return NULL;
+#endif
+    return janet_makejfile(f, iswrite ? JANET_FILE_WRITE : JANET_FILE_READ);
+}
+#endif
+
 static Janet os_execute_impl(int32_t argc, Janet *argv, int is_async) {
     janet_arity(argc, 1, 3);
 
@@ -534,7 +595,8 @@ static Janet os_execute_impl(int32_t argc, Janet *argv, int is_async) {
     }
 
     /* Optional stdio redirections */
-    JanetFile *new_in = NULL, *new_out = NULL, *new_err = NULL;
+    JanetAbstract orig_in = NULL, orig_out = NULL, orig_err = NULL;
+    JanetHandle new_in = JANET_HANDLE_NONE, new_out = JANET_HANDLE_NONE, new_err = JANET_HANDLE_NONE;
     JanetHandle pipe_in = JANET_HANDLE_NONE, pipe_out = JANET_HANDLE_NONE, pipe_err = JANET_HANDLE_NONE;
     int pipe_errflag = 0; /* Track errors setting up pipes */
 
@@ -547,17 +609,17 @@ static Janet os_execute_impl(int32_t argc, Janet *argv, int is_async) {
         if (janet_keyeq(maybe_stdin, "pipe")) {
             new_in = make_pipes(&pipe_in, 1, &pipe_errflag);
         } else if (!janet_checktype(maybe_stdin, JANET_NIL)) {
-            new_in = janet_getjfile(&maybe_stdin, 0);
+            new_in = janet_getjstream(&maybe_stdin, 0, &orig_in);
         }
         if (janet_keyeq(maybe_stdout, "pipe")) {
             new_out = make_pipes(&pipe_out, 0, &pipe_errflag);
         } else if (!janet_checktype(maybe_stdout, JANET_NIL)) {
-            new_out = janet_getjfile(&maybe_stdout, 0);
+            new_out = janet_getjstream(&maybe_stdout, 0, &orig_out);
         }
         if (janet_keyeq(maybe_stderr, "err")) {
             new_err = make_pipes(&pipe_err, 0, &pipe_errflag);
         } else if (!janet_checktype(maybe_stderr, JANET_NIL)) {
-            new_err = janet_getjfile(&maybe_stderr, 0);
+            new_err = janet_getjstream(&maybe_stderr, 0, &orig_err);
         }
     }
 
@@ -596,8 +658,8 @@ static Janet os_execute_impl(int32_t argc, Janet *argv, int is_async) {
 
     if (pipe_in != JANET_HANDLE_NONE) {
         startupInfo.hStdInput = pipe_in;
-    } else if (new_in != NULL) {
-        startupInfo.hStdInput = (HANDLE) _get_osfhandle(_fileno(new_in->file));
+    } else if (new_in != JANET_HANDLE_NONE) {
+        startupInfo.hStdInput = new_in;
     } else {
         startupInfo.hStdInput = (HANDLE) _get_osfhandle(0);
     }
@@ -605,8 +667,8 @@ static Janet os_execute_impl(int32_t argc, Janet *argv, int is_async) {
 
     if (pipe_out != JANET_HANDLE_NONE) {
         startupInfo.hStdOutput = pipe_out;
-    } else if (new_out != NULL) {
-        startupInfo.hStdOutput = (HANDLE) _get_osfhandle(_fileno(new_out->file));
+    } else if (new_out != JANET_HANDLE_NONE) {
+        startupInfo.hStdOutput = new_out;
     } else {
         startupInfo.hStdOutput = (HANDLE) _get_osfhandle(1);
     }
@@ -614,7 +676,7 @@ static Janet os_execute_impl(int32_t argc, Janet *argv, int is_async) {
     if (pipe_err != JANET_HANDLE_NONE) {
         startupInfo.hStdError = pipe_err;
     } else if (new_err != NULL) {
-        startupInfo.hStdError = (HANDLE) _get_osfhandle(_fileno(new_err->file));
+        startupInfo.hStdError = new_err;
     } else {
         startupInfo.hStdError = (HANDLE) _get_osfhandle(2);
     }
@@ -680,18 +742,18 @@ static Janet os_execute_impl(int32_t argc, Janet *argv, int is_async) {
     posix_spawn_file_actions_init(&actions);
     if (pipe_in != JANET_HANDLE_NONE) {
         posix_spawn_file_actions_adddup2(&actions, pipe_in, 0);
-    } else if (new_in != NULL) {
-        posix_spawn_file_actions_adddup2(&actions, fileno(new_in->file), 0);
+    } else if (new_in != JANET_HANDLE_NONE) {
+        posix_spawn_file_actions_adddup2(&actions, new_in, 0);
     }
     if (pipe_out != JANET_HANDLE_NONE) {
         posix_spawn_file_actions_adddup2(&actions, pipe_out, 1);
-    } else if (new_out != NULL) {
-        posix_spawn_file_actions_adddup2(&actions, fileno(new_out->file), 1);
+    } else if (new_out != JANET_HANDLE_NONE) {
+        posix_spawn_file_actions_adddup2(&actions, new_out, 1);
     }
     if (pipe_err != JANET_HANDLE_NONE) {
         posix_spawn_file_actions_adddup2(&actions, pipe_err, 2);
-    } else if (new_err != NULL) {
-        posix_spawn_file_actions_adddup2(&actions, fileno(new_err->file), 2);
+    } else if (new_err != JANET_HANDLE_NONE) {
+        posix_spawn_file_actions_adddup2(&actions, new_err, 2);
     }
 
     pid_t pid;
@@ -746,10 +808,13 @@ static Janet os_execute_impl(int32_t argc, Janet *argv, int is_async) {
 #else
         proc->pid = pid;
 #endif
-        proc->in = new_in;
-        proc->out = new_out;
-        proc->err = new_err;
+        proc->in = get_stdio_for_handle(new_in, orig_in, 0);
+        proc->out = get_stdio_for_handle(new_out, orig_out, 1);
+        proc->err = get_stdio_for_handle(new_err, orig_err, 1);
         proc->flags = 0;
+        if (proc->in == NULL || proc->out == NULL || proc->err == NULL) {
+            janet_panic("failed to construct proc");
+        }
         return janet_wrap_abstract(proc);
     } else if (janet_flag_at(flags, 2) && status) {
         janet_panicf("command failed with non-zero exit code %d", status);
@@ -1529,6 +1594,172 @@ static Janet os_permission_int(int32_t argc, Janet *argv) {
     return janet_wrap_integer(os_get_unix_mode(argv, 0));
 }
 
+#ifdef JANET_EV
+
+/*
+ * Define a few functions on streams the require JANET_EV to be defined.
+ */
+
+static jmode_t os_optmode(int32_t argc, const Janet *argv, int32_t n, int32_t dflt) {
+    if (argc > n) return os_getmode(argv, n);
+    return janet_perm_from_unix(dflt);
+}
+
+static Janet os_open(int32_t argc, Janet *argv) {
+    janet_arity(argc, 1, 3);
+    const char *path = janet_getcstring(argv, 0);
+    const uint8_t *opt_flags = janet_optkeyword(argv, argc, 1, (const uint8_t *) "r");
+    jmode_t mode = os_optmode(argc, argv, 2, 0666);
+    uint32_t stream_flags = 0;
+    JanetHandle fd;
+#ifdef JANET_WINDOWS
+    DWORD desiredAccess = 0;
+    DWORD shareMode = 0;
+    DWORD creationDisp = 0;
+    DWORD flagsAndAttributes = FILE_FLAG_OVERLAPPED;
+    /* We map unix-like open flags to the creationDisp parameter */
+    int creatUnix = 0;
+#define OCREAT 1
+#define OEXCL 2
+#define OTRUNC 4
+    for (const uint8_t *c = opt_flags; *c; c++) {
+        switch (*c) {
+            default:
+                break;
+            case 'r':
+                desiredAccess |= GENERIC_READ;
+                stream_flags |= JANET_STREAM_READABLE;
+                break;
+            case 'w':
+                desiredAccess |= GENERIC_WRITE;
+                stream_flags |= JANET_STREAM_WRITABLE;
+                break;
+            case 'c':
+                creatUnix |= OCREAT;
+                break;
+            case 'e':
+                creatUnix |= OEXCL;
+                break;
+            case 't':
+                creatUnix |= OTRUNC;
+                break;
+            /* Windows only flags */
+            case 'D':
+                shareMode |= FILE_SHARE_DELETE;
+                break;
+            case 'R':
+                shareMode |= FILE_SHARE_READ;
+                break;
+            case 'W':
+                shareMode |= FILE_SHARE_WRITE;
+                break;
+            case 'H':
+                flagsAndAttributes |= FILE_ATTRIBUTE_HIDDEN;
+                break;
+            case 'O':
+                flagsAndAttributes |= FILE_ATTRIBUTE_READONLY;
+                break;
+            case 'F':
+                flagsAndAttributes |= FILE_ATTRIBUTE_OFFLINE;
+                break;
+            case 'T':
+                flagsAndAttributes |= FILE_ATTRIBUTE_TEMPORARY;
+                break;
+            case 'd':
+                flagsAndAttributes |= FILE_FLAG_DELETE_ON_CLOSE;
+                break;
+            case 'b':
+                flagsAndAttributes |= FILE_FLAG_NO_BUFFERING;
+                break;
+                /* we could potentially add more here -
+                 * https://docs.microsoft.com/en-us/windows/win32/api/fileapi/nf-fileapi-createfilea
+                 */
+        }
+    }
+    switch (creatUnix) {
+        default:
+            janet_panic("invalid creation flags");
+        case 0:
+            creationDisp = OPEN_EXISTING;
+            break;
+        case OCREAT:
+            creationDisp = OPEN_ALWAYS;
+            break;
+        case OCREAT + OEXCL:
+            creationDisp = CREATE_NEW;
+            break;
+        case OCREAT + OTRUNC:
+            creationDisp = CREATE_ALWAYS;
+            break;
+        case OTRUNC:
+            creationDisp = TRUNCATE_EXISTING;
+            break;
+    }
+    fd = CreateFileA(path, desiredAccess, shareMode, NULL, creationDisp, flagsAndAttributes, NULL);
+    if (fd == INVALID_HANDLE_VALUE) janet_panicv(janet_ev_lasterr());
+#else
+    int open_flags = O_NONBLOCK;
+#ifdef JANET_LINUX
+    open_flags |= O_CLOEXEC;
+#endif
+    for (const uint8_t *c = opt_flags; *c; c++) {
+        switch (*c) {
+            default:
+                break;
+            case 'r':
+                open_flags = (open_flags & O_WRONLY)
+                             ? ((open_flags & ~O_WRONLY) | O_RDWR)
+                             : (open_flags | O_RDONLY);
+                stream_flags |= JANET_STREAM_READABLE;
+                break;
+            case 'w':
+                open_flags = (open_flags & O_RDONLY)
+                             ? ((open_flags & ~O_RDONLY) | O_RDWR)
+                             : (open_flags | O_WRONLY);
+                stream_flags |= JANET_STREAM_WRITABLE;
+                break;
+            case 'c':
+                open_flags |= O_CREAT;
+                break;
+            case 'e':
+                open_flags |= O_EXCL;
+                break;
+            case 't':
+                open_flags |= O_TRUNC;
+                break;
+            /* posix only */
+            case 'x':
+                open_flags |= O_SYNC;
+                break;
+            case 'C':
+                open_flags |= O_NOCTTY;
+                break;
+            case 'a':
+                open_flags |= O_APPEND;
+                break;
+        }
+    }
+    do {
+        fd = open(path, open_flags, mode);
+    } while (fd == -1 && errno == EINTR);
+    if (fd == -1) janet_panicv(janet_ev_lasterr());
+#endif
+    return janet_wrap_abstract(janet_stream(fd, stream_flags, NULL));
+}
+
+static Janet os_pipe(int32_t argc, Janet *argv) {
+    (void) argv;
+    janet_fixarity(argc, 0);
+    JanetHandle fds[2];
+    if (janet_make_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);
+    Janet tup[2] = {janet_wrap_abstract(reader), janet_wrap_abstract(writer)};
+    return janet_wrap_tuple(janet_tuple_n(tup, 2));
+}
+
+#endif
+
 #endif /* JANET_REDUCED_OS */
 
 static const JanetReg os_cfuns[] = {
@@ -1798,6 +2029,42 @@ static const JanetReg os_cfuns[] = {
         JDOC("(os/perm-int bytes)\n\n"
              "Parse a 9 character permission string and return an integer that can be used by chmod.")
     },
+#ifdef JANET_EV
+    {
+        "os/open", os_open,
+        JDOC("(os/open path &opt flags mode)\n\n"
+             "Create a stream from a file, like the POSIX open system call. Returns a new stream. "
+             "mode should be a file mode as passed to os/chmod, but only if the create flag is given. "
+             "The default mode is 8r666. "
+             "Allowed flags are as follows:\n\n"
+             "\t:r - open this file for reading\n"
+             "\t:w - open this file for writing\n"
+             "\t:c - create a new file (O_CREATE)\n"
+             "\t:e - fail if the file exists (O_EXCL)\n"
+             "\t:t - shorten an existing file to length 0 (O_TRUNC)\n\n"
+             "Posix only flags:\n"
+             "\t:a - append to a file (O_APPEND)\n"
+             "\t:x - O_SYNC\n"
+             "\t:C - O_NOCTTY\n\n"
+             "Windows only flags:\n"
+             "\t:R - share reads (FILE_SHARE_READ)\n"
+             "\t:W - share writes (FILE_SHARE_WRITE)\n"
+             "\t:D - share deletes (FILE_SHARE_DELETE)\n"
+             "\t:H - FILE_ATTRIBUTE_HIDDEN\n"
+             "\t:O - FILE_ATTRIBUTE_READONLY\n"
+             "\t:F - FILE_ATTRIBUTE_OFFLINE\n"
+             "\t:T - FILE_ATTRIBUTE_TEMPORARY\n"
+             "\t:d - FILE_FLAG_DELETE_ON_CLOSE\n"
+             "\t:b - FILE_FLAG_NO_BUFFERING\n")
+    },
+    {
+        "os/pipe", os_pipe,
+        JDOC("(os/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.")
+    },
+#endif
 #endif
     {NULL, NULL, NULL}
 };

src/core/parse.c

@@ -313,11 +313,48 @@ static int stringend(JanetParser *p, JanetParseState *state) {
     uint8_t *bufstart = p->buf;
     int32_t buflen = (int32_t) p->bufcount;
     if (state->flags & PFLAG_LONGSTRING) {
-        /* Check for leading newline character so we can remove it */
-        if (bufstart[0] == '\n') {
-            bufstart++;
-            buflen--;
+        /* Post process to remove leading whitespace */
+        JanetParseState top = p->states[p->statecount - 1];
+        int32_t indent_col = (int32_t) top.column - 1;
+        uint8_t *r = bufstart, *end = r + buflen;
+        /* Check if there are any characters before the start column -
+         * if so, do not reindent. */
+        int reindent = 1;
+        while (reindent && (r < end)) {
+            if (*r++ == '\n') {
+                for (int32_t j = 0; (r < end) && (*r != '\n') && (j < indent_col); j++, r++) {
+                    if (*r != ' ') {
+                        reindent = 0;
+                        break;
+                    }
+                }
+            }
         }
+        /* Now reindent if able to, otherwise just drop leading newline. */
+        if (!reindent) {
+            if (buflen > 0 && bufstart[0] == '\n') {
+                buflen--;
+                bufstart++;
+            }
+        } else {
+            uint8_t *w = bufstart;
+            r = bufstart;
+            while (r < end) {
+                if (*r == '\n') {
+                    if (r == bufstart) {
+                        /* Skip leading newline */
+                        r++;
+                    } else {
+                        *w++ = *r++;
+                    }
+                    for (int32_t j = 0; (r < end) && (*r != '\n') && (j < indent_col); j++, r++);
+                } else {
+                    *w++ = *r++;
+                }
+            }
+            buflen = (int32_t)(w - bufstart);
+        }
+        /* Check for trailing newline character so we can remove it */
         if (buflen > 0 && bufstart[buflen - 1] == '\n') {
             buflen--;
         }

src/core/peg.c

@@ -45,8 +45,10 @@ typedef struct {
     JanetBuffer *scratch;
     JanetBuffer *tags;
     const Janet *extrav;
+    int32_t *linemap;
     int32_t extrac;
     int32_t depth;
+    int32_t linemaplen;
     enum {
         PEG_MODE_NORMAL,
         PEG_MODE_ACCUMULATE
@@ -87,6 +89,57 @@ static void pushcap(PegState *s, Janet capture, uint32_t tag) {
     }
 }
 
+/* Lazily generate line map to get line and column information for PegState.
+ * line and column are 1-indexed. */
+typedef struct {
+    int32_t line;
+    int32_t col;
+} LineCol;
+static LineCol get_linecol_from_position(PegState *s, int32_t position) {
+    /* Generate if not made yet */
+    if (s->linemaplen < 0) {
+        int32_t newline_count = 0;
+        for (const uint8_t *c = s->text_start; c < s->text_end; c++) {
+            if (*c == '\n') newline_count++;
+        }
+        int32_t *mem = janet_smalloc(sizeof(int32_t) * newline_count);
+        size_t index = 0;
+        for (const uint8_t *c = s->text_start; c < s->text_end; c++) {
+            if (*c == '\n') mem[index++] = (int32_t)(c - s->text_start);
+        }
+        s->linemaplen = newline_count;
+        s->linemap = mem;
+    }
+    /* Do binary search for line. Slightly modified from classic binary search:
+     * - if we find that our current character is a line break, just return immediately.
+     *   a newline character is consider to be on the same line as the character before
+     *   (\n is line terminator, not line separator).
+     * - in the not-found case, we still want to find the greatest-indexed newline that
+     *   is before position. we use that to calcuate the line and column.
+     * - in the case that lo = 0 and s->linemap[0] is still greater than position, we
+     *   are on the first line and our column is position + 1. */
+    int32_t hi = s->linemaplen; /* hi is greater than the actual line */
+    int32_t lo = 0; /* lo is less than or equal to the actual line */
+    LineCol ret;
+    while (lo + 1 < hi) {
+        int32_t mid = lo + (hi - lo) / 2;
+        if (s->linemap[mid] >= position) {
+            hi = mid;
+        } else {
+            lo = mid;
+        }
+    }
+    /* first line case */
+    if (s->linemaplen == 0 || (lo == 0 && s->linemap[0] >= position)) {
+        ret.line = 1;
+        ret.col = position + 1;
+    } else {
+        ret.line = lo + 2;
+        ret.col = position - s->linemap[lo];
+    }
+    return ret;
+}
+
 /* Convert a uint64_t to a int64_t by wrapping to a maximum number of bytes */
 static int64_t peg_convert_u64_s64(uint64_t from, int width) {
     int shift = 8 * (8 - width);
@@ -280,6 +333,18 @@ tail:
             return text;
         }
 
+        case RULE_LINE: {
+            LineCol lc = get_linecol_from_position(s, (int32_t)(text - s->text_start));
+            pushcap(s, janet_wrap_number((double)(lc.line)), rule[1]);
+            return text;
+        }
+
+        case RULE_COLUMN: {
+            LineCol lc = get_linecol_from_position(s, (int32_t)(text - s->text_start));
+            pushcap(s, janet_wrap_number((double)(lc.col)), rule[1]);
+            return text;
+        }
+
         case RULE_ARGUMENT: {
             int32_t index = ((int32_t *)rule)[1];
             Janet capture = (index >= s->extrac) ? janet_wrap_nil() : s->extrav[index];
@@ -420,8 +485,8 @@ tail:
             } else {
                 /* Throw generic error */
                 int32_t start = (int32_t)(text - s->text_start);
-                int32_t end = (int32_t)(result - s->text_start);
-                janet_panicf("match error in range (%d:%d)", start, end);
+                LineCol lc = get_linecol_from_position(s, start);
+                janet_panicf("match error at line %d, column %d", lc.line, lc.col);
             }
             return NULL;
         }
@@ -830,7 +895,13 @@ static void spec_not(Builder *b, int32_t argc, const Janet *argv) {
     spec_onerule(b, argc, argv, RULE_NOT);
 }
 static void spec_error(Builder *b, int32_t argc, const Janet *argv) {
-    spec_onerule(b, argc, argv, RULE_ERROR);
+    if (argc == 0) {
+        Reserve r = reserve(b, 2);
+        uint32_t rule = peg_compile1(b, janet_wrap_number(0));
+        emit_1(r, RULE_ERROR, rule);
+    } else {
+        spec_onerule(b, argc, argv, RULE_ERROR);
+    }
 }
 static void spec_drop(Builder *b, int32_t argc, const Janet *argv) {
     spec_onerule(b, argc, argv, RULE_DROP);
@@ -880,6 +951,12 @@ static void spec_tag1(Builder *b, int32_t argc, const Janet *argv, uint32_t op)
 static void spec_position(Builder *b, int32_t argc, const Janet *argv) {
     spec_tag1(b, argc, argv, RULE_POSITION);
 }
+static void spec_line(Builder *b, int32_t argc, const Janet *argv) {
+    spec_tag1(b, argc, argv, RULE_LINE);
+}
+static void spec_column(Builder *b, int32_t argc, const Janet *argv) {
+    spec_tag1(b, argc, argv, RULE_COLUMN);
+}
 
 static void spec_backmatch(Builder *b, int32_t argc, const Janet *argv) {
     spec_tag1(b, argc, argv, RULE_BACKMATCH);
@@ -983,6 +1060,7 @@ static const SpecialPair peg_specials[] = {
     {"capture", spec_capture},
     {"choice", spec_choice},
     {"cmt", spec_matchtime},
+    {"column", spec_column},
     {"constant", spec_constant},
     {"drop", spec_drop},
     {"error", spec_error},
@@ -992,6 +1070,7 @@ static const SpecialPair peg_specials[] = {
     {"int", spec_int_le},
     {"int-be", spec_int_be},
     {"lenprefix", spec_lenprefix},
+    {"line", spec_line},
     {"look", spec_look},
     {"not", spec_not},
     {"opt", spec_opt},
@@ -1229,6 +1308,8 @@ static void *peg_unmarshal(JanetMarshalContext *ctx) {
             case RULE_NOTNCHAR:
             case RULE_RANGE:
             case RULE_POSITION:
+            case RULE_LINE:
+            case RULE_COLUMN:
             case RULE_BACKMATCH:
                 /* [1 word] */
                 i += 2;
@@ -1438,6 +1519,8 @@ static PegCall peg_cfun_init(int32_t argc, Janet *argv, int get_replace) {
     ret.s.tags = janet_buffer(10);
     ret.s.constants = ret.peg->constants;
     ret.s.bytecode = ret.peg->bytecode;
+    ret.s.linemap = NULL;
+    ret.s.linemaplen = -1;
     return ret;
 }
 

src/core/specials.c

@@ -336,10 +336,8 @@ static int defleaf(
 
         /* Put value in table when evaulated */
         janetc_emit_sss(c, JOP_PUT, tabslot, valsym, s, 0);
-        return 1;
-    } else {
-        return namelocal(c, sym, 0, s);
     }
+    return namelocal(c, sym, 0, s);
 }
 
 static JanetSlot janetc_def(JanetFopts opts, int32_t argn, const Janet *argv) {
@@ -470,6 +468,28 @@ static JanetSlot janetc_do(JanetFopts opts, int32_t argn, const Janet *argv) {
     return ret;
 }
 
+
+/* Compile an upscope form. Upscope forms execute their body sequentially and
+ * evaluate to the last expression in the body, but without lexical scope. */
+static JanetSlot janetc_upscope(JanetFopts opts, int32_t argn, const Janet *argv) {
+    int32_t i;
+    JanetSlot ret = janetc_cslot(janet_wrap_nil());
+    JanetCompiler *c = opts.compiler;
+    JanetFopts subopts = janetc_fopts_default(c);
+    for (i = 0; i < argn; i++) {
+        if (i != argn - 1) {
+            subopts.flags = JANET_FOPTS_DROP;
+        } else {
+            subopts = opts;
+        }
+        ret = janetc_value(subopts, argv[i]);
+        if (i != argn - 1) {
+            janetc_freeslot(c, ret);
+        }
+    }
+    return ret;
+}
+
 /* Add a funcdef to the top most function scope */
 static int32_t janetc_addfuncdef(JanetCompiler *c, JanetFuncDef *def) {
     JanetScope *scope = c->scope;
@@ -854,6 +874,7 @@ static const JanetSpecial janetc_specials[] = {
     {"set", janetc_varset},
     {"splice", janetc_splice},
     {"unquote", janetc_unquote},
+    {"upscope", janetc_upscope},
     {"var", janetc_var},
     {"while", janetc_while}
 };

src/core/strtod.c

@@ -447,7 +447,7 @@ int janet_scan_int64(const uint8_t *str, int32_t len, int64_t *out) {
     int neg;
     uint64_t bi;
     if (scan_uint64(str, len, &bi, &neg)) {
-        if (neg && bi <= (UINT64_MAX / 2)) {
+        if (neg && bi <= ((UINT64_MAX / 2) + 1)) {
             if (bi > INT64_MAX) {
                 *out = INT64_MIN;
             } else {

src/core/thread.c

@@ -585,6 +585,14 @@ void janet_threads_deinit(void) {
     janet_vm_thread_decode = NULL;
 }
 
+JanetThread *janet_thread_current(void) {
+    if (NULL == janet_vm_thread_current) {
+        janet_vm_thread_current = janet_make_thread(janet_vm_mailbox, janet_get_core_table("make-image-dict"));
+        janet_gcroot(janet_wrap_abstract(janet_vm_thread_current));
+    }
+    return janet_vm_thread_current;
+}
+
 /*
  * Cfuns
  */
@@ -592,11 +600,7 @@ void janet_threads_deinit(void) {
 static Janet cfun_thread_current(int32_t argc, Janet *argv) {
     (void) argv;
     janet_fixarity(argc, 0);
-    if (NULL == janet_vm_thread_current) {
-        janet_vm_thread_current = janet_make_thread(janet_vm_mailbox, janet_get_core_table("make-image-dict"));
-        janet_gcroot(janet_wrap_abstract(janet_vm_thread_current));
-    }
-    return janet_wrap_abstract(janet_vm_thread_current);
+    return janet_wrap_abstract(janet_thread_current());
 }
 
 static Janet cfun_thread_new(int32_t argc, Janet *argv) {

src/core/util.h

@@ -140,10 +140,12 @@ void janet_lib_thread(JanetTable *env);
 #endif
 #ifdef JANET_NET
 void janet_lib_net(JanetTable *env);
+extern const JanetAbstractType janet_address_type;
 #endif
 #ifdef JANET_EV
 void janet_lib_ev(JanetTable *env);
 void janet_ev_mark(void);
+int janet_make_pipe(JanetHandle handles[2]);
 #endif
 
 #endif

src/core/vm.c

@@ -1376,7 +1376,6 @@ static JanetSignal janet_continue_no_check(JanetFiber *fiber, Janet in, Janet *o
         /* Normal setup */
         if (janet_vm_root_fiber == NULL) janet_vm_root_fiber = fiber;
         janet_vm_fiber = fiber;
-        janet_gcroot(janet_wrap_fiber(fiber));
         janet_fiber_set_status(fiber, JANET_STATUS_ALIVE);
         signal = run_vm(fiber, in);
     }
@@ -1384,7 +1383,6 @@ static JanetSignal janet_continue_no_check(JanetFiber *fiber, Janet in, Janet *o
     /* Restore */
     if (janet_vm_root_fiber == fiber) janet_vm_root_fiber = NULL;
     janet_fiber_set_status(fiber, signal);
-    janet_gcunroot(janet_wrap_fiber(fiber));
     janet_restore(&tstate);
     *out = tstate.payload;
 

src/include/janet.h

@@ -20,6 +20,8 @@
 * IN THE SOFTWARE.
 */
 
+#include "janetconf.h"
+
 #ifndef JANET_H_defined
 #define JANET_H_defined
 
@@ -35,8 +37,6 @@ extern "C" {
 
 /***** START SECTION CONFIG *****/
 
-#include "janetconf.h"
-
 #ifndef JANET_VERSION
 #define JANET_VERSION "latest"
 #endif
@@ -499,25 +499,31 @@ typedef void *JanetAbstract;
 
 /* Event Loop Types */
 #ifdef JANET_EV
-#define JANET_POLL_FLAG_CLOSED 0x1
-#define JANET_POLL_FLAG_SOCKET 0x2
-#define JANET_POLL_FLAG_IOCP 0x4
+
+#define JANET_STREAM_CLOSED 0x1
+#define JANET_STREAM_SOCKET 0x2
+#define JANET_STREAM_IOCP 0x4
+#define JANET_STREAM_READABLE 0x200
+#define JANET_STREAM_WRITABLE 0x400
+#define JANET_STREAM_ACCEPTABLE 0x800
+#define JANET_STREAM_UDPSERVER 0x1000
 
 typedef enum {
     JANET_ASYNC_EVENT_INIT,
     JANET_ASYNC_EVENT_MARK,
     JANET_ASYNC_EVENT_DEINIT,
     JANET_ASYNC_EVENT_CLOSE,
+    JANET_ASYNC_EVENT_ERR,
+    JANET_ASYNC_EVENT_HUP,
     JANET_ASYNC_EVENT_READ,
     JANET_ASYNC_EVENT_WRITE,
-    JANET_ASYNC_EVENT_TIMEOUT,
+    JANET_ASYNC_EVENT_CANCEL,
     JANET_ASYNC_EVENT_COMPLETE, /* Used on windows for IOCP */
     JANET_ASYNC_EVENT_USER
 } JanetAsyncEvent;
 
 #define JANET_ASYNC_LISTEN_READ (1 << JANET_ASYNC_EVENT_READ)
 #define JANET_ASYNC_LISTEN_WRITE (1 << JANET_ASYNC_EVENT_WRITE)
-#define JANET_ASYNC_LISTEN_SPAWNER 0x1000
 
 typedef enum {
     JANET_ASYNC_STATUS_NOT_DONE,
@@ -526,18 +532,19 @@ typedef enum {
 
 /* Typedefs */
 typedef struct JanetListenerState JanetListenerState;
-typedef struct JanetPollable JanetPollable;
+typedef struct JanetStream JanetStream;
 typedef JanetAsyncStatus(*JanetListener)(JanetListenerState *state, JanetAsyncEvent event);
 
 /* Wrapper around file descriptors and HANDLEs that can be polled. */
-struct JanetPollable {
+struct JanetStream {
     JanetHandle handle;
     uint32_t flags;
-    /* Linked list of all in-flight IO routines for this pollable */
+    /* Linked list of all in-flight IO routines for this stream */
     JanetListenerState *state;
-    /* internal - used to disallow multiple concurrent reads / writes on the same pollable.
+    const void *methods; /* Methods for this stream */
+    /* internal - used to disallow multiple concurrent reads / writes on the same stream.
      * 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. */
+     * for some implementations. You can read and write at the same time on the same stream, though. */
     int _mask;
 };
 
@@ -545,7 +552,7 @@ struct JanetPollable {
 struct JanetListenerState {
     JanetListener machine;
     JanetFiber *fiber;
-    JanetPollable *pollable;
+    JanetStream *stream;
     void *event; /* Used to pass data from asynchronous IO event. Contents depend on both
                     implementation of the event loop and the particular event. */
 #ifdef JANET_WINDOWS
@@ -553,7 +560,7 @@ struct JanetListenerState {
     int bytes; /* Used to track how many bytes were transfered. */
 #endif
     /* internal */
-    int _index; /* not used in all implementations */
+    size_t _index;
     int _mask;
     JanetListenerState *_next;
 };
@@ -1248,28 +1255,56 @@ extern enum JanetInstructionType janet_instructions[JOP_INSTRUCTION_COUNT];
 
 #ifdef JANET_EV
 
+extern JANET_API const JanetAbstractType janet_stream_type;
+
 /* Run the event loop */
 JANET_API void janet_loop(void);
 
-/* Wrapper around pollables */
-JANET_API void janet_pollable_init(JanetPollable *pollable, JanetHandle handle);
-JANET_API void janet_pollable_mark(JanetPollable *pollable);
-JANET_API void janet_pollable_deinit(JanetPollable *pollable);
+/* Wrapper around streams */
+JANET_API JanetStream *janet_stream(JanetHandle handle, uint32_t flags, const JanetMethod *methods);
+JANET_API void janet_stream_close(JanetStream *stream);
+JANET_API Janet janet_cfun_stream_close(int32_t argc, Janet *argv);
+JANET_API Janet janet_cfun_stream_read(int32_t argc, Janet *argv);
+JANET_API Janet janet_cfun_stream_chunk(int32_t argc, Janet *argv);
+JANET_API Janet janet_cfun_stream_write(int32_t argc, Janet *argv);
+JANET_API void janet_stream_flags(JanetStream *stream, uint32_t flags);
 
 /* Queue a fiber to run on the event loop */
 JANET_API void janet_schedule(JanetFiber *fiber, Janet value);
 JANET_API void janet_cancel(JanetFiber *fiber, Janet value);
 JANET_API void janet_schedule_signal(JanetFiber *fiber, Janet value, JanetSignal sig);
 
-/* Start a state machine listening for events from a pollable */
-JANET_API JanetListenerState *janet_listen(JanetPollable *pollable, JanetListener behavior, int mask, size_t size, void *user);
+/* Start a state machine listening for events from a stream */
+JANET_API JanetListenerState *janet_listen(JanetStream *stream, JanetListener behavior, int mask, size_t size, void *user);
 
 /* Shorthand for yielding to event loop in C */
 JANET_NO_RETURN JANET_API void janet_await(void);
 
 /* For use inside listeners - adds a timeout to the current fiber, such that
  * it will be resumed after sec seconds if no other event schedules the current fiber. */
-void janet_addtimeout(double sec);
+JANET_API void janet_addtimeout(double sec);
+
+/* Get last error from a an IO operation */
+JANET_API Janet janet_ev_lasterr(void);
+
+/* Read async from a stream */
+JANET_API void janet_ev_read(JanetStream *stream, JanetBuffer *buf, int32_t nbytes);
+JANET_API void janet_ev_readchunk(JanetStream *stream, JanetBuffer *buf, int32_t nbytes);
+#ifdef JANET_NET
+JANET_API void janet_ev_recv(JanetStream *stream, JanetBuffer *buf, int32_t nbytes, int flags);
+JANET_API void janet_ev_recvchunk(JanetStream *stream, JanetBuffer *buf, int32_t nbytes, int flags);
+JANET_API void janet_ev_recvfrom(JanetStream *stream, JanetBuffer *buf, int32_t nbytes, int flags);
+#endif
+
+/* Write async to a stream */
+JANET_API void janet_ev_write_buffer(JanetStream *stream, JanetBuffer *buf);
+JANET_API void janet_ev_write_string(JanetStream *stream, JanetString str);
+#ifdef JANET_NET
+JANET_API void janet_ev_send_buffer(JanetStream *stream, JanetBuffer *buf, int flags);
+JANET_API void janet_ev_send_string(JanetStream *stream, JanetString str, int flags);
+JANET_API void janet_ev_sendto_buffer(JanetStream *stream, JanetBuffer *buf, void *dest, int flags);
+JANET_API void janet_ev_sendto_string(JanetStream *stream, JanetString str, void *dest, int flags);
+#endif
 
 #endif
 
@@ -1581,6 +1616,9 @@ JANET_API Janet janet_resolve_core(const char *name);
 
 /* New C API */
 
+/* Shorthand for janet C function declarations */
+#define JANET_CFUN(name) Janet name (int32_t argc, Janet *argv)
+
 /* Allow setting entry name for static libraries */
 #ifdef __cplusplus
 #define JANET_MODULE_PREFIX extern "C"
@@ -1746,7 +1784,9 @@ typedef enum {
     RULE_THRU,         /* [rule] */
     RULE_LENPREFIX,    /* [rule_a, rule_b (repeat rule_b rule_a times)] */
     RULE_READINT,      /* [(signedness << 4) | (endianess << 5) | bytewidth, tag] */
-} JanetPegOpcode;
+    RULE_LINE,         /* [tag] */
+    RULE_COLUMN        /* [tag] */
+} JanetPegOpcod;
 
 typedef struct {
     uint32_t *bytecode;
@@ -1837,6 +1877,7 @@ extern JANET_API const JanetAbstractType janet_thread_type;
 
 JANET_API int janet_thread_receive(Janet *msg_out, double timeout);
 JANET_API int janet_thread_send(JanetThread *thread, Janet msg, double timeout);
+JANET_API JanetThread *janet_thread_current(void);
 
 #endif
 

test/helper.janet

@@ -15,8 +15,8 @@
   (def truncated
     (if (> (length e) 40) (string (string/slice e 0 35) "...") (string e)))
   (if x
-    (xprintf stdout "\e[32m✔\e[0m %s: %v" truncated x)
-    (xprintf stdout "\n\e[31m✘\e[0m %s: %v" truncated x))
+    (eprintf "\e[32m✔\e[0m %s: %v" truncated x)
+    (eprintf "\n\e[31m✘\e[0m %s: %v" truncated x))
   x)
 
 (defmacro assert-error
@@ -32,10 +32,10 @@
 (defn start-suite [x]
   (set suite-num x)
   (set start-time (os/clock))
-  (print "\nRunning test suite " x " tests...\n  "))
+  (eprint "\nRunning test suite " x " tests...\n  "))
 
 (defn end-suite []
   (def delta (- (os/clock) start-time))
-  (printf "\n\nTest suite %d finished in %.3f seconds" suite-num delta)
-  (print num-tests-passed " of " num-tests-run " tests passed.\n")
+  (eprintf "\n\nTest suite %d finished in %.3f seconds" suite-num delta)
+  (eprint num-tests-passed " of " num-tests-run " tests passed.\n")
   (if (not= num-tests-passed num-tests-run) (os/exit 1)))

test/suite0003.janet

@@ -360,6 +360,14 @@
 (check-match janet-longstring "``` `` ```" true)
 (check-match janet-longstring "``  ```" false)
 
+# Line and column capture
+
+(def line-col (peg/compile '(any (* (line) (column) 1))))
+(check-deep line-col "abcd" @[1 1 1 2 1 3 1 4])
+(check-deep line-col "" @[])
+(check-deep line-col "abcd\n" @[1 1 1 2 1 3 1 4 1 5])
+(check-deep line-col "abcd\nz" @[1 1 1 2 1 3 1 4 1 5 2 1])
+
 # Backmatch
 
 (def backmatcher-1 '(* (capture (any "x") :1) "y" (backmatch :1) -1))

test/suite0009.janet

@@ -33,15 +33,14 @@
     (:write stream b)
     (buffer/clear b)))
 
-(def s (net/server "127.0.0.1" "8000"))
+(def s (net/server "127.0.0.1" "8000" handler))
 (assert s "made server 1")
 
-(ev/go
-  (coro
-    (while (not (net/closed? s))
-      (def conn (net/accept s))
-      (unless conn (break))
-      (ev/call handler conn))))
+# We need some sleep for windows to let the server stabilize
+# or else the first read can fail. Might be a strange windows
+# "bug", but needs further investigating. Otherwise, `build_win test`
+# can sometimes fail on windows, leading to flaky testing.
+(ev/sleep 0.2)
 
 (defn test-echo [msg]
   (with [conn (net/connect "127.0.0.1" "8000")]
@@ -55,4 +54,32 @@
 
 (:close s)
 
+# Create pipe
+
+(var pipe-counter 0)
+(def chan (ev/chan 10))
+(let [[reader writer] (os/pipe)]
+  (ev/spawn
+    (while (ev/read reader 3)
+      (++ pipe-counter))
+    (assert (= 20 pipe-counter) "ev/pipe 1")
+    (ev/give chan 1))
+
+  (for i 0 10
+    (ev/write writer "xxx---"))
+
+  (ev/close writer)
+  (ev/take chan))
+
+(var result nil)
+(def fiber
+  (ev/spawn 
+    (set result (protect (ev/sleep 0.4)))
+    (assert (= result '(false "boop")) "ev/cancel 1")))
+(ev/sleep 0.1)
+(ev/cancel fiber "boop")
+(ev/sleep 0.1)
+
+(assert-error "bad arity to ev/call" (ev/call inc 1 2 3))
+
 (end-suite)

test/suite0010.janet

@@ -65,4 +65,8 @@
 (assert (= :brackets (tuple/type (postwalk identity '[]))) "walk square brackets 1")
 (assert (= :brackets (tuple/type (walk identity '[]))) "walk square brackets 2")
 
+# # off by 1 error in inttypes
+(assert (= (int/s64 "-0x8000_0000_0000_0000") (+ (int/s64 "0x7FFF_FFFF_FFFF_FFFF") 1)) "int types wrap around")
+
+
 (end-suite)

tools/msi/janet.wxs

@@ -126,9 +126,6 @@
                 <File Source="dist\janet.h"/>
                 <RemoveFolder Id="RemoveCDir" On="uninstall" />
             </Component>
-            <Component Directory="CDir">
-                <File Source="dist\janetconf.h"/>
-            </Component>
             <Component Directory="CDir">
                 <File Source="dist\janet.lib"/>
             </Component>

tools/patch-header.janet

@@ -0,0 +1,3 @@
+# Patch janet.h 
+(def [_ janeth janetconf output] (dyn :args))
+(spit output (string/replace `#include "janetconf.h"` (slurp janetconf) (slurp janeth)))