Fix meson install test.

Make more use of the built in GC code for abstracts to
be sure things are more correct. Issue before was streams could
be freed before IOCP events arrived.

CHANGELOG.md

@@ -1,6 +1,17 @@
 # Changelog
 All notable changes to this project will be documented in this file.
 
+## 1.31.0 - 2023-09-17
+- Report line and column when using `janet_dobytes`
+- Add `:unless` loop modifier
+- Allow calling `reverse` on generators.
+- Improve performance of a number of core functions including `partition`, `mean`, `keys`, `values`, `pairs`, `interleave`.
+- Add `lengthable?`
+- Add `os/sigaction`
+- Change `every?` and `any?` to behave like the functional versions of the `and` and `or` macros.
+- Fix bug with garbage collecting threaded abstract types.
+- Add `:signal` to the `sandbox` function to allow intercepting signals.
+
 ## 1.30.0 - 2023-08-05
 - Change indexing of `array/remove` to start from -1 at the end instead of -2.
 - Add new string escape sequences `\\a`, `\\b`, `\\?`, and `\\'`.

Makefile

@@ -196,9 +196,9 @@ build/%.bin.o: src/%.c $(JANET_HEADERS) $(JANET_LOCAL_HEADERS) Makefile
 ########################
 
 ifeq ($(UNAME), Darwin)
-SONAME=libjanet.1.30.dylib
+SONAME=libjanet.1.31.dylib
 else
-SONAME=libjanet.so.1.30
+SONAME=libjanet.so.1.31
 endif
 
 build/c/shell.c: src/mainclient/shell.c

build_win.bat

@@ -41,32 +41,32 @@ if not exist build\boot mkdir build\boot
 @rem Build the bootstrap interpreter
 for %%f in (src\core\*.c) do (
     %JANET_COMPILE% /DJANET_BOOTSTRAP /Fobuild\boot\%%~nf.obj %%f
-    @if errorlevel 1 goto :BUILDFAIL
+    @if not errorlevel 0 goto :BUILDFAIL
 )
 for %%f in (src\boot\*.c) do (
     %JANET_COMPILE% /DJANET_BOOTSTRAP /Fobuild\boot\%%~nf.obj %%f
-    @if errorlevel 1 goto :BUILDFAIL
+    @if not errorlevel 0 goto :BUILDFAIL
 )
 %JANET_LINK% /out:build\janet_boot.exe build\boot\*.obj
-@if errorlevel 1 goto :BUILDFAIL
+@if not errorlevel 0 goto :BUILDFAIL
 build\janet_boot . > build\c\janet.c
 
 @rem Build the sources
 %JANET_COMPILE% /Fobuild\janet.obj build\c\janet.c
-@if errorlevel 1 goto :BUILDFAIL
+@if not errorlevel 0 goto :BUILDFAIL
 %JANET_COMPILE% /Fobuild\shell.obj src\mainclient\shell.c
-@if errorlevel 1 goto :BUILDFAIL
+@if not errorlevel 0 goto :BUILDFAIL
 
 @rem Build the resources
 rc /nologo /fobuild\janet_win.res janet_win.rc
 
 @rem Link everything to main client
 %JANET_LINK% /out:janet.exe build\janet.obj build\shell.obj build\janet_win.res
-@if errorlevel 1 goto :BUILDFAIL
+@if not errorlevel 0 goto :BUILDFAIL
 
-@rem Build static library (libjanet.a)
+@rem Build static library (libjanet.lib)
 %JANET_LINK_STATIC% /out:build\libjanet.lib build\janet.obj
-@if errorlevel 1 goto :BUILDFAIL
+@if not errorlevel 0 goto :BUILDFAIL
 
 echo === Successfully built janet.exe for Windows ===
 echo === Run 'build_win test' to run tests. ==
@@ -98,7 +98,7 @@ exit /b 0
 :TEST
 for %%f in (test/suite*.janet) do (
     janet.exe test\%%f
-    @if errorlevel 1 goto TESTFAIL
+    @if not errorlevel 0 goto TESTFAIL
 )
 exit /b 0
 
@@ -117,6 +117,7 @@ copy README.md dist\README.md
 
 copy janet.lib dist\janet.lib
 copy janet.exp dist\janet.exp
+copy janet.def dist\janet.def
 
 janet.exe tools\patch-header.janet src\include\janet.h src\conf\janetconf.h build\janet.h
 copy build\janet.h dist\janet.h

examples/sigaction.janet

@@ -1,10 +1,41 @@
-(defn action []
-  (print "Handled SIGHUP!")
-  (flush))
+###
+### Usage: janet examples/sigaction.janet 1|2|3|4 &
+###
+### Then at shell: kill -s SIGTERM $!
+###
 
-(defn main [_]
-  # Set the interrupt-interpreter argument to `true` to allow
-  # interrupting the busy loop `(forever)`. By default, will not
-  # interrupt the interpreter.
-  (os/sigaction :hup action true)
+(defn action
+  []
+  (print "Handled SIGTERM!")
+  (flush)
+  (os/exit 1))
+
+(defn main1
+  []
+  (os/sigaction :term action true)
   (forever))
+
+(defn main2
+  []
+  (os/sigaction :term action)
+  (forever))
+
+(defn main3
+  []
+  (os/sigaction :term action true)
+  (forever (ev/sleep math/inf)))
+
+(defn main4
+  []
+  (os/sigaction :term action)
+  (forever (ev/sleep math/inf)))
+
+(defn main
+  [& args]
+  (def which (scan-number (get args 1 "1")))
+  (case which
+    1 (main1) # should work
+    2 (main2) # will not work
+    3 (main3) # should work
+    4 (main4) # should work
+    (error "bad main")))

meson.build

@@ -20,7 +20,7 @@
 
 project('janet', 'c',
   default_options : ['c_std=c99', 'build.c_std=c99', 'b_lundef=false', 'default_library=both'],
-  version : '1.30.0')
+  version : '1.31.0')
 
 # Global settings
 janet_path = join_paths(get_option('prefix'), get_option('libdir'), 'janet')
@@ -169,7 +169,7 @@ janet_boot = executable('janet-boot', core_src, boot_src,
 
 # Build janet.c
 janetc = custom_target('janetc',
-  input : [janet_boot],
+  input : [janet_boot, 'src/boot/boot.janet'],
   output : 'janet.c',
   capture : true,
   command : [
@@ -182,23 +182,30 @@ if not get_option('single_threaded')
   janet_dependencies += thread_dep
 endif
 
+if cc.has_argument('-fvisibility=hidden')
+  lib_cflags = ['-fvisibility=hidden']
+else
+  lib_cflags = []
+endif
 libjanet = library('janet', janetc,
   include_directories : incdir,
   dependencies : janet_dependencies,
   version: meson.project_version(),
   soversion: version_parts[0] + '.' + version_parts[1],
+  c_args : lib_cflags,
   install : true)
 
 # Extra c flags - adding -fvisibility=hidden matches the Makefile and
 # shaves off about 10k on linux x64, likely similar on other platforms.
 if cc.has_argument('-fvisibility=hidden')
-  extra_cflags = ['-fvisibility=hidden']
+  extra_cflags = ['-fvisibility=hidden', '-DJANET_DLL_IMPORT']
 else
-  extra_cflags = []
+  extra_cflags = ['-DJANET_DLL_IMPORT']
 endif
-janet_mainclient = executable('janet', janetc, mainclient_src,
+janet_mainclient = executable('janet', mainclient_src,
   include_directories : incdir,
   dependencies : janet_dependencies,
+  link_with: [libjanet],
   c_args : extra_cflags,
   install : true)
 
@@ -281,11 +288,12 @@ patched_janet = custom_target('patched-janeth',
   install : true,
   install_dir : join_paths(get_option('includedir'), 'janet'),
   build_by_default : true,
-  output : ['janet.h'],
+  output : ['janet_' + meson.project_version() + '.h'],
   command : [janet_nativeclient, '@INPUT@', '@OUTPUT@'])
 
 # Create a version of the janet.h header that matches what jpm often expects
 if meson.version().version_compare('>=0.61')
   install_symlink('janet.h', pointing_to: 'janet/janet.h', install_dir: get_option('includedir'))
+  install_symlink('janet.h', pointing_to: 'janet_' + meson.project_version() + '.h', install_dir: join_paths(get_option('includedir'), 'janet'))
 endif
 

meson_options.txt

@@ -18,7 +18,7 @@ option('realpath', type : 'boolean', value : true)
 option('simple_getline', type : 'boolean', value : false)
 option('epoll', type : 'boolean', value : false)
 option('kqueue', type : 'boolean', value : false)
-option('interpreter_interrupt', type : 'boolean', value : false)
+option('interpreter_interrupt', type : 'boolean', value : true)
 option('ffi', type : 'boolean', value : true)
 option('ffi_jit', type : 'boolean', value : true)
 

src/boot/boot.janet

@@ -419,9 +419,11 @@
     (error (string "expected tuple for range, got " x))))
 
 (defn- range-template
-  [binding object rest op comparison]
+  [binding object kind rest op comparison]
   (let [[start stop step] (check-indexed object)]
-    (for-template binding start stop (or step 1) comparison op [rest])))
+    (case kind
+      :range (for-template binding (if stop start 0) (or stop start) (or step 1) comparison op [rest])
+      :down (for-template binding start (or stop 0) (or step 1) comparison op [rest]))))
 
 (defn- each-template
   [binding inx kind body]
@@ -471,16 +473,17 @@
           :repeat (with-syms [iter]
                     ~(do (var ,iter ,verb) (while (> ,iter 0) ,rest (-- ,iter))))
           :when ~(when ,verb ,rest)
+          :unless ~(unless ,verb ,rest)
           (error (string "unexpected loop modifier " binding))))))
 
   # 3 term expression
   (def {(+ i 2) object} head)
   (let [rest (loop1 body head (+ i 3))]
     (case verb
-      :range (range-template binding object rest + <)
-      :range-to (range-template binding object rest + <=)
-      :down (range-template binding object rest - >)
-      :down-to (range-template binding object rest - >=)
+      :range (range-template binding object :range rest + <)
+      :range-to (range-template binding object :range rest + <=)
+      :down (range-template binding object :down rest - >)
+      :down-to (range-template binding object :down rest - >=)
       :keys (each-template binding object :keys [rest])
       :pairs (each-template binding object :pairs [rest])
       :in (each-template binding object :each [rest])
@@ -587,7 +590,10 @@
   * `:repeat n` -- repeats the next inner loop `n` times.
 
   * `:when condition` -- only evaluates the current loop body when `condition`
-    is true.
+    is truthy.
+
+  * `:unless condition` -- only evaluates the current loop body when `condition`
+    is falsey.
 
   The `loop` macro always evaluates to nil.
   ```
@@ -641,7 +647,12 @@
 (defn mean
   "Returns the mean of xs. If empty, returns NaN."
   [xs]
-  (/ (sum xs) (length xs)))
+  (if (lengthable? xs)
+    (/ (sum xs) (length xs))
+    (do
+      (var [accum total] [0 0])
+      (each x xs (+= accum x) (++ total))
+      (/ accum total))))
 
 (defn product
   "Returns the product of xs. If xs is empty, returns 1."
@@ -702,30 +713,38 @@
   [f]
   (fn [x] (not (f x))))
 
+(defmacro- do-extreme
+  [order args]
+  ~(do
+     (def ds ,args)
+     (var k (next ds nil))
+     (var ret (get ds k))
+     (while (,not= nil (set k (next ds k)))
+       (def x (in ds k))
+       (if (,order x ret) (set ret x)))
+     ret))
+
 (defn extreme
   ``Returns the most extreme value in `args` based on the function `order`.
   `order` should take two values and return true or false (a comparison).
   Returns nil if `args` is empty.``
-  [order args]
-  (var [ret] args)
-  (each x args (if (order x ret) (set ret x)))
-  ret)
+  [order args] (do-extreme order args))
 
 (defn max
   "Returns the numeric maximum of the arguments."
-  [& args] (extreme > args))
+  [& args] (do-extreme > args))
 
 (defn min
   "Returns the numeric minimum of the arguments."
-  [& args] (extreme < args))
+  [& args] (do-extreme < args))
 
 (defn max-of
   "Returns the numeric maximum of the argument sequence."
-  [args] (extreme > args))
+  [args] (do-extreme > args))
 
 (defn min-of
   "Returns the numeric minimum of the argument sequence."
-  [args] (extreme < args))
+  [args] (do-extreme < args))
 
 (defn first
   "Get the first element from an indexed data structure."
@@ -956,7 +975,6 @@
        1 (map-n 1 ,maptype ,res ,f ,ind ,inds)
        2 (map-n 2 ,maptype ,res ,f ,ind ,inds)
        3 (map-n 3 ,maptype ,res ,f ,ind ,inds)
-       4 (map-n 4 ,maptype ,res ,f ,ind ,inds)
        (do
          (def iter-keys (array/new-filled ninds))
          (def call-buffer (array/new-filled ninds))
@@ -1425,7 +1443,7 @@
 
 (defn every?
   ``Evaluates to the last element of `ind` if all preceding elements are truthy,
-  otherwise evaluates to the first falsey argument.``
+  otherwise evaluates to the first falsey element.``
   [ind]
   (var res true)
   (loop [x :in ind :while res]
@@ -1445,28 +1463,29 @@
   `Reverses the order of the elements in a given array or buffer and returns it
   mutated.`
   [t]
-  (def len-1 (- (length t) 1))
-  (def half (/ len-1 2))
-  (forv i 0 half
-    (def j (- len-1 i))
-    (def l (in t i))
-    (def r (in t j))
-    (put t i r)
-    (put t j l))
+  (var i 0)
+  (var j (length t))
+  (while (< i (-- j))
+    (def ti (in t i))
+    (put t i (in t j))
+    (put t j ti)
+    (++ i))
   t)
 
 (defn reverse
   `Reverses the order of the elements in a given array or tuple and returns
-  a new array. If a string or buffer is provided, returns an array of its
-  byte values, reversed.`
+  a new array. If a string or buffer is provided, returns a buffer instead.`
   [t]
-  (var n (length t))
-  (def ret (if (bytes? t)
-             (buffer/new-filled n)
-             (array/new-filled n)))
-  (each v t
-    (put ret (-- n) v))
-  ret)
+  (if (lengthable? t)
+    (do
+      (var n (length t))
+      (def ret (if (bytes? t)
+                 (buffer/new-filled n)
+                 (array/new-filled n)))
+      (each v t
+        (put ret (-- n) v))
+      ret)
+    (reverse! (seq [v :in t] v))))
 
 (defn invert
   ``Given an associative data structure `ds`, returns a new table where the
@@ -1576,32 +1595,41 @@
 (defn keys
   "Get the keys of an associative data structure."
   [x]
-  (def arr (array/new-filled (length x)))
-  (var i 0)
-  (eachk k x
-    (put arr i k)
-    (++ i))
-  arr)
+  (if (lengthable? x)
+    (do
+      (def arr (array/new-filled (length x)))
+      (var i 0)
+      (eachk k x
+        (put arr i k)
+        (++ i))
+      arr)
+    (seq [k :keys x] k)))
 
 (defn values
   "Get the values of an associative data structure."
   [x]
-  (def arr (array/new-filled (length x)))
-  (var i 0)
-  (each v x
-    (put arr i v)
-    (++ i))
-  arr)
+  (if (lengthable? x)
+    (do
+      (def arr (array/new-filled (length x)))
+      (var i 0)
+      (each v x
+        (put arr i v)
+        (++ i))
+      arr)
+    (seq [v :in x] v)))
 
 (defn pairs
   "Get the key-value pairs of an associative data structure."
   [x]
-  (def arr (array/new-filled (length x)))
-  (var i 0)
-  (eachp p x
-    (put arr i p)
-    (++ i))
-  arr)
+  (if (lengthable? x)
+    (do
+      (def arr (array/new-filled (length x)))
+      (var i 0)
+      (eachp p x
+        (put arr i p)
+        (++ i))
+      arr)
+    (seq [p :pairs x] p)))
 
 (defn frequencies
   "Get the number of occurrences of each value in an indexed data structure."
@@ -1646,14 +1674,7 @@
 (defn interleave
   "Returns an array of the first elements of each col, then the second elements, etc."
   [& cols]
-  (def res @[])
-  (def ncol (length cols))
-  (when (> ncol 0)
-    (def len (min ;(map length cols)))
-    (loop [i :range [0 len]
-           ci :range [0 ncol]]
-      (array/push res (in (in cols ci) i))))
-  res)
+  (mapcat tuple ;cols))
 
 (defn distinct
   "Returns an array of the deduplicated values in `xs`."
@@ -1700,29 +1721,46 @@
   ``Returns a sequence of the elements of `ind` separated by
   `sep`. Returns a new array.``
   [sep ind]
+  (var k (next ind nil))
+  (if (not= nil k)
+    (if (lengthable? ind)
+      (do
+        (def ret (array/new-filled (- (* 2 (length ind)) 1) sep))
+        (var i 0)
+        (while (not= nil k)
+          (put ret i (in ind k))
+          (set k (next ind k))
+          (+= i 2))
+        ret)
+      (do
+        (def ret @[(in ind k)])
+        (while (not= nil (set k (next ind k)))
+          (array/push ret sep (in ind k)))
+        ret))
+    @[]))
+
+(defn- partition-slice
+  [f n ind]
+  (var [start end] [0 n])
   (def len (length ind))
-  (def ret (array/new (- (* 2 len) 1)))
-  (if (> len 0) (put ret 0 (in ind 0)))
-  (var i 1)
-  (while (< i len)
-    (array/push ret sep (in ind i))
-    (++ i))
+  (def parts (div len n))
+  (def ret (array/new-filled parts))
+  (forv k 0 parts
+    (put ret k (f ind start end))
+    (set start end)
+    (+= end n))
+  (if (< start len)
+    (array/push ret (f ind start)))
   ret)
 
 (defn partition
   ``Partition an indexed data structure `ind` into tuples
   of size `n`. Returns a new array.``
   [n ind]
-  (var i 0) (var nextn n)
-  (def len (length ind))
-  (def ret (array/new (math/ceil (/ len n))))
-  (def slicer (if (bytes? ind) string/slice tuple/slice))
-  (while (<= nextn len)
-    (array/push ret (slicer ind i nextn))
-    (set i nextn)
-    (+= nextn n))
-  (if (not= i len) (array/push ret (slicer ind i)))
-  ret)
+  (cond
+    (indexed? ind) (partition-slice tuple/slice n ind)
+    (bytes? ind) (partition-slice string/slice n ind)
+    (partition-slice tuple/slice n (values ind))))
 
 ###
 ###
@@ -2869,7 +2907,12 @@
     (if (= :dead fs)
       (when is-repl
         (put env '_ @{:value x})
-        (printf (get env *pretty-format* "%q") x)
+        (def pf (get env *pretty-format* "%q"))
+        (try
+          (printf pf x)
+          ([e]
+            (eprintf "bad pretty format %v: %v" pf e)
+            (eflush)))
         (flush))
       (do
         (debug/stacktrace f x "")
@@ -3723,7 +3766,7 @@
     [host port &opt handler type]
     (def s (net/listen host port type))
     (if handler
-      (ev/call (fn [] (net/accept-loop s handler))))
+      (ev/go (fn [] (net/accept-loop s handler))))
     s))
 
 ###

src/conf/janetconf.h

@@ -4,10 +4,10 @@
 #define JANETCONF_H
 
 #define JANET_VERSION_MAJOR 1
-#define JANET_VERSION_MINOR 30
+#define JANET_VERSION_MINOR 31
 #define JANET_VERSION_PATCH 0
 #define JANET_VERSION_EXTRA ""
-#define JANET_VERSION "1.30.0"
+#define JANET_VERSION "1.31.0"
 
 /* #define JANET_BUILD "local" */
 

src/core/array.c

@@ -177,8 +177,8 @@ JANET_CORE_FN(cfun_array_peek,
 }
 
 JANET_CORE_FN(cfun_array_push,
-              "(array/push arr x)",
-              "Insert an element in the end of an array. Modifies the input array and returns it.") {
+              "(array/push arr & xs)",
+              "Push all the elements of xs to the end of an array. Modifies the input array and returns it.") {
     janet_arity(argc, 1, -1);
     JanetArray *array = janet_getarray(argv, 0);
     if (INT32_MAX - argc + 1 <= array->count) {

src/core/corelib.c

@@ -680,6 +680,13 @@ JANET_CORE_FN(janet_core_is_dictionary,
     return janet_wrap_boolean(janet_checktypes(argv[0], JANET_TFLAG_DICTIONARY));
 }
 
+JANET_CORE_FN(janet_core_is_lengthable,
+              "(lengthable? x)",
+              "Check if x is a bytes, indexed, or dictionary.") {
+    janet_fixarity(argc, 1);
+    return janet_wrap_boolean(janet_checktypes(argv[0], JANET_TFLAG_LENGTHABLE));
+}
+
 JANET_CORE_FN(janet_core_signal,
               "(signal what x)",
               "Raise a signal with payload x. ") {
@@ -1079,6 +1086,7 @@ static void janet_load_libs(JanetTable *env) {
         JANET_CORE_REG("bytes?", janet_core_is_bytes),
         JANET_CORE_REG("indexed?", janet_core_is_indexed),
         JANET_CORE_REG("dictionary?", janet_core_is_dictionary),
+        JANET_CORE_REG("lengthable?", janet_core_is_lengthable),
         JANET_CORE_REG("slice", janet_core_slice),
         JANET_CORE_REG("range", janet_core_range),
         JANET_CORE_REG("signal", janet_core_signal),

src/core/ev.c

@@ -127,7 +127,7 @@ static int32_t janet_q_count(JanetQueue *q) {
            : (q->tail - q->head);
 }
 
-static int janet_q_push(JanetQueue *q, void *item, size_t itemsize) {
+static int janet_q_maybe_resize(JanetQueue *q, size_t itemsize) {
     int32_t count = janet_q_count(q);
     /* Resize if needed */
     if (count + 1 >= q->capacity) {
@@ -151,11 +151,27 @@ static int janet_q_push(JanetQueue *q, void *item, size_t itemsize) {
         }
         q->capacity = newcap;
     }
+    return 0;
+}
+
+static int janet_q_push(JanetQueue *q, void *item, size_t itemsize) {
+    if (janet_q_maybe_resize(q, itemsize)) return 1;
     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_push_head(JanetQueue *q, void *item, size_t itemsize) {
+    if (janet_q_maybe_resize(q, itemsize)) return 1;
+    int32_t newhead = q->head - 1;
+    if (newhead < 0) {
+        newhead += q->capacity;
+    }
+    memcpy((char *) q->data + itemsize * newhead, item, itemsize);
+    q->head = newhead;
+    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);
@@ -164,7 +180,7 @@ static int janet_q_pop(JanetQueue *q, void *out, size_t itemsize) {
 }
 
 /* Forward declaration */
-static void janet_unlisten(JanetListenerState *state, int is_gc);
+static void janet_unlisten(JanetListenerState *state);
 
 /* Get current timestamp (millisecond precision) */
 static JanetTimestamp ts_now(void);
@@ -238,76 +254,100 @@ static void add_timeout(JanetTimeout to) {
     }
 }
 
+static int janet_listener_gc(void *p, size_t s);
+static int janet_listener_mark(void *p, size_t s);
+
+static const JanetAbstractType janet_listener_AT = {
+    "core/ev-listener",
+    janet_listener_gc,
+    janet_listener_mark,
+    JANET_ATEND_GCMARK
+};
+
 /* Create a new event listener */
 static JanetListenerState *janet_listen_impl(JanetStream *stream, JanetListener behavior, int mask, size_t size, void *user) {
     if (stream->flags & JANET_STREAM_CLOSED) {
         janet_panic("cannot listen on closed stream");
     }
-    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 = janet_malloc(size);
-    if (NULL == state) {
-        JANET_OUT_OF_MEMORY;
-    }
+    if ((mask & JANET_ASYNC_LISTEN_READ) && stream->read_state) goto bad_listen_read;
+    if ((mask & JANET_ASYNC_LISTEN_WRITE) && stream->write_state) goto bad_listen_write;
+    janet_assert(size >= sizeof(JanetListenerState), "bad size");
+    JanetListenerState *state = janet_abstract(&janet_listener_AT, size);
     state->machine = behavior;
     state->fiber = janet_vm.root_fiber;
+    state->flags = 0;
     janet_vm.root_fiber->waiting = state;
+    if (mask & JANET_ASYNC_LISTEN_READ) stream->read_state = state;
+    if (mask & JANET_ASYNC_LISTEN_WRITE) stream->write_state = state;
     state->stream = stream;
-    state->_mask = mask;
-    stream->_mask |= mask;
-    state->_next = stream->state;
-    stream->state = state;
-
-    /* Keep track of a listener for GC purposes */
-    int resize = janet_vm.listener_cap == janet_vm.listener_count;
-    if (resize) {
-        size_t newcap = janet_vm.listener_count ? janet_vm.listener_cap * 2 : 16;
-        janet_vm.listeners = janet_realloc(janet_vm.listeners, newcap * sizeof(JanetListenerState *));
-        if (NULL == janet_vm.listeners) {
-            JANET_OUT_OF_MEMORY;
-        }
-        janet_vm.listener_cap = newcap;
-    }
-    size_t index = janet_vm.listener_count++;
-    janet_vm.listeners[index] = state;
-    state->_index = index;
-
-    /* Emit INIT event for convenience */
     state->event = user;
     state->machine(state, JANET_ASYNC_EVENT_INIT);
+    janet_ev_inc_refcount();
+    state->index = janet_vm.listeners->count;
+    janet_array_push(janet_vm.listeners, janet_wrap_abstract(state));
     return state;
+bad_listen_write:
+    janet_panic("cannot listen for duplicate write event on stream");
+bad_listen_read:
+    janet_panic("cannot listen for duplicate read event on stream");
 }
 
-/* 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, int is_gc) {
-    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;
-    /* Remove mask */
-    state->stream->_mask &= ~(state->_mask);
-    /* Ensure fiber does not reference this state */
-    if (!is_gc) {
-        JanetFiber *fiber = state->fiber;
-        if (NULL != fiber && fiber->waiting == state) {
-            fiber->waiting = NULL;
-        }
+void janet_fiber_did_resume(JanetFiber *fiber) {
+    if (fiber->waiting) {
+        janet_unlisten(fiber->waiting);
+        fiber->waiting = NULL;
+    }
+}
+
+static void janet_unlisten_impl(JanetListenerState *state) {
+    /* Move last listener to position of this listener - O(1) removal and keep things densely packed. */
+    if (state->stream) {
+        Janet popped = janet_array_pop(janet_vm.listeners);
+        janet_assert(janet_checktype(popped, JANET_ABSTRACT), "pop check");
+        JanetListenerState *popped_state = (JanetListenerState *) janet_unwrap_abstract(popped);
+        janet_vm.listeners->data[state->index] = popped;
+        popped_state->index = state->index;
+        state->index = UINT32_MAX; /* just in case */
+        janet_ev_dec_refcount();
+        if (state->stream->read_state == state) {
+            state->stream->read_state = NULL;
+        }
+        if (state->stream->write_state == state) {
+            state->stream->write_state = NULL;
+        }
+        state->stream = NULL;
+    }
+}
+
+static int janet_listener_gc(void *p, size_t size) {
+    (void) size;
+    JanetListenerState *state = (JanetListenerState *)p;
+    if (state->stream) {
+        janet_ev_dec_refcount();
+    }
+    if (state->machine) {
+        state->machine(state, JANET_ASYNC_EVENT_DEINIT);
+    }
+    return 0;
+}
+
+static int janet_listener_mark(void *p, size_t size) {
+    (void) size;
+    JanetListenerState *state = (JanetListenerState *)p;
+    if (state->stream) {
+        janet_mark(janet_wrap_abstract(state->stream));
+    }
+    if (state->fiber) {
+        janet_mark(janet_wrap_fiber(state->fiber));
+    }
+    state->machine(state, JANET_ASYNC_EVENT_MARK);
+    return 0;
+}
+
+static void janet_stream_checktoclose(JanetStream *stream) {
+    if ((stream->flags & JANET_STREAM_TOCLOSE) && !stream->read_state && !stream->write_state) {
+        janet_stream_close(stream);
     }
-    /* Untrack a listener for gc purposes */
-    size_t index = state->_index;
-    janet_vm.listeners[index] = janet_vm.listeners[--janet_vm.listener_count];
-    janet_vm.listeners[index]->_index = index;
-    janet_free(state);
 }
 
 static const JanetMethod ev_default_stream_methods[] = {
@@ -323,52 +363,52 @@ JanetStream *janet_stream(JanetHandle handle, uint32_t flags, const JanetMethod
     JanetStream *stream = janet_abstract(&janet_stream_type, sizeof(JanetStream));
     stream->handle = handle;
     stream->flags = flags;
-    stream->state = NULL;
-    stream->_mask = 0;
+    stream->read_state = NULL;
+    stream->write_state = NULL;
     if (methods == NULL) methods = ev_default_stream_methods;
     stream->methods = methods;
     return stream;
 }
 
-/* Close a stream */
-static void janet_stream_close_impl(JanetStream *stream, int is_gc) {
-    if (stream->flags & JANET_STREAM_CLOSED) return;
-    JanetListenerState *state = stream->state;
-    while (NULL != state) {
-        if (!is_gc) {
-            state->machine(state, JANET_ASYNC_EVENT_CLOSE);
-        }
-        JanetListenerState *next_state = state->_next;
-        janet_unlisten(state, is_gc);
-        state = next_state;
-    }
-    stream->state = NULL;
+static void janet_stream_close_impl(JanetStream *stream) {
     stream->flags |= JANET_STREAM_CLOSED;
 #ifdef JANET_WINDOWS
+    if (stream->handle != INVALID_HANDLE_VALUE) {
 #ifdef JANET_NET
-    if (stream->flags & JANET_STREAM_SOCKET) {
-        closesocket((SOCKET) stream->handle);
-    } else
+        if (stream->flags & JANET_STREAM_SOCKET) {
+            closesocket((SOCKET) stream->handle);
+        } else
 #endif
-    {
-        CloseHandle(stream->handle);
+        {
+            CloseHandle(stream->handle);
+        }
+        stream->handle = INVALID_HANDLE_VALUE;
     }
-    stream->handle = INVALID_HANDLE_VALUE;
 #else
-    close(stream->handle);
-    stream->handle = -1;
+    if (stream->handle != -1) {
+        close(stream->handle);
+        stream->handle = -1;
+    }
 #endif
 }
 
 void janet_stream_close(JanetStream *stream) {
-    janet_stream_close_impl(stream, 0);
+    if (stream->read_state) {
+        stream->read_state->machine(stream->read_state, JANET_ASYNC_EVENT_CLOSE);
+        janet_unlisten(stream->read_state);
+    }
+    if (stream->write_state) {
+        stream->write_state->machine(stream->write_state, JANET_ASYNC_EVENT_CLOSE);
+        janet_unlisten(stream->write_state);
+    }
+    janet_stream_close_impl(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_impl(stream, 1);
+    janet_stream_close_impl(stream);
     return 0;
 }
 
@@ -376,13 +416,11 @@ static int janet_stream_gc(void *p, size_t s) {
 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;
+    if (NULL != stream->read_state) {
+        janet_mark(janet_wrap_abstract(stream->read_state));
+    }
+    if (NULL != stream->write_state) {
+        janet_mark(janet_wrap_abstract(stream->write_state));
     }
     return 0;
 }
@@ -435,8 +473,8 @@ static void *janet_stream_unmarshal(JanetMarshalContext *ctx) {
     }
     JanetStream *p = janet_unmarshal_abstract(ctx, sizeof(JanetStream));
     /* Can't share listening state and such across threads */
-    p->_mask = 0;
-    p->state = NULL;
+    p->read_state = NULL;
+    p->write_state = NULL;
     p->flags = (uint32_t) janet_unmarshal_int(ctx);
     p->methods =  janet_unmarshal_ptr(ctx);
 #ifdef JANET_WINDOWS
@@ -468,7 +506,7 @@ const JanetAbstractType janet_stream_type = {
 };
 
 /* Register a fiber to resume with value */
-void janet_schedule_signal(JanetFiber *fiber, Janet value, JanetSignal sig) {
+static void janet_schedule_general(JanetFiber *fiber, Janet value, JanetSignal sig, int soon) {
     if (fiber->gc.flags & JANET_FIBER_EV_FLAG_CANCELED) return;
     if (!(fiber->gc.flags & JANET_FIBER_FLAG_ROOT)) {
         Janet task_element = janet_wrap_fiber(fiber);
@@ -477,7 +515,19 @@ void janet_schedule_signal(JanetFiber *fiber, Janet value, JanetSignal sig) {
     JanetTask t = { fiber, value, sig, ++fiber->sched_id };
     fiber->gc.flags |= JANET_FIBER_FLAG_ROOT;
     if (sig == JANET_SIGNAL_ERROR) fiber->gc.flags |= JANET_FIBER_EV_FLAG_CANCELED;
-    janet_q_push(&janet_vm.spawn, &t, sizeof(t));
+    if (soon) {
+        janet_q_push_head(&janet_vm.spawn, &t, sizeof(t));
+    } else {
+        janet_q_push(&janet_vm.spawn, &t, sizeof(t));
+    }
+}
+
+void janet_schedule_signal(JanetFiber *fiber, Janet value, JanetSignal sig) {
+    janet_schedule_general(fiber, value, sig, 0);
+}
+
+void janet_schedule_soon(JanetFiber *fiber, Janet value, JanetSignal sig) {
+    janet_schedule_general(fiber, value, sig, 1);
 }
 
 void janet_cancel(JanetFiber *fiber, Janet value) {
@@ -488,14 +538,6 @@ 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) {
-        fiber->waiting->machine(fiber->waiting, JANET_ASYNC_EVENT_CANCEL);
-        janet_unlisten(fiber->waiting, 0);
-    }
-}
-
 /* Mark all pending tasks */
 void janet_ev_mark(void) {
 
@@ -524,16 +566,6 @@ void janet_ev_mark(void) {
             janet_mark(janet_wrap_fiber(janet_vm.tq[i].curr_fiber));
         }
     }
-
-    /* Pending listeners */
-    for (size_t i = 0; i < janet_vm.listener_count; i++) {
-        JanetListenerState *state = janet_vm.listeners[i];
-        if (NULL != state->fiber) {
-            janet_mark(janet_wrap_fiber(state->fiber));
-        }
-        janet_stream_mark(state->stream, sizeof(JanetStream));
-        (state->machine)(state, JANET_ASYNC_EVENT_MARK);
-    }
 }
 
 static int janet_channel_push(JanetChannel *channel, Janet x, int mode);
@@ -554,9 +586,6 @@ static Janet make_supervisor_event(const char *name, JanetFiber *fiber, int thre
 /* Common init code */
 void janet_ev_init_common(void) {
     janet_q_init(&janet_vm.spawn);
-    janet_vm.listener_count = 0;
-    janet_vm.listener_cap = 0;
-    janet_vm.listeners = NULL;
     janet_vm.tq = NULL;
     janet_vm.tq_count = 0;
     janet_vm.tq_capacity = 0;
@@ -564,6 +593,8 @@ void janet_ev_init_common(void) {
     janet_table_init_raw(&janet_vm.active_tasks, 0);
     janet_table_init_raw(&janet_vm.signal_handlers, 0);
     janet_rng_seed(&janet_vm.ev_rng, 0);
+    janet_vm.listeners = janet_array(0);
+    janet_gcroot(janet_wrap_array(janet_vm.listeners));
 #ifndef JANET_WINDOWS
     pthread_attr_init(&janet_vm.new_thread_attr);
     pthread_attr_setdetachstate(&janet_vm.new_thread_attr, PTHREAD_CREATE_DETACHED);
@@ -574,11 +605,10 @@ void janet_ev_init_common(void) {
 void janet_ev_deinit_common(void) {
     janet_q_deinit(&janet_vm.spawn);
     janet_free(janet_vm.tq);
-    janet_free(janet_vm.listeners);
-    janet_vm.listeners = NULL;
     janet_table_deinit(&janet_vm.threaded_abstracts);
     janet_table_deinit(&janet_vm.active_tasks);
     janet_table_deinit(&janet_vm.signal_handlers);
+    janet_vm.listeners = NULL;
 #ifndef JANET_WINDOWS
     pthread_attr_destroy(&janet_vm.new_thread_attr);
 #endif
@@ -604,7 +634,11 @@ void janet_addtimeout(double sec) {
 
 void janet_ev_inc_refcount(void) {
 #ifdef JANET_WINDOWS
-    InterlockedIncrement(&janet_vm.extra_listeners);
+#ifdef JANET_64
+    InterlockedIncrement64((int64_t volatile *) &janet_vm.extra_listeners);
+#else
+    InterlockedIncrement((int32_t volatile *) &janet_vm.extra_listeners);
+#endif
 #else
     __atomic_add_fetch(&janet_vm.extra_listeners, 1, __ATOMIC_RELAXED);
 #endif
@@ -612,7 +646,11 @@ void janet_ev_inc_refcount(void) {
 
 void janet_ev_dec_refcount(void) {
 #ifdef JANET_WINDOWS
-    InterlockedDecrement(&janet_vm.extra_listeners);
+#ifdef JANET_64
+    InterlockedDecrement64((int64_t volatile *) &janet_vm.extra_listeners);
+#else
+    InterlockedDecrement((int32_t volatile *) &janet_vm.extra_listeners);
+#endif
 #else
     __atomic_add_fetch(&janet_vm.extra_listeners, -1, __ATOMIC_RELAXED);
 #endif
@@ -1294,39 +1332,11 @@ const JanetAbstractType janet_channel_type = {
 void janet_loop1_impl(int has_timeout, JanetTimestamp timeout);
 
 int janet_loop_done(void) {
-    return !(janet_vm.listener_count ||
-             (janet_vm.spawn.head != janet_vm.spawn.tail) ||
+    return !((janet_vm.spawn.head != janet_vm.spawn.tail) ||
              janet_vm.tq_count ||
              janet_vm.extra_listeners);
 }
 
-static void janet_loop1_poll(void) {
-    /* Poll for events */
-    if (janet_vm.listener_count || janet_vm.tq_count || janet_vm.extra_listeners) {
-        JanetTimeout to;
-        memset(&to, 0, sizeof(to));
-        int has_timeout;
-        /* Drop timeouts that are no longer needed */
-        while ((has_timeout = peek_timeout(&to))) {
-            if (to.curr_fiber != NULL) {
-                if (!janet_fiber_can_resume(to.curr_fiber)) {
-                    janet_table_remove(&janet_vm.active_tasks, janet_wrap_fiber(to.curr_fiber));
-                    pop_timeout(0);
-                    continue;
-                }
-            } else if (to.fiber->sched_id != to.sched_id) {
-                pop_timeout(0);
-                continue;
-            }
-            break;
-        }
-        /* Run polling implementation only if pending timeouts or pending events */
-        if (janet_vm.tq_count || janet_vm.listener_count || janet_vm.extra_listeners) {
-            janet_loop1_impl(has_timeout, to.when);
-        }
-    }
-}
-
 JanetFiber *janet_loop1(void) {
     /* Schedule expired timers */
     JanetTimeout to;
@@ -1349,8 +1359,10 @@ JanetFiber *janet_loop1(void) {
         }
     }
 
-    /* Run scheduled fibers */
+    /* Run scheduled fibers unless interrupts need to be handled. */
     while (janet_vm.spawn.head != janet_vm.spawn.tail) {
+        /* Don't run until all interrupts have been marked as handled by calling janet_interpreter_interrupt_handled */
+        if (janet_vm.auto_suspend) break;
         JanetTask task = {NULL, janet_wrap_nil(), JANET_SIGNAL_OK, 0};
         janet_q_pop(&janet_vm.spawn, &task, sizeof(task));
         if (task.fiber->gc.flags & JANET_FIBER_EV_FLAG_SUSPENDED) janet_ev_dec_refcount();
@@ -1379,12 +1391,34 @@ JanetFiber *janet_loop1(void) {
             janet_stacktrace_ext(task.fiber, res, "");
         }
         if (sig == JANET_SIGNAL_INTERRUPT) {
-            /* On interrupts, return the interrupted fiber immediately */
             return task.fiber;
         }
     }
 
-    janet_loop1_poll();
+    /* Poll for events */
+    if (janet_vm.tq_count || janet_vm.extra_listeners) {
+        JanetTimeout to;
+        memset(&to, 0, sizeof(to));
+        int has_timeout;
+        /* Drop timeouts that are no longer needed */
+        while ((has_timeout = peek_timeout(&to))) {
+            if (to.curr_fiber != NULL) {
+                if (!janet_fiber_can_resume(to.curr_fiber)) {
+                    janet_table_remove(&janet_vm.active_tasks, janet_wrap_fiber(to.curr_fiber));
+                    pop_timeout(0);
+                    continue;
+                }
+            } else if (to.fiber->sched_id != to.sched_id) {
+                pop_timeout(0);
+                continue;
+            }
+            break;
+        }
+        /* Run polling implementation only if pending timeouts or pending events */
+        if (janet_vm.tq_count || janet_vm.extra_listeners) {
+            janet_loop1_impl(has_timeout, to.when);
+        }
+    }
 
     /* No fiber was interrupted */
     return NULL;
@@ -1405,12 +1439,6 @@ void janet_loop(void) {
     while (!janet_loop_done()) {
         JanetFiber *interrupted_fiber = janet_loop1();
         if (NULL != interrupted_fiber) {
-            /* Allow an extra poll before rescheduling to allow posted events to be handled
-             * before entering a possibly infinite, blocking loop. */
-            Janet x = janet_wrap_fiber(interrupted_fiber);
-            janet_gcroot(x);
-            janet_loop1_poll();
-            janet_gcunroot(x);
             janet_schedule(interrupted_fiber, janet_wrap_nil());
         }
     }
@@ -1479,8 +1507,8 @@ JanetListenerState *janet_listen(JanetStream *stream, JanetListener behavior, in
     return state;
 }
 
-static void janet_unlisten(JanetListenerState *state, int is_gc) {
-    janet_unlisten_impl(state, is_gc);
+static void janet_unlisten(JanetListenerState *state) {
+    janet_unlisten_impl(state);
 }
 
 void janet_loop1_impl(int has_timeout, JanetTimestamp to) {
@@ -1502,7 +1530,11 @@ void janet_loop1_impl(int has_timeout, JanetTimestamp to) {
     }
     BOOL result = GetQueuedCompletionStatus(janet_vm.iocp, &num_bytes_transfered, &completionKey, &overlapped, (DWORD) waittime);
 
-    if (result || overlapped) {
+    if (!result) {
+        return;
+    }
+
+    if (overlapped) {
         if (0 == completionKey) {
             /* Custom event */
             JanetSelfPipeEvent *response = (JanetSelfPipeEvent *)(overlapped);
@@ -1513,24 +1545,22 @@ void janet_loop1_impl(int has_timeout, JanetTimestamp to) {
         } 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, 0);
-                    }
-                    break;
-                } else {
-                    state = state->_next;
+            janet_assert(stream->handle != INVALID_HANDLE_VALUE, "got closed stream event");
+            JanetListenerState *state = NULL;
+            if (stream->read_state && stream->read_state->tag == overlapped) {
+                state = stream->read_state;
+            } else if (stream->write_state && stream->write_state->tag == overlapped) {
+                state = stream->write_state;
+            }
+            if (state != NULL) {
+                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);
                 }
             }
-            /* Close the stream if requested and no more listeners are left */
-            if ((stream->flags & JANET_STREAM_TOCLOSE) && !stream->state) {
-                janet_stream_close(stream);
-            }
+            janet_stream_checktoclose(stream);
         }
     }
 }
@@ -1545,26 +1575,17 @@ static JanetTimestamp ts_now(void) {
     return res;
 }
 
-static int make_epoll_events(int mask) {
-    int events = 0;
-    if (mask & JANET_ASYNC_LISTEN_READ)
-        events |= EPOLLIN;
-    if (mask & JANET_ASYNC_LISTEN_WRITE)
-        events |= EPOLLOUT;
-    return events;
-}
-
 static void janet_epoll_sync_callback(JanetEVGenericMessage msg) {
     JanetListenerState *state = msg.argp;
     JanetAsyncStatus status1 = JANET_ASYNC_STATUS_NOT_DONE;
     JanetAsyncStatus status2 = JANET_ASYNC_STATUS_NOT_DONE;
-    if (state->stream->_mask & JANET_ASYNC_LISTEN_WRITE)
-        status1 = state->machine(state, JANET_ASYNC_EVENT_WRITE);
-    if (state->stream->_mask & JANET_ASYNC_LISTEN_READ)
-        status2 = state->machine(state, JANET_ASYNC_EVENT_READ);
+    if (state == state->stream->read_state)
+        status1 = state->machine(state, JANET_ASYNC_EVENT_READ);
+    if (state == state->stream->write_state)
+        status2 = state->machine(state, JANET_ASYNC_EVENT_WRITE);
     if (status1 == JANET_ASYNC_STATUS_DONE ||
             status2 == JANET_ASYNC_STATUS_DONE) {
-        janet_unlisten(state, 0);
+        janet_unlisten(state);
     } else {
         /* Repost event */
         janet_ev_post_event(NULL, janet_epoll_sync_callback, msg);
@@ -1573,11 +1594,13 @@ static void janet_epoll_sync_callback(JanetEVGenericMessage msg) {
 
 /* 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 is_first = !stream->read_state && !stream->write_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.events = 0;
+    if (stream->read_state) ev.events |= EPOLLIN;
+    if (stream->write_state) ev.events |= EPOLLOUT;
     ev.data.ptr = stream;
     int status;
     do {
@@ -1591,13 +1614,13 @@ JanetListenerState *janet_listen(JanetStream *stream, JanetListener behavior, in
              * event to a file. So we just post a custom event to do the read/write
              * asap. */
             /* Use flag to indicate state is not registered in epoll */
-            state->_mask |= (1 << JANET_ASYNC_EVENT_COMPLETE);
+            state->flags = 1;
             JanetEVGenericMessage msg = {0};
             msg.argp = state;
             janet_ev_post_event(NULL, janet_epoll_sync_callback, msg);
         } else {
             /* Unexpected error */
-            janet_unlisten_impl(state, 0);
+            janet_unlisten_impl(state);
             janet_panicv(janet_ev_lasterr());
         }
     }
@@ -1605,15 +1628,19 @@ JanetListenerState *janet_listen(JanetStream *stream, JanetListener behavior, in
 }
 
 /* Tell system we are done listening for a certain event */
-static void janet_unlisten(JanetListenerState *state, int is_gc) {
+static void janet_unlisten(JanetListenerState *state) {
     JanetStream *stream = state->stream;
-    if (!(stream->flags & JANET_STREAM_CLOSED)) {
+    if (stream && (stream->handle != -1)) {
         /* Use flag to indicate state is not registered in epoll */
-        if (!(state->_mask & (1 << JANET_ASYNC_EVENT_COMPLETE))) {
-            int is_last = (state->_next == NULL && stream->state == state);
+        if (!state->flags) {
+            int is_read = (stream->read_state != state) && stream->read_state;
+            int is_write = (stream->write_state != state) && stream->write_state;
+            int is_last = !is_read && !is_write;
             int op = is_last ? EPOLL_CTL_DEL : EPOLL_CTL_MOD;
             struct epoll_event ev;
-            ev.events = make_epoll_events(stream->_mask & ~state->_mask);
+            ev.events = 0;
+            if (is_read) ev.events |= EPOLLIN;
+            if (is_write) ev.events |= EPOLLOUT;
             ev.data.ptr = stream;
             int status;
             do {
@@ -1625,7 +1652,7 @@ static void janet_unlisten(JanetListenerState *state, int is_gc) {
         }
     }
     /* Destroy state machine and free memory */
-    janet_unlisten_impl(state, is_gc);
+    janet_unlisten_impl(state);
 }
 
 #define JANET_EPOLL_MAX_EVENTS 64
@@ -1662,10 +1689,11 @@ void janet_loop1_impl(int has_timeout, JanetTimestamp timeout) {
         } else {
             JanetStream *stream = p;
             int mask = events[i].events;
-            JanetListenerState *state = stream->state;
-            while (NULL != state) {
+            JanetListenerState *states[2] = {stream->read_state, stream->write_state};
+            for (int j = 0; j < 2; j++) {
+                JanetListenerState *state = states[j];
+                if (!state) continue;
                 state->event = events + i;
-                JanetListenerState *next_state = state->_next;
                 JanetAsyncStatus status1 = JANET_ASYNC_STATUS_NOT_DONE;
                 JanetAsyncStatus status2 = JANET_ASYNC_STATUS_NOT_DONE;
                 JanetAsyncStatus status3 = JANET_ASYNC_STATUS_NOT_DONE;
@@ -1681,14 +1709,11 @@ void janet_loop1_impl(int has_timeout, JanetTimestamp timeout) {
                 if (status1 == JANET_ASYNC_STATUS_DONE ||
                         status2 == JANET_ASYNC_STATUS_DONE ||
                         status3 == JANET_ASYNC_STATUS_DONE ||
-                        status4 == JANET_ASYNC_STATUS_DONE)
-                    janet_unlisten(state, 0);
-                state = next_state;
-            }
-            /* Close the stream if requested and no more listeners are left */
-            if ((stream->flags & JANET_STREAM_TOCLOSE) && !stream->state) {
-                janet_stream_close(stream);
+                        status4 == JANET_ASYNC_STATUS_DONE) {
+                    janet_unlisten(state);
+                }
             }
+            janet_stream_checktoclose(stream);
         }
     }
 }
@@ -1782,46 +1807,44 @@ JanetListenerState *janet_listen(JanetStream *stream, JanetListener behavior, in
     struct kevent kev[2];
 
     int length = 0;
-    if (state->stream->_mask & JANET_ASYNC_LISTEN_READ) {
+    if (mask & JANET_ASYNC_LISTEN_READ) {
         EV_SETx(&kev[length], stream->handle, EVFILT_READ, EV_ADD | EV_ENABLE, 0, 0, stream);
         length++;
     }
-    if (state->stream->_mask & JANET_ASYNC_LISTEN_WRITE) {
+    if (mask & JANET_ASYNC_LISTEN_WRITE) {
         EV_SETx(&kev[length], stream->handle, EVFILT_WRITE, EV_ADD | EV_ENABLE, 0, 0, stream);
         length++;
     }
 
-    if (length > 0) {
-        add_kqueue_events(kev, length);
-    }
+    janet_assert(length, "expected to add kqueue events");
+    add_kqueue_events(kev, length);
 
     return state;
 }
 
-static void janet_unlisten(JanetListenerState *state, int is_gc) {
+static void janet_unlisten(JanetListenerState *state) {
     JanetStream *stream = state->stream;
-    if (!(stream->flags & JANET_STREAM_CLOSED)) {
-        /* Use flag to indicate state is not registered in kqueue */
-        if (!(state->_mask & (1 << JANET_ASYNC_EVENT_COMPLETE))) {
-            int is_last = (state->_next == NULL && stream->state == state);
-            int op = is_last ? EV_DELETE : EV_DISABLE | EV_ADD;
-            struct kevent kev[2];
-            EV_SETx(&kev[1], stream->handle, EVFILT_WRITE, op, 0, 0, stream);
+    if (stream && (stream->handle != -1)) {
+        int is_read = (stream->read_state != state) && stream->read_state;
+        int is_write = (stream->write_state != state) && stream->write_state;
+        int is_last = !is_read && !is_write;
+        int op = is_last ? EV_DELETE : EV_DISABLE | EV_ADD;
+        struct kevent kev[2];
+        EV_SETx(&kev[1], stream->handle, EVFILT_WRITE, op, 0, 0, stream);
 
-            int length = 0;
-            if (stream->_mask & JANET_ASYNC_EVENT_WRITE) {
-                EV_SETx(&kev[length], stream->handle, EVFILT_WRITE, op, 0, 0, stream);
-                length++;
-            }
-            if (stream->_mask & JANET_ASYNC_EVENT_READ) {
-                EV_SETx(&kev[length], stream->handle, EVFILT_READ, op, 0, 0, stream);
-                length++;
-            }
-
-            add_kqueue_events(kev, length);
+        int length = 0;
+        if (stream->read_state == state) {
+            EV_SETx(&kev[length], stream->handle, EVFILT_WRITE, op, 0, 0, stream);
+            length++;
         }
+        if (stream->write_state == state) {
+            EV_SETx(&kev[length], stream->handle, EVFILT_READ, op, 0, 0, stream);
+            length++;
+        }
+
+        add_kqueue_events(kev, length);
     }
-    janet_unlisten_impl(state, is_gc);
+    janet_unlisten_impl(state);
 }
 
 #define JANET_KQUEUE_MAX_EVENTS 64
@@ -1861,14 +1884,14 @@ void janet_loop1_impl(int has_timeout, JanetTimestamp timeout) {
             janet_ev_handle_selfpipe();
         } else {
             JanetStream *stream = p;
-            JanetListenerState *state = stream->state;
-            while (NULL != state) {
-                JanetListenerState *next_state = state->_next;
+            JanetListenerState *states[2] = {stream->read_state, stream->write_state};
+            for (int j = 0; j < 2; j++) {
+                JanetListenerState *state = states[j];
+                if (!state) continue;
                 state->event = events + i;
                 JanetAsyncStatus statuses[4];
                 for (int i = 0; i < 4; i++)
                     statuses[i] = JANET_ASYNC_STATUS_NOT_DONE;
-
                 if (!(events[i].flags & EV_ERROR)) {
                     if (events[i].filter == EVFILT_WRITE)
                         statuses[0] = state->machine(state, JANET_ASYNC_EVENT_WRITE);
@@ -1882,15 +1905,11 @@ void janet_loop1_impl(int has_timeout, JanetTimestamp timeout) {
                 if (statuses[0] == JANET_ASYNC_STATUS_DONE ||
                         statuses[1] == JANET_ASYNC_STATUS_DONE ||
                         statuses[2] == JANET_ASYNC_STATUS_DONE ||
-                        statuses[3] == JANET_ASYNC_STATUS_DONE)
-                    janet_unlisten(state, 0);
-
-                state = next_state;
-            }
-            /* Close the stream if requested and no more listeners are left */
-            if ((stream->flags & JANET_STREAM_TOCLOSE) && !stream->state) {
-                janet_stream_close(stream);
+                        statuses[3] == JANET_ASYNC_STATUS_DONE) {
+                    janet_unlisten(state);
+                }
             }
+            janet_stream_checktoclose(stream);
         }
     }
 }
@@ -1928,20 +1947,11 @@ static JanetTimestamp ts_now(void) {
     return res;
 }
 
-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;
-}
-
 /* Wait for the next event */
 JanetListenerState *janet_listen(JanetStream *stream, JanetListener behavior, int mask, size_t size, void *user) {
-    size_t oldsize = janet_vm.listener_cap;
+    size_t oldsize = janet_vm.listeners->capacity;
     JanetListenerState *state = janet_listen_impl(stream, behavior, mask, size, user);
-    size_t newsize = janet_vm.listener_cap;
+    size_t newsize = janet_vm.listeners->capacity;
     if (newsize > oldsize) {
         janet_vm.fds = janet_realloc(janet_vm.fds, (newsize + 1) * sizeof(struct pollfd));
         if (NULL == janet_vm.fds) {
@@ -1950,15 +1960,19 @@ JanetListenerState *janet_listen(JanetStream *stream, JanetListener behavior, in
     }
     struct pollfd ev;
     ev.fd = stream->handle;
-    ev.events = make_poll_events(state->stream->_mask);
+    ev.events = 0;
+    if (stream->read_state) ev.events |= POLLIN;
+    if (stream->write_state) ev.events |= POLLOUT;
     ev.revents = 0;
-    janet_vm.fds[state->_index + 1] = ev;
+    janet_vm.fds[state->index + 1] = ev;
     return state;
 }
 
-static void janet_unlisten(JanetListenerState *state, int is_gc) {
-    janet_vm.fds[state->_index + 1] = janet_vm.fds[janet_vm.listener_count];
-    janet_unlisten_impl(state, is_gc);
+static void janet_unlisten(JanetListenerState *state) {
+    if (state->stream) {
+        janet_vm.fds[state->index + 1] = janet_vm.fds[janet_vm.listeners->count];
+    }
+    janet_unlisten_impl(state);
 }
 
 void janet_loop1_impl(int has_timeout, JanetTimestamp timeout) {
@@ -1970,7 +1984,7 @@ void janet_loop1_impl(int has_timeout, JanetTimestamp timeout) {
             JanetTimestamp now = ts_now();
             to = now > timeout ? 0 : (int)(timeout - now);
         }
-        ready = poll(janet_vm.fds, janet_vm.listener_count + 1, to);
+        ready = poll(janet_vm.fds, janet_vm.listeners->count + 1, to);
     } while (ready == -1 && errno == EINTR);
     if (ready == -1) {
         JANET_EXIT("failed to poll events");
@@ -1983,10 +1997,10 @@ void janet_loop1_impl(int has_timeout, JanetTimestamp timeout) {
     }
 
     /* Step state machines */
-    for (size_t i = 0; i < janet_vm.listener_count; i++) {
+    for (int32_t i = 0; i < janet_vm.listeners->count; i++) {
         struct pollfd *pfd = janet_vm.fds + i + 1;
         /* Skip fds where nothing interesting happened */
-        JanetListenerState *state = janet_vm.listeners[i];
+        JanetListenerState *state = (JanetListenerState *) janet_unwrap_abstract(janet_vm.listeners->data[i]);
         /* Normal event */
         int mask = pfd->revents;
         JanetAsyncStatus status1 = JANET_ASYNC_STATUS_NOT_DONE;
@@ -2006,12 +2020,10 @@ void janet_loop1_impl(int has_timeout, JanetTimestamp timeout) {
         if (status1 == JANET_ASYNC_STATUS_DONE ||
                 status2 == JANET_ASYNC_STATUS_DONE ||
                 status3 == JANET_ASYNC_STATUS_DONE ||
-                status4 == JANET_ASYNC_STATUS_DONE)
-            janet_unlisten(state, 0);
-        /* Close the stream if requested and no more listeners are left */
-        if ((stream->flags & JANET_STREAM_TOCLOSE) && !stream->state) {
-            janet_stream_close(stream);
+                status4 == JANET_ASYNC_STATUS_DONE) {
+            janet_unlisten(state);
         }
+        janet_stream_checktoclose(stream);
     }
 }
 
@@ -2498,8 +2510,7 @@ void janet_ev_recvfrom(JanetStream *stream, JanetBuffer *buf, int32_t nbytes, in
 typedef enum {
     JANET_ASYNC_WRITEMODE_WRITE,
     JANET_ASYNC_WRITEMODE_SEND,
-    JANET_ASYNC_WRITEMODE_SENDTO,
-    JANET_ASYNC_WRITEMODE_CONNECT
+    JANET_ASYNC_WRITEMODE_SENDTO
 } JanetWriteMode;
 
 typedef struct {
@@ -2523,41 +2534,15 @@ typedef struct {
 #endif
 } StateWrite;
 
-static JanetAsyncStatus handle_connect(JanetListenerState *s) {
-#ifdef JANET_WINDOWS
-    int res = 0;
-    int size = sizeof(res);
-    int r = getsockopt((SOCKET)s->stream->handle, SOL_SOCKET, SO_ERROR, (char *)&res, &size);
-#else
-    int res = 0;
-    socklen_t size = sizeof res;
-    int r = getsockopt(s->stream->handle, SOL_SOCKET, SO_ERROR, &res, &size);
-#endif
-    if (r == 0) {
-        if (res == 0) {
-            janet_schedule(s->fiber, janet_wrap_abstract(s->stream));
-        } else {
-            s->stream->flags |= JANET_STREAM_TOCLOSE;
-            janet_cancel(s->fiber, janet_cstringv(strerror(res)));
-        }
-    } else {
-        s->stream->flags |= JANET_STREAM_TOCLOSE;
-        janet_cancel(s->fiber, janet_ev_lasterr());
-    }
-    return JANET_ASYNC_STATUS_DONE;
-}
-
 JanetAsyncStatus ev_machine_write(JanetListenerState *s, JanetAsyncEvent event) {
     StateWrite *state = (StateWrite *) s;
     switch (event) {
         default:
             break;
         case JANET_ASYNC_EVENT_MARK: {
-            if (state->mode != JANET_ASYNC_WRITEMODE_CONNECT) {
-                janet_mark(state->is_buffer
-                           ? janet_wrap_buffer(state->src.buf)
-                           : janet_wrap_string(state->src.str));
-            }
+            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));
             }
@@ -2579,11 +2564,6 @@ JanetAsyncStatus ev_machine_write(JanetListenerState *s, JanetAsyncEvent event)
         }
         break;
         case JANET_ASYNC_EVENT_USER: {
-#ifdef JANET_NET
-            if (state->mode == JANET_ASYNC_WRITEMODE_CONNECT) {
-                return handle_connect(s);
-            }
-#endif
             /* Begin write */
             int32_t len;
             const uint8_t *bytes;
@@ -2647,11 +2627,6 @@ JanetAsyncStatus ev_machine_write(JanetListenerState *s, JanetAsyncEvent event)
             janet_cancel(s->fiber, janet_cstringv("stream hup"));
             return JANET_ASYNC_STATUS_DONE;
         case JANET_ASYNC_EVENT_WRITE: {
-#ifdef JANET_NET
-            if (state->mode == JANET_ASYNC_WRITEMODE_CONNECT) {
-                return handle_connect(s);
-            }
-#endif
             int32_t start, len;
             const uint8_t *bytes;
             start = state->start;
@@ -2753,10 +2728,6 @@ void janet_ev_sendto_buffer(JanetStream *stream, JanetBuffer *buf, void *dest, i
 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);
 }
-
-void janet_ev_connect(JanetStream *stream, int flags) {
-    janet_ev_write_generic(stream, NULL, NULL, JANET_ASYNC_WRITEMODE_CONNECT, 0, flags);
-}
 #endif
 
 /* For a pipe ID */

src/core/fiber.c

@@ -39,8 +39,8 @@ static void fiber_reset(JanetFiber *fiber) {
     fiber->env = NULL;
     fiber->last_value = janet_wrap_nil();
 #ifdef JANET_EV
-    fiber->waiting = NULL;
     fiber->sched_id = 0;
+    fiber->waiting = NULL;
     fiber->supervisor_channel = NULL;
 #endif
     janet_fiber_set_status(fiber, JANET_STATUS_NEW);
@@ -85,7 +85,6 @@ 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;
     fiber->supervisor_channel = NULL;
 #endif
     return fiber;

src/core/gc.c

@@ -268,6 +268,9 @@ recur:
     if (fiber->supervisor_channel) {
         janet_mark_abstract(fiber->supervisor_channel);
     }
+    if (fiber->waiting) {
+        janet_mark_abstract(fiber->waiting);
+    }
 #endif
 
     /* Explicit tail recursion */
@@ -370,14 +373,15 @@ void janet_sweep() {
                     if (head->type->gc) {
                         janet_assert(!head->type->gc(head->data, head->size), "finalizer failed");
                     }
-                    /* Mark as tombstone in place */
-                    items[i].key = janet_wrap_nil();
-                    items[i].value = janet_wrap_false();
-                    janet_vm.threaded_abstracts.deleted++;
-                    janet_vm.threaded_abstracts.count--;
                     /* Free memory */
                     janet_free(janet_abstract_head(abst));
                 }
+
+                /* Mark as tombstone in place */
+                items[i].key = janet_wrap_nil();
+                items[i].value = janet_wrap_false();
+                janet_vm.threaded_abstracts.deleted++;
+                janet_vm.threaded_abstracts.count--;
             }
 
             /* Reset for next sweep */
@@ -437,6 +441,7 @@ void janet_collect(void) {
     uint32_t i;
     if (janet_vm.gc_suspend) return;
     depth = JANET_RECURSION_GUARD;
+    janet_vm.gc_mark_phase = 1;
     /* Try and prevent many major collections back to back.
      * A full collection will take O(janet_vm.block_count) time.
      * If we have a large heap, make sure our interval is not too
@@ -456,6 +461,7 @@ void janet_collect(void) {
         Janet x = janet_vm.roots[--janet_vm.root_count];
         janet_mark(x);
     }
+    janet_vm.gc_mark_phase = 0;
     janet_sweep();
     janet_vm.next_collection = 0;
     janet_free_all_scratch();
@@ -559,7 +565,9 @@ void janet_gcunlock(int handle) {
     janet_vm.gc_suspend = handle;
 }
 
-/* Scratch memory API */
+/* Scratch memory API
+ * Scratch memory allocations do not need to be free (but optionally can be), and will be automatically cleaned
+ * up in the next call to janet_collect. */
 
 void *janet_smalloc(size_t size) {
     JanetScratch *s = janet_malloc(sizeof(JanetScratch) + size);

src/core/io.c

@@ -131,7 +131,7 @@ JANET_CORE_FN(cfun_io_temp,
 }
 
 JANET_CORE_FN(cfun_io_fopen,
-              "(file/open path &opt mode)",
+              "(file/open path &opt mode buffer-size)",
               "Open a file. `path` is an absolute or relative path, and "
               "`mode` is a set of flags indicating the mode to open the file in. "
               "`mode` is a keyword where each character represents a flag. If the file "
@@ -145,7 +145,7 @@ JANET_CORE_FN(cfun_io_fopen,
               "* + - append to the file instead of overwriting it\n\n"
               "* n - error if the file cannot be opened instead of returning nil\n\n"
               "See fopen (<stdio.h>, C99) for further details.") {
-    janet_arity(argc, 1, 2);
+    janet_arity(argc, 1, 3);
     const uint8_t *fname = janet_getstring(argv, 0);
     const uint8_t *fmode;
     int32_t flags;
@@ -158,6 +158,15 @@ JANET_CORE_FN(cfun_io_fopen,
         flags = JANET_FILE_READ;
     }
     FILE *f = fopen((const char *)fname, (const char *)fmode);
+    if (f != NULL) {
+        size_t bufsize = janet_optsize(argv, argc, 2, BUFSIZ);
+        if (bufsize != BUFSIZ) {
+            int result = setvbuf(f, NULL, bufsize ? _IOFBF : _IONBF, bufsize);
+            if (result) {
+                janet_panic("failed to set buffer size for file");
+            }
+        }
+    }
     return f ? janet_makefile(f, flags)
            : (flags & JANET_FILE_NONIL) ? (janet_panicf("failed to open file %s: %s", fname, strerror(errno)), janet_wrap_nil())
            : janet_wrap_nil();

src/core/marsh.c

@@ -1048,7 +1048,6 @@ static const uint8_t *unmarshal_one_fiber(
     fiber->env = NULL;
     fiber->last_value = janet_wrap_nil();
 #ifdef JANET_EV
-    fiber->waiting = NULL;
     fiber->sched_id = 0;
     fiber->supervisor_channel = NULL;
 #endif

src/core/net.c

@@ -111,6 +111,62 @@ static void janet_net_socknoblock(JSock s) {
 #endif
 }
 
+/* State machine for async connect */
+
+typedef struct {
+    JanetListenerState head;
+    int did_connect;
+} NetStateConnect;
+
+JanetAsyncStatus net_machine_connect(JanetListenerState *s, JanetAsyncEvent event) {
+    NetStateConnect *state = (NetStateConnect *)s;
+    switch (event) {
+        default:
+            return JANET_ASYNC_STATUS_NOT_DONE;
+        case JANET_ASYNC_EVENT_CLOSE:
+            janet_cancel(s->fiber, janet_cstringv("stream closed"));
+            return JANET_ASYNC_STATUS_DONE;
+        case JANET_ASYNC_EVENT_HUP:
+        case JANET_ASYNC_EVENT_ERR:
+        case JANET_ASYNC_EVENT_COMPLETE:
+        case JANET_ASYNC_EVENT_WRITE:
+        case JANET_ASYNC_EVENT_USER:
+            break;
+    }
+    JanetStream *stream = s->stream;
+#ifdef JANET_WINDOWS
+    int res = 0;
+    int size = sizeof(res);
+    int r = getsockopt((SOCKET)stream->handle, SOL_SOCKET, SO_ERROR, (char *)&res, &size);
+#else
+    int res = 0;
+    socklen_t size = sizeof res;
+    int r = getsockopt(stream->handle, SOL_SOCKET, SO_ERROR, &res, &size);
+#endif
+    if (r == 0) {
+        if (res == 0) {
+            state->did_connect = 1;
+            janet_schedule(s->fiber, janet_wrap_abstract(s->stream));
+        } else {
+            janet_cancel(s->fiber, janet_cstringv(strerror(res)));
+            stream->flags |= JANET_STREAM_TOCLOSE;
+        }
+    } else {
+        janet_cancel(s->fiber, janet_ev_lasterr());
+        stream->flags |= JANET_STREAM_TOCLOSE;
+    }
+    return JANET_ASYNC_STATUS_DONE;
+}
+
+static void net_sched_connect(JanetStream *stream) {
+    JanetListenerState *s = janet_listen(stream, net_machine_connect, JANET_ASYNC_LISTEN_WRITE, sizeof(NetStateConnect), NULL);
+    NetStateConnect *state = (NetStateConnect *)s;
+    state->did_connect = 0;
+#ifdef JANET_WINDOWS
+    net_machine_connect(s, JANET_ASYNC_EVENT_USER);
+#endif
+}
+
 /* State machine for accepting connections. */
 
 #ifdef JANET_WINDOWS
@@ -496,7 +552,7 @@ JANET_CORE_FN(cfun_net_connect,
     }
 #endif
 
-    if (status != 0) {
+    if (status) {
 #ifdef JANET_WINDOWS
         if (err != WSAEWOULDBLOCK) {
 #else
@@ -508,9 +564,7 @@ JANET_CORE_FN(cfun_net_connect,
         }
     }
 
-    /* Handle the connect() result in the event loop*/
-    janet_ev_connect(stream, MSG_NOSIGNAL);
-
+    net_sched_connect(stream);
     janet_await();
 }
 

src/core/os.c

@@ -809,6 +809,7 @@ static void close_handle(JanetHandle handle) {
 #ifndef JANET_WINDOWS
 static void janet_signal_callback(JanetEVGenericMessage msg) {
     int sig = msg.tag;
+    if (msg.argi) janet_interpreter_interrupt_handled(NULL);
     Janet handlerv = janet_table_get(&janet_vm.signal_handlers, janet_wrap_integer(sig));
     if (!janet_checktype(handlerv, JANET_FUNCTION)) {
         /* Let another thread/process try to handle this */
@@ -825,10 +826,8 @@ static void janet_signal_callback(JanetEVGenericMessage msg) {
     }
     JanetFunction *handler = janet_unwrap_function(handlerv);
     JanetFiber *fiber = janet_fiber(handler, 64, 0, NULL);
-    janet_schedule(fiber, janet_wrap_nil());
-    if (msg.argi) {
-        janet_ev_dec_refcount();
-    }
+    janet_schedule_soon(fiber, janet_wrap_nil(), JANET_SIGNAL_OK);
+    janet_ev_dec_refcount();
 }
 
 static void janet_signal_trampoline_no_interrupt(int sig) {
@@ -837,6 +836,7 @@ static void janet_signal_trampoline_no_interrupt(int sig) {
     memset(&msg, 0, sizeof(msg));
     msg.tag = sig;
     janet_ev_post_event(&janet_vm, janet_signal_callback, msg);
+    janet_ev_inc_refcount();
 }
 
 static void janet_signal_trampoline(int sig) {
@@ -845,18 +845,20 @@ static void janet_signal_trampoline(int sig) {
     memset(&msg, 0, sizeof(msg));
     msg.tag = sig;
     msg.argi = 1;
+    janet_interpreter_interrupt(NULL);
     janet_ev_post_event(&janet_vm, janet_signal_callback, msg);
     janet_ev_inc_refcount();
-    janet_interpreter_interrupt(NULL);
 }
 #endif
 
 JANET_CORE_FN(os_sigaction,
               "(os/sigaction which &opt handler interrupt-interpreter)",
-              "Add a signal handler for a given action. Use nil for the `handler` argument to remove a signal handler.") {
+              "Add a signal handler for a given action. Use nil for the `handler` argument to remove a signal handler. "
+              "All signal handlers are the same as supported by `os/proc-kill`.") {
     janet_sandbox_assert(JANET_SANDBOX_SIGNAL);
     janet_arity(argc, 1, 3);
 #ifdef JANET_WINDOWS
+    (void) argv;
     janet_panic("unsupported on this platform");
 #else
     /* TODO - per thread signal masks */
@@ -880,7 +882,11 @@ JANET_CORE_FN(os_sigaction,
     sigfillset(&mask);
     memset(&action, 0, sizeof(action));
     if (can_interrupt) {
+#ifdef JANET_NO_INTERPRETER_INTERRUPT
+        janet_panic("interpreter interrupt not enabled");
+#else
         action.sa_handler = janet_signal_trampoline;
+#endif
     } else {
         action.sa_handler = janet_signal_trampoline_no_interrupt;
     }
@@ -1431,8 +1437,8 @@ JANET_CORE_FN(os_getenv,
     janet_sandbox_assert(JANET_SANDBOX_ENV);
     janet_arity(argc, 1, 2);
     const char *cstr = janet_getcstring(argv, 0);
-    const char *res = getenv(cstr);
     janet_lock_environ();
+    const char *res = getenv(cstr);
     Janet ret = res
                 ? janet_cstringv(res)
                 : argc == 2

src/core/run.c

@@ -57,12 +57,20 @@ int janet_dobytes(JanetTable *env, const uint8_t *bytes, int32_t len, const char
                 }
             } else {
                 ret = janet_wrap_string(cres.error);
+                int32_t line = (int32_t) parser.line;
+                int32_t col = (int32_t) parser.column;
+                if ((cres.error_mapping.line > 0) &&
+                        (cres.error_mapping.column > 0)) {
+                    line = cres.error_mapping.line;
+                    col = cres.error_mapping.column;
+                }
                 if (cres.macrofiber) {
-                    janet_eprintf("compile error in %s: ", sourcePath);
+                    janet_eprintf("%s:%d:%d: compile error", sourcePath,
+                                  line, col);
                     janet_stacktrace_ext(cres.macrofiber, ret, "");
                 } else {
-                    janet_eprintf("compile error in %s: %s\n", sourcePath,
-                                  (const char *)cres.error);
+                    janet_eprintf("%s:%d:%d: compile error: %s\n", sourcePath,
+                                  line, col, (const char *)cres.error);
                 }
                 errflags |= 0x02;
                 done = 1;

src/core/specials.c

@@ -530,6 +530,32 @@ static JanetSlot janetc_def(JanetFopts opts, int32_t argn, const Janet *argv) {
     return ret;
 }
 
+/* Check if a form matches the pattern (= nil _) or (not= nil _) */
+static int janetc_check_nil_form(JanetFopts opts, Janet x, Janet *capture, uint32_t fun_tag) {
+    if (!janet_checktype(x, JANET_TUPLE)) return 0;
+    JanetTuple tup = janet_unwrap_tuple(x);
+    if (3 != janet_tuple_length(tup)) return 0;
+    Janet op1 = tup[0];
+    if (janet_checktype(op1, JANET_SYMBOL)) {
+        Janet entry = janet_table_get(opts.compiler->env, op1);
+        if (janet_checktype(entry, JANET_TABLE)) {
+            op1 = janet_table_get(janet_unwrap_table(entry), janet_ckeywordv("value"));
+        }
+    }
+    if (!janet_checktype(op1, JANET_FUNCTION)) return 0;
+    JanetFunction *fun = janet_unwrap_function(op1);
+    uint32_t tag = fun->def->flags & JANET_FUNCDEF_FLAG_TAG;
+    if (tag != fun_tag) return 0;
+    if (janet_checktype(tup[1], JANET_NIL)) {
+        *capture = tup[2];
+        return 1;
+    } else if (janet_checktype(tup[2], JANET_NIL)) {
+        *capture = tup[1];
+        return 1;
+    }
+    return 0;
+}
+
 /*
  * :condition
  * ...
@@ -550,6 +576,7 @@ static JanetSlot janetc_if(JanetFopts opts, int32_t argn, const Janet *argv) {
     JanetScope condscope, tempscope;
     const int tail = opts.flags & JANET_FOPTS_TAIL;
     const int drop = opts.flags & JANET_FOPTS_DROP;
+    uint8_t ifnjmp = JOP_JUMP_IF_NOT;
 
     if (argn < 2 || argn > 3) {
         janetc_cerror(c, "expected 2 or 3 arguments to if");
@@ -572,7 +599,16 @@ static JanetSlot janetc_if(JanetFopts opts, int32_t argn, const Janet *argv) {
 
     /* Compile condition */
     janetc_scope(&condscope, c, 0, "if");
-    cond = janetc_value(condopts, argv[0]);
+
+    Janet condform = argv[0];
+    if (janetc_check_nil_form(opts, condform, &condform, JANET_FUN_EQ)) {
+        ifnjmp = JOP_JUMP_IF_NOT_NIL;
+    }
+    if (janetc_check_nil_form(opts, condform, &condform, JANET_FUN_NEQ)) {
+        ifnjmp = JOP_JUMP_IF_NIL;
+    }
+
+    cond = janetc_value(condopts, condform);
 
     /* Check constant condition. */
     /* TODO: Use type info for more short circuits */
@@ -595,7 +631,7 @@ static JanetSlot janetc_if(JanetFopts opts, int32_t argn, const Janet *argv) {
     }
 
     /* Compile jump to right */
-    labeljr = janetc_emit_si(c, JOP_JUMP_IF_NOT, cond, 0, 0);
+    labeljr = janetc_emit_si(c, ifnjmp, cond, 0, 0);
 
     /* Condition left body */
     janetc_scope(&tempscope, c, 0, "if-true");
@@ -605,7 +641,7 @@ static JanetSlot janetc_if(JanetFopts opts, int32_t argn, const Janet *argv) {
 
     /* Compile jump to done */
     labeljd = janet_v_count(c->buffer);
-    if (!tail) janetc_emit(c, JOP_JUMP);
+    if (!tail && !(drop && janet_checktype(falsebody, JANET_NIL))) janetc_emit(c, JOP_JUMP);
 
     /* Compile right body */
     labelr = janet_v_count(c->buffer);
@@ -741,20 +777,6 @@ static JanetSlot janetc_break(JanetFopts opts, int32_t argn, const Janet *argv)
     }
 }
 
-/* Check if a form matches the pattern (not= nil _) */
-static int janetc_check_notnil_form(Janet x, Janet *capture) {
-    if (!janet_checktype(x, JANET_TUPLE)) return 0;
-    JanetTuple tup = janet_unwrap_tuple(x);
-    if (!janet_checktype(tup[0], JANET_FUNCTION)) return 0;
-    if (3 != janet_tuple_length(tup)) return 0;
-    JanetFunction *fun = janet_unwrap_function(tup[0]);
-    uint32_t tag = fun->def->flags & JANET_FUNCDEF_FLAG_TAG;
-    if (tag != JANET_FUN_NEQ) return 0;
-    if (!janet_checktype(tup[1], JANET_NIL)) return 0;
-    *capture = tup[2];
-    return 1;
-}
-
 /*
  * :whiletop
  * ...
@@ -771,6 +793,7 @@ static JanetSlot janetc_while(JanetFopts opts, int32_t argn, const Janet *argv)
     JanetScope tempscope;
     int32_t labelwt, labeld, labeljt, labelc, i;
     int infinite = 0;
+    int is_nil_form = 0;
     int is_notnil_form = 0;
     uint8_t ifjmp = JOP_JUMP_IF;
     uint8_t ifnjmp = JOP_JUMP_IF_NOT;
@@ -784,11 +807,16 @@ static JanetSlot janetc_while(JanetFopts opts, int32_t argn, const Janet *argv)
 
     janetc_scope(&tempscope, c, JANET_SCOPE_WHILE, "while");
 
-    /* Check for `(not= nil _)` in condition, and if so, use the
+    /* Check for `(= nil _)` or `(not= nil _)` in condition, and if so, use the
      * jmpnl or jmpnn instructions. This let's us implement `(each ...)`
      * more efficiently. */
     Janet condform = argv[0];
-    if (janetc_check_notnil_form(condform, &condform)) {
+    if (janetc_check_nil_form(opts, condform, &condform, JANET_FUN_EQ)) {
+        is_nil_form = 1;
+        ifjmp = JOP_JUMP_IF_NIL;
+        ifnjmp = JOP_JUMP_IF_NOT_NIL;
+    }
+    if (janetc_check_nil_form(opts, condform, &condform, JANET_FUN_NEQ)) {
         is_notnil_form = 1;
         ifjmp = JOP_JUMP_IF_NOT_NIL;
         ifnjmp = JOP_JUMP_IF_NIL;
@@ -800,7 +828,9 @@ static JanetSlot janetc_while(JanetFopts opts, int32_t argn, const Janet *argv)
     /* Check for constant condition */
     if (cond.flags & JANET_SLOT_CONSTANT) {
         /* Loop never executes */
-        int never_executes = is_notnil_form
+        int never_executes = is_nil_form
+                             ? !janet_checktype(cond.constant, JANET_NIL)
+                             : is_notnil_form
                              ? janet_checktype(cond.constant, JANET_NIL)
                              : !janet_truthy(cond.constant);
         if (never_executes) {

src/core/state.c

@@ -24,6 +24,11 @@
 #include "features.h"
 #include <janet.h>
 #include "state.h"
+#include "util.h"
+#endif
+
+#ifdef JANET_WINDOWS
+#include <windows.h>
 #endif
 
 JANET_THREAD_LOCAL JanetVM janet_vm;
@@ -57,5 +62,18 @@ void janet_vm_load(JanetVM *from) {
  * use NULL to interrupt the current VM when convenient */
 void janet_interpreter_interrupt(JanetVM *vm) {
     vm = vm ? vm : &janet_vm;
-    vm->auto_suspend = 1;
+#ifdef JANET_WINDOWS
+    InterlockedIncrement(&vm->auto_suspend);
+#else
+    __atomic_add_fetch(&vm->auto_suspend, 1, __ATOMIC_RELAXED);
+#endif
+}
+
+void janet_interpreter_interrupt_handled(JanetVM *vm) {
+    vm = vm ? vm : &janet_vm;
+#ifdef JANET_WINDOWS
+    InterlockedDecrement(&vm->auto_suspend);
+#else
+    __atomic_add_fetch(&vm->auto_suspend, -1, __ATOMIC_RELAXED);
+#endif
 }

src/core/state.h

@@ -89,7 +89,7 @@ struct JanetVM {
 
     /* If this flag is true, suspend on function calls and backwards jumps.
      * When this occurs, this flag will be reset to 0. */
-    volatile int auto_suspend;
+    volatile int32_t auto_suspend;
 
     /* The current running fiber on the current thread.
      * Set and unset by functions in vm.c */
@@ -125,6 +125,7 @@ struct JanetVM {
     size_t next_collection;
     size_t block_count;
     int gc_suspend;
+    int gc_mark_phase;
 
     /* GC roots */
     Janet *roots;
@@ -154,12 +155,10 @@ struct JanetVM {
     JanetQueue spawn;
     JanetTimeout *tq;
     JanetRNG ev_rng;
-    JanetListenerState **listeners;
-    size_t listener_count;
-    size_t listener_cap;
     volatile size_t extra_listeners; /* used in signal handler, must be volatile */
     JanetTable threaded_abstracts; /* All abstract types that can be shared between threads (used in this thread) */
     JanetTable active_tasks; /* All possibly live task fibers - used just for tracking */
+    JanetArray *listeners; /* For GC */
     JanetTable signal_handlers;
 #ifdef JANET_WINDOWS
     void **iocp;

src/core/table.c

@@ -111,12 +111,11 @@ static void janet_table_rehash(JanetTable *t, int32_t size) {
             JANET_OUT_OF_MEMORY;
         }
     }
-    int32_t i, oldcapacity;
-    oldcapacity = t->capacity;
+    int32_t oldcapacity = t->capacity;
     t->data = newdata;
     t->capacity = size;
     t->deleted = 0;
-    for (i = 0; i < oldcapacity; i++) {
+    for (int32_t i = 0; i < oldcapacity; i++) {
         JanetKV *kv = olddata + i;
         if (!janet_checktype(kv->key, JANET_NIL)) {
             JanetKV *newkv = janet_table_find(t, kv->key);

src/core/util.h

@@ -49,7 +49,7 @@
 #ifndef JANET_EXIT
 #include <stdio.h>
 #define JANET_EXIT(m) do { \
-    fprintf(stderr, "C runtime error at line %d in file %s: %s\n",\
+    fprintf(stderr, "janet interpreter runtime error at line %d in file %s: %s\n",\
         __LINE__,\
         __FILE__,\
         (m));\

src/core/vm.c

@@ -116,7 +116,6 @@
 #else
 #define vm_maybe_auto_suspend(COND) do { \
     if ((COND) && janet_vm.auto_suspend) { \
-        janet_vm.auto_suspend = 0; \
         fiber->flags |= (JANET_FIBER_RESUME_NO_USEVAL | JANET_FIBER_RESUME_NO_SKIP); \
         vm_return(JANET_SIGNAL_INTERRUPT, janet_wrap_nil()); \
     } \
@@ -862,7 +861,7 @@ static JanetSignal run_vm(JanetFiber *fiber, Janet in) {
     vm_pcnext();
 
     VM_OP(JOP_EQUALS_IMMEDIATE)
-    stack[A] = janet_wrap_boolean(janet_unwrap_number(stack[B]) == (double) CS);
+    stack[A] = janet_wrap_boolean(janet_checktype(stack[B], JANET_NUMBER) && (janet_unwrap_number(stack[B]) == (double) CS));
     vm_pcnext();
 
     VM_OP(JOP_NOT_EQUALS)
@@ -870,7 +869,7 @@ static JanetSignal run_vm(JanetFiber *fiber, Janet in) {
     vm_pcnext();
 
     VM_OP(JOP_NOT_EQUALS_IMMEDIATE)
-    stack[A] = janet_wrap_boolean(janet_unwrap_number(stack[B]) != (double) CS);
+    stack[A] = janet_wrap_boolean(!janet_checktype(stack[B], JANET_NUMBER) || (janet_unwrap_number(stack[B]) != (double) CS));
     vm_pcnext();
 
     VM_OP(JOP_COMPARE)
@@ -1589,6 +1588,7 @@ int janet_init(void) {
     janet_vm.next_collection = 0;
     janet_vm.gc_interval = 0x400000;
     janet_vm.block_count = 0;
+    janet_vm.gc_mark_phase = 0;
 
     janet_symcache_init();
 

src/include/janet.h

@@ -235,9 +235,22 @@ extern "C" {
 #endif
 
 /* How to export symbols */
+#ifndef JANET_EXPORT
+#ifdef JANET_WINDOWS
+#define JANET_EXPORT __declspec(dllexport)
+#else
+#define JANET_EXPORT __attribute__((visibility ("default")))
+#endif
+#endif
+
+/* How declare API functions */
 #ifndef JANET_API
 #ifdef JANET_WINDOWS
+#ifdef JANET_DLL_IMPORT
+#define JANET_API __declspec(dllimport)
+#else
 #define JANET_API __declspec(dllexport)
+#endif
 #else
 #define JANET_API __attribute__((visibility ("default")))
 #endif
@@ -578,7 +591,6 @@ typedef enum {
     JANET_ASYNC_EVENT_HUP,
     JANET_ASYNC_EVENT_READ,
     JANET_ASYNC_EVENT_WRITE,
-    JANET_ASYNC_EVENT_CANCEL,
     JANET_ASYNC_EVENT_COMPLETE, /* Used on windows for IOCP */
     JANET_ASYNC_EVENT_USER
 } JanetAsyncEvent;
@@ -600,13 +612,9 @@ typedef JanetAsyncStatus(*JanetListener)(JanetListenerState *state, JanetAsyncEv
 struct JanetStream {
     JanetHandle handle;
     uint32_t flags;
-    /* Linked list of all in-flight IO routines for this stream */
-    JanetListenerState *state;
+    JanetListenerState *read_state;
+    JanetListenerState *write_state;
     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 stream, though. */
-    int _mask;
 };
 
 /* Interface for state machine based event loop */
@@ -616,14 +624,12 @@ struct JanetListenerState {
     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. */
+    uint32_t index; /* Used for GC and poll implentation */
+    uint32_t flags;
 #ifdef JANET_WINDOWS
     void *tag; /* Used to associate listeners with an overlapped structure */
     int bytes; /* Used to track how many bytes were transfered. */
 #endif
-    /* internal */
-    size_t _index;
-    int _mask;
-    JanetListenerState *_next;
 };
 #endif
 
@@ -730,6 +736,7 @@ JANET_API Janet janet_wrap_integer(int32_t x);
         ? janet_nanbox_isnumber(x) \
         : janet_nanbox_checkauxtype((x), (t)))
 
+/* Use JANET_API so that modules will use a local version of these functions if possible */
 JANET_API void *janet_nanbox_to_pointer(Janet x);
 JANET_API Janet janet_nanbox_from_pointer(void *p, uint64_t tagmask);
 JANET_API Janet janet_nanbox_from_cpointer(const void *p, uint64_t tagmask);
@@ -912,8 +919,8 @@ struct JanetFiber {
      * that is, fibers that are scheduled on the event loop and behave much like threads
      * in a multi-tasking system. It would be possible to move these fields to a new
      * type, say "JanetTask", that as separate from fibers to save a bit of space. */
-    JanetListenerState *waiting;
     uint32_t sched_id; /* Increment everytime fiber is scheduled by event loop */
+    JanetListenerState *waiting;
     void *supervisor_channel; /* Channel to push self to when complete */
 #endif
 };
@@ -1388,6 +1395,7 @@ JANET_API void janet_stream_flags(JanetStream *stream, uint32_t flags);
 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);
+JANET_API void janet_schedule_soon(JanetFiber *fiber, Janet value, JanetSignal sig);
 
 /* 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);
@@ -1484,7 +1492,6 @@ JANET_API void janet_ev_readchunk(JanetStream *stream, JanetBuffer *buf, int32_t
 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);
-JANET_API void janet_ev_connect(JanetStream *stream, int flags);
 #endif
 
 /* Write async to a stream */
@@ -1799,6 +1806,7 @@ JANET_API void janet_vm_free(JanetVM *vm);
 JANET_API void janet_vm_save(JanetVM *into);
 JANET_API void janet_vm_load(JanetVM *from);
 JANET_API void janet_interpreter_interrupt(JanetVM *vm);
+JANET_API void janet_interpreter_interrupt_handled(JanetVM *vm);
 JANET_API JanetSignal janet_continue(JanetFiber *fiber, Janet in, Janet *out);
 JANET_API JanetSignal janet_continue_signal(JanetFiber *fiber, Janet in, Janet *out, JanetSignal sig);
 JANET_API JanetSignal janet_pcall(JanetFunction *fun, int32_t argn, const Janet *argv, Janet *out, JanetFiber **f);
@@ -1946,10 +1954,10 @@ JANET_API void janet_register(const char *name, JanetCFunction cfun);
 #endif
 #ifndef JANET_ENTRY_NAME
 #define JANET_MODULE_ENTRY \
-    JANET_MODULE_PREFIX JANET_API JanetBuildConfig _janet_mod_config(void) { \
+    JANET_MODULE_PREFIX JANET_EXPORT JanetBuildConfig _janet_mod_config(void) { \
         return janet_config_current(); \
     } \
-    JANET_MODULE_PREFIX JANET_API void _janet_init
+    JANET_MODULE_PREFIX JANET_EXPORT void _janet_init
 #else
 #define JANET_MODULE_ENTRY JANET_MODULE_PREFIX JANET_API void JANET_ENTRY_NAME
 #endif

test/suite-boot.janet

@@ -186,6 +186,11 @@
 (assert (= txs [[-1 -1] [-1 0] [-1 1] [0 -1] [0 1] [1 -1] [1 0] [1 1]])
         "nested seq")
 
+# :unless modifier
+(assert (deep= (seq [i :range [0 10] :unless (odd? i)] i)
+               @[0 2 4 6 8])
+        ":unless modifier")
+
 # 515891b03
 (assert (deep= (tabseq [i :in (range 3)] i (* 3 i))
                @{0 0 1 3 2 6}))
@@ -204,6 +209,12 @@
 (assert (deep= (seq [x :down-to [10 0]] x) (seq [x :down [10 -1]] x))
         "loop :down-to")
 
+# one-term :range forms
+(assert (deep= (seq [x :range [10]] x) (seq [x :range [0 10]] x))
+        "one-term :range")
+(assert (deep= (seq [x :down [10]] x) (seq [x :down [10 0]] x))
+        "one-term :down")
+
 # 7880d7320
 (def res @{})
 (loop [[k v] :pairs @{1 2 3 4 5 6}]
@@ -343,6 +354,13 @@
         "sort 5")
 (assert (<= ;(sort (map (fn [x] (math/random)) (range 1000)))) "sort 6")
 
+# #1283  
+(assert (deep=
+          (partition 2 (generate [ i :in [:a :b :c :d :e]] i))
+          '@[(:a :b) (:c :d) (:e)]))
+(assert (= (mean (generate [i :in [2 3 5 7 11]] i))
+           5.6))
+
 # And and or
 # c16a9d846
 (assert (= (and true true) true) "and true true")

test/suite-corelib.janet

@@ -172,5 +172,10 @@
 (assert (= (length (range -10)) 0) "(range -10)")
 (assert (= (length (range 1 10)) 9) "(range 1 10)")
 
+# iterating over generator
+(assert-no-error "iterate over coro 1" (values (generate [x :range [0 10]] x)))
+(assert-no-error "iterate over coro 2" (keys (generate [x :range [0 10]] x)))
+(assert-no-error "iterate over coro 3" (pairs (generate [x :range [0 10]] x)))
+
 (end-suite)