Improve assert-no-error test helper.

No labels before declarations.

.builds/freebsd.yml

@@ -9,3 +9,4 @@ tasks:
     gmake
     gmake test
     sudo gmake install
+    sudo gmake uninstall

.builds/openbsd.yml

@@ -11,6 +11,7 @@ tasks:
     gmake test
     doas gmake install
     gmake test-install
+    doas gmake uninstall
 - meson_min: |
     cd janet
     meson setup build_meson_min --buildtype=release -Dsingle_threaded=true -Dnanbox=false -Ddynamic_modules=false -Ddocstrings=false -Dnet=false -Dsourcemaps=false -Dpeg=false -Dassembler=false -Dint_types=false -Dreduced_os=true -Dffi=false
@@ -29,4 +30,3 @@ tasks:
     ninja
     ninja test
     doas ninja install
-

.gitignore

@@ -57,6 +57,7 @@ xxd.exe
 # VSCode
 .vs
 .clangd
+.cache
 
 # Swap files
 *.swp

CHANGELOG.md

@@ -1,6 +1,14 @@
 # Changelog
 All notable changes to this project will be documented in this file.
 
+## Unreleased - ???
+- Expose atomic refcount abstraction in janet.h
+- Add `array/weak` for weak references in arrays
+- Add support for weak tables via `table/weak`, `table/weak-keys`, and `table/weak-values`.
+- Fix compiler bug with using the result of `(break x)` expression in some contexts.
+- Rework internal event loop code to be better behaved on Windows
+- Update meson build to work better on windows
+
 ## 1.31.0 - 2023-09-17
 - Report line and column when using `janet_dobytes`
 - Add `:unless` loop modifier

README.md

@@ -383,7 +383,7 @@ Usually, one of a few reasons:
 ### Can I bind to Rust/Zig/Go/Java/Nim/C++/D/Pascal/Fortran/Odin/Jai/(Some new "Systems" Programming Language)?
 
 Probably, if that language has a good interface with C. But the programmer may need to do
-some extra work to map Janet's internal memory model may need some to that of the bound language. Janet
+some extra work to map Janet's internal memory model to that of the bound language. Janet
 also uses `setjmp`/`longjmp` for non-local returns internally. This
 approach is out of favor with many programmers now and doesn't always play well with other languages
 that have exceptions or stack-unwinding.

examples/weak-tables.janet

@@ -0,0 +1,20 @@
+(def weak-k (table/weak-keys 10))
+(def weak-v (table/weak-values 10))
+(def weak-kv (table/weak 10))
+
+(put weak-kv (gensym) 10)
+(put weak-kv :hello :world)
+(put weak-k :abc123zz77asda :stuff)
+(put weak-k true :abc123zz77asda)
+(put weak-k :zyzzyz false)
+(put weak-v (gensym) 10)
+(put weak-v 20 (gensym))
+(print "before gc")
+(tracev weak-k)
+(tracev weak-v)
+(tracev weak-kv)
+(gccollect)
+(print "after gc")
+(tracev weak-k)
+(tracev weak-v)
+(tracev weak-kv)

meson.build

@@ -293,7 +293,7 @@ patched_janet = custom_target('patched-janeth',
 
 # 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/janet_' + meson.project_version() + '.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
 

src/boot/boot.janet

@@ -1232,7 +1232,7 @@
 (defdyn *debug* "Enables a built in debugger on errors and other useful features for debugging in a repl.")
 (defdyn *exit* "When set, will cause the current context to complete. Can be set to exit from repl (or file), for example.")
 (defdyn *exit-value* "Set the return value from `run-context` upon an exit. By default, `run-context` will return nil.")
-(defdyn *task-id* "When spawning a thread or fiber, the task-id can be assigned for concurrecny control.")
+(defdyn *task-id* "When spawning a thread or fiber, the task-id can be assigned for concurrency control.")
 
 (defdyn *macro-form*
   "Inside a macro, is bound to the source form that invoked the macro")
@@ -1267,7 +1267,7 @@
 
 (defn keep-syntax
   ``Creates a tuple with the tuple type and sourcemap of `before` but the
-  elements of `after`. If either one of its argements is not a tuple, returns
+  elements of `after`. If either one of its arguments is not a tuple, returns
   `after` unmodified. Useful to preserve syntactic information when transforming
   an ast in macros.``
   [before after]

src/core/abstract.c

@@ -97,14 +97,6 @@ size_t janet_os_rwlock_size(void) {
     return sizeof(void *);
 }
 
-static int32_t janet_incref(JanetAbstractHead *ab) {
-    return InterlockedIncrement((LONG volatile *) &ab->gc.data.refcount);
-}
-
-static int32_t janet_decref(JanetAbstractHead *ab) {
-    return InterlockedDecrement((LONG volatile *) &ab->gc.data.refcount);
-}
-
 void janet_os_mutex_init(JanetOSMutex *mutex) {
     InitializeCriticalSection((CRITICAL_SECTION *) mutex);
 }
@@ -157,14 +149,6 @@ size_t janet_os_rwlock_size(void) {
     return sizeof(pthread_rwlock_t);
 }
 
-static int32_t janet_incref(JanetAbstractHead *ab) {
-    return __atomic_add_fetch(&ab->gc.data.refcount, 1, __ATOMIC_RELAXED);
-}
-
-static int32_t janet_decref(JanetAbstractHead *ab) {
-    return __atomic_add_fetch(&ab->gc.data.refcount, -1, __ATOMIC_RELAXED);
-}
-
 void janet_os_mutex_init(JanetOSMutex *mutex) {
     pthread_mutexattr_t attr;
     pthread_mutexattr_init(&attr);
@@ -212,11 +196,11 @@ void janet_os_rwlock_wunlock(JanetOSRWLock *rwlock) {
 #endif
 
 int32_t janet_abstract_incref(void *abst) {
-    return janet_incref(janet_abstract_head(abst));
+    return janet_atomic_inc(&janet_abstract_head(abst)->gc.data.refcount);
 }
 
 int32_t janet_abstract_decref(void *abst) {
-    return janet_decref(janet_abstract_head(abst));
+    return janet_atomic_dec(&janet_abstract_head(abst)->gc.data.refcount);
 }
 
 #endif

src/core/array.c

@@ -30,9 +30,7 @@
 
 #include <string.h>
 
-/* Creates a new array */
-JanetArray *janet_array(int32_t capacity) {
-    JanetArray *array = janet_gcalloc(JANET_MEMORY_ARRAY, sizeof(JanetArray));
+static void janet_array_impl(JanetArray *array, int32_t capacity) {
     Janet *data = NULL;
     if (capacity > 0) {
         janet_vm.next_collection += capacity * sizeof(Janet);
@@ -44,6 +42,19 @@ JanetArray *janet_array(int32_t capacity) {
     array->count = 0;
     array->capacity = capacity;
     array->data = data;
+}
+
+/* Creates a new array */
+JanetArray *janet_array(int32_t capacity) {
+    JanetArray *array = janet_gcalloc(JANET_MEMORY_ARRAY, sizeof(JanetArray));
+    janet_array_impl(array, capacity);
+    return array;
+}
+
+/* Creates a new array with weak references */
+JanetArray *janet_array_weak(int32_t capacity) {
+    JanetArray *array = janet_gcalloc(JANET_MEMORY_ARRAY_WEAK, sizeof(JanetArray));
+    janet_array_impl(array, capacity);
     return array;
 }
 
@@ -132,6 +143,15 @@ JANET_CORE_FN(cfun_array_new,
     return janet_wrap_array(array);
 }
 
+JANET_CORE_FN(cfun_array_weak,
+              "(array/weak capacity)",
+              "Creates a new empty array with a pre-allocated capacity and support for weak references. Similar to `array/new`.") {
+    janet_fixarity(argc, 1);
+    int32_t cap = janet_getinteger(argv, 0);
+    JanetArray *array = janet_array_weak(cap);
+    return janet_wrap_array(array);
+}
+
 JANET_CORE_FN(cfun_array_new_filled,
               "(array/new-filled count &opt value)",
               "Creates a new array of `count` elements, all set to `value`, which defaults to nil. Returns the new array.") {
@@ -352,6 +372,7 @@ JANET_CORE_FN(cfun_array_clear,
 void janet_lib_array(JanetTable *env) {
     JanetRegExt array_cfuns[] = {
         JANET_CORE_REG("array/new", cfun_array_new),
+        JANET_CORE_REG("array/weak", cfun_array_weak),
         JANET_CORE_REG("array/new-filled", cfun_array_new_filled),
         JANET_CORE_REG("array/fill", cfun_array_fill),
         JANET_CORE_REG("array/pop", cfun_array_pop),

src/core/capi.c

@@ -491,6 +491,24 @@ void *janet_optabstract(const Janet *argv, int32_t argc, int32_t n, const JanetA
     return janet_getabstract(argv, n, at);
 }
 
+/* Atomic refcounts */
+
+JanetAtomicInt janet_atomic_inc(JanetAtomicInt volatile *x) {
+#ifdef JANET_WINDOWS
+    return InterlockedIncrement(x);
+#else
+    return __atomic_add_fetch(x, 1, __ATOMIC_RELAXED);
+#endif
+}
+
+JanetAtomicInt janet_atomic_dec(JanetAtomicInt volatile *x) {
+#ifdef JANET_WINDOWS
+    return InterlockedDecrement(x);
+#else
+    return __atomic_add_fetch(x, -1, __ATOMIC_RELAXED);
+#endif
+}
+
 /* Some definitions for function-like macros */
 
 JANET_API JanetStructHead *(janet_struct_head)(JanetStruct st) {

src/core/emit.c

@@ -26,6 +26,7 @@
 #include "emit.h"
 #include "vector.h"
 #include "regalloc.h"
+#include "util.h"
 #endif
 
 /* Get a register */
@@ -128,7 +129,8 @@ static void janetc_movenear(JanetCompiler *c,
                     ((uint32_t)(src.envindex) << 16) |
                     ((uint32_t)(dest) << 8) |
                     JOP_LOAD_UPVALUE);
-    } else if (src.index > 0xFF || src.index != dest) {
+    } else if (src.index != dest) {
+        janet_assert(src.index >= 0, "bad slot");
         janetc_emit(c,
                     ((uint32_t)(src.index) << 16) |
                     ((uint32_t)(dest) << 8) |
@@ -155,6 +157,7 @@ static void janetc_moveback(JanetCompiler *c,
                     ((uint32_t)(src) << 8) |
                     JOP_SET_UPVALUE);
     } else if (dest.index != src) {
+        janet_assert(dest.index >= 0, "bad slot");
         janetc_emit(c,
                     ((uint32_t)(dest.index) << 16) |
                     ((uint32_t)(src) << 8) |

src/core/ffi.c

@@ -1381,7 +1381,7 @@ JANET_CORE_FN(cfun_ffi_buffer_write,
               "(ffi/write ffi-type data &opt buffer index)",
               "Append a native type to a buffer such as it would appear in memory. This can be used "
               "to pass pointers to structs in the ffi, or send C/C++/native structs over the network "
-              "or to files. Returns a modifed buffer or a new buffer if one is not supplied.") {
+              "or to files. Returns a modified buffer or a new buffer if one is not supplied.") {
     janet_sandbox_assert(JANET_SANDBOX_FFI_USE);
     janet_arity(argc, 2, 4);
     JanetFFIType type = decode_ffi_type(argv[0]);
@@ -1548,7 +1548,7 @@ JANET_CORE_FN(cfun_ffi_pointer_cfunction,
 
 JANET_CORE_FN(cfun_ffi_supported_calling_conventions,
               "(ffi/calling-conventions)",
-              "Get an array of all supported calling conventions on the current arhcitecture. Some architectures may have some FFI "
+              "Get an array of all supported calling conventions on the current architecture. Some architectures may have some FFI "
               "functionality (ffi/malloc, ffi/free, ffi/read, ffi/write, etc.) but not support "
               "any calling conventions. This function can be used to get all supported calling conventions "
               "that can be used on this architecture. All architectures support the :none calling "

src/core/fiber.c

@@ -40,7 +40,9 @@ static void fiber_reset(JanetFiber *fiber) {
     fiber->last_value = janet_wrap_nil();
 #ifdef JANET_EV
     fiber->sched_id = 0;
-    fiber->waiting = NULL;
+    fiber->ev_callback = NULL;
+    fiber->ev_state = NULL;
+    fiber->ev_stream = NULL;
     fiber->supervisor_channel = NULL;
 #endif
     janet_fiber_set_status(fiber, JANET_STATUS_NEW);

src/core/fiber.h

@@ -59,6 +59,9 @@
 #define JANET_FIBER_EV_FLAG_CANCELED 0x10000
 #define JANET_FIBER_EV_FLAG_SUSPENDED 0x20000
 #define JANET_FIBER_FLAG_ROOT 0x40000
+#define JANET_FIBER_EV_FLAG_IN_FLIGHT 0x1
+
+/* used only on windows, should otherwise be unset */
 
 #define janet_fiber_set_status(f, s) do {\
     (f)->flags &= ~JANET_FIBER_STATUS_MASK;\

src/core/gc.c

@@ -132,6 +132,24 @@ static void janet_mark_many(const Janet *values, int32_t n) {
     }
 }
 
+/* Mark a bunch of key values items in memory */
+static void janet_mark_keys(const JanetKV *kvs, int32_t n) {
+    const JanetKV *end = kvs + n;
+    while (kvs < end) {
+        janet_mark(kvs->key);
+        kvs++;
+    }
+}
+
+/* Mark a bunch of key values items in memory */
+static void janet_mark_values(const JanetKV *kvs, int32_t n) {
+    const JanetKV *end = kvs + n;
+    while (kvs < end) {
+        janet_mark(kvs->value);
+        kvs++;
+    }
+}
+
 /* Mark a bunch of key values items in memory */
 static void janet_mark_kvs(const JanetKV *kvs, int32_t n) {
     const JanetKV *end = kvs + n;
@@ -146,7 +164,9 @@ static void janet_mark_array(JanetArray *array) {
     if (janet_gc_reachable(array))
         return;
     janet_gc_mark(array);
-    janet_mark_many(array->data, array->count);
+    if (janet_gc_type((JanetGCObject *) array) == JANET_MEMORY_ARRAY) {
+        janet_mark_many(array->data, array->count);
+    }
 }
 
 static void janet_mark_table(JanetTable *table) {
@@ -154,7 +174,15 @@ recur: /* Manual tail recursion */
     if (janet_gc_reachable(table))
         return;
     janet_gc_mark(table);
-    janet_mark_kvs(table->data, table->capacity);
+    enum JanetMemoryType memtype = janet_gc_type(table);
+    if (memtype == JANET_MEMORY_TABLE_WEAKK) {
+        janet_mark_values(table->data, table->capacity);
+    } else if (memtype == JANET_MEMORY_TABLE_WEAKV) {
+        janet_mark_keys(table->data, table->capacity);
+    } else if (memtype == JANET_MEMORY_TABLE) {
+        janet_mark_kvs(table->data, table->capacity);
+    }
+    /* do nothing for JANET_MEMORY_TABLE_WEAKKV */
     if (table->proto) {
         table = table->proto;
         goto recur;
@@ -268,8 +296,11 @@ recur:
     if (fiber->supervisor_channel) {
         janet_mark_abstract(fiber->supervisor_channel);
     }
-    if (fiber->waiting) {
-        janet_mark_abstract(fiber->waiting);
+    if (fiber->ev_stream) {
+        janet_mark_abstract(fiber->ev_stream);
+    }
+    if (fiber->ev_callback) {
+        fiber->ev_callback(fiber, JANET_ASYNC_EVENT_MARK);
     }
 #endif
 
@@ -295,9 +326,17 @@ static void janet_deinit_block(JanetGCObject *mem) {
         case JANET_MEMORY_TABLE:
             janet_free(((JanetTable *) mem)->data);
             break;
-        case JANET_MEMORY_FIBER:
-            janet_free(((JanetFiber *)mem)->data);
-            break;
+        case JANET_MEMORY_FIBER: {
+            JanetFiber *f = (JanetFiber *)mem;
+#ifdef JANET_EV
+            if (f->ev_state && !(f->flags & JANET_FIBER_EV_FLAG_IN_FLIGHT)) {
+                janet_ev_dec_refcount();
+                janet_free(f->ev_state);
+            }
+#endif
+            janet_free(f->data);
+        }
+        break;
         case JANET_MEMORY_BUFFER:
             janet_buffer_deinit((JanetBuffer *) mem);
             break;
@@ -329,12 +368,98 @@ static void janet_deinit_block(JanetGCObject *mem) {
     }
 }
 
+/* Check that a value x has been visited in the mark phase */
+static int janet_check_liveref(Janet x) {
+    switch (janet_type(x)) {
+        default:
+            return 1;
+        case JANET_ARRAY:
+        case JANET_TABLE:
+        case JANET_FUNCTION:
+        case JANET_BUFFER:
+        case JANET_FIBER:
+            return janet_gc_reachable(janet_unwrap_pointer(x));
+        case JANET_STRING:
+        case JANET_SYMBOL:
+        case JANET_KEYWORD:
+            return janet_gc_reachable(janet_string_head(janet_unwrap_string(x)));
+        case JANET_ABSTRACT:
+            return janet_gc_reachable(janet_abstract_head(janet_unwrap_abstract(x)));
+        case JANET_TUPLE:
+            return janet_gc_reachable(janet_tuple_head(janet_unwrap_tuple(x)));
+        case JANET_STRUCT:
+            return janet_gc_reachable(janet_struct_head(janet_unwrap_struct(x)));
+    }
+}
+
 /* Iterate over all allocated memory, and free memory that is not
  * marked as reachable. Flip the gc color flag for next sweep. */
 void janet_sweep() {
     JanetGCObject *previous = NULL;
-    JanetGCObject *current = janet_vm.blocks;
+    JanetGCObject *current = janet_vm.weak_blocks;
     JanetGCObject *next;
+
+    /* Sweep weak heap to drop weak refs */
+    while (NULL != current) {
+        next = current->data.next;
+        if (current->flags & (JANET_MEM_REACHABLE | JANET_MEM_DISABLED)) {
+            /* Check for dead references */
+            enum JanetMemoryType type = janet_gc_type(current);
+            if (type == JANET_MEMORY_ARRAY_WEAK) {
+                JanetArray *array = (JanetArray *) current;
+                for (uint32_t i = 0; i < (uint32_t) array->count; i++) {
+                    if (!janet_check_liveref(array->data[i])) {
+                        array->data[i] = janet_wrap_nil();
+                    }
+                }
+            } else {
+                JanetTable *table = (JanetTable *) current;
+                int check_values = (type == JANET_MEMORY_TABLE_WEAKV) || (type == JANET_MEMORY_TABLE_WEAKKV);
+                int check_keys = (type == JANET_MEMORY_TABLE_WEAKK) || (type == JANET_MEMORY_TABLE_WEAKKV);
+                JanetKV *end = table->data + table->capacity;
+                JanetKV *kvs = table->data;
+                while (kvs < end) {
+                    int drop = 0;
+                    if (check_keys && !janet_check_liveref(kvs->key)) drop = 1;
+                    if (check_values && !janet_check_liveref(kvs->value)) drop = 1;
+                    if (drop) {
+                        /* Inlined from janet_table_remove without search */
+                        table->count--;
+                        table->deleted++;
+                        kvs->key = janet_wrap_nil();
+                        kvs->value = janet_wrap_false();
+                    }
+                    kvs++;
+                }
+            }
+        }
+        current = next;
+    }
+
+    /* Sweep weak heap to free blocks */
+    previous = NULL;
+    current = janet_vm.weak_blocks;
+    while (NULL != current) {
+        next = current->data.next;
+        if (current->flags & (JANET_MEM_REACHABLE | JANET_MEM_DISABLED)) {
+            previous = current;
+            current->flags &= ~JANET_MEM_REACHABLE;
+        } else {
+            janet_vm.block_count--;
+            janet_deinit_block(current);
+            if (NULL != previous) {
+                previous->data.next = next;
+            } else {
+                janet_vm.weak_blocks = next;
+            }
+            janet_free(current);
+        }
+        current = next;
+    }
+
+    /* Sweep main heap to free blocks */
+    previous = NULL;
+    current = janet_vm.blocks;
     while (NULL != current) {
         next = current->data.next;
         if (current->flags & (JANET_MEM_REACHABLE | JANET_MEM_DISABLED)) {
@@ -352,6 +477,7 @@ void janet_sweep() {
         }
         current = next;
     }
+
 #ifdef JANET_EV
     /* Sweep threaded abstract types for references to decrement */
     JanetKV *items = janet_vm.threaded_abstracts.data;
@@ -409,8 +535,15 @@ void *janet_gcalloc(enum JanetMemoryType type, size_t size) {
 
     /* Prepend block to heap list */
     janet_vm.next_collection += size;
-    mem->data.next = janet_vm.blocks;
-    janet_vm.blocks = mem;
+    if (type < JANET_MEMORY_TABLE_WEAKK) {
+        /* normal heap */
+        mem->data.next = janet_vm.blocks;
+        janet_vm.blocks = mem;
+    } else {
+        /* weak heap */
+        mem->data.next = janet_vm.weak_blocks;
+        janet_vm.weak_blocks = mem;
+    }
     janet_vm.block_count++;
 
     return (void *)mem;
@@ -442,7 +575,7 @@ void janet_collect(void) {
     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.
+    /* Try to 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
      * small so we won't make many collections over it. This is just a

src/core/gc.h

@@ -57,6 +57,10 @@ enum JanetMemoryType {
     JANET_MEMORY_FUNCENV,
     JANET_MEMORY_FUNCDEF,
     JANET_MEMORY_THREADED_ABSTRACT,
+    JANET_MEMORY_TABLE_WEAKK,
+    JANET_MEMORY_TABLE_WEAKV,
+    JANET_MEMORY_TABLE_WEAKKV,
+    JANET_MEMORY_ARRAY_WEAK
 };
 
 /* To allocate collectable memory, one must call janet_alloc, initialize the memory,

src/core/inttypes.c

@@ -239,7 +239,7 @@ JANET_CORE_FN(cfun_to_bytes,
               "Write the bytes of an `int/s64` or `int/u64` into a buffer.\n"
               "The `buffer` parameter specifies an existing buffer to write to, if unset a new buffer will be created.\n"
               "Returns the modified buffer.\n"
-              "The `endianness` paramater indicates the byte order:\n"
+              "The `endianness` parameter indicates the byte order:\n"
               "- `nil` (unset): system byte order\n"
               "- `:le`: little-endian, least significant byte first\n"
               "- `:be`: big-endian, most significant byte first\n") {

src/core/marsh.c

@@ -1050,6 +1050,9 @@ static const uint8_t *unmarshal_one_fiber(
 #ifdef JANET_EV
     fiber->sched_id = 0;
     fiber->supervisor_channel = NULL;
+    fiber->ev_state = NULL;
+    fiber->ev_callback = NULL;
+    fiber->ev_stream = NULL;
 #endif
 
     /* Push fiber to seen stack */

src/core/math.c

@@ -119,7 +119,7 @@ double janet_rng_double(JanetRNG *rng) {
 
 JANET_CORE_FN(cfun_rng_make,
               "(math/rng &opt seed)",
-              "Creates a Psuedo-Random number generator, with an optional seed. "
+              "Creates a Pseudo-Random number generator, with an optional seed. "
               "The seed should be an unsigned 32 bit integer or a buffer. "
               "Do not use this for cryptography. Returns a core/rng abstract type."
              ) {
@@ -411,11 +411,11 @@ void janet_lib_math(JanetTable *env) {
     JANET_CORE_DEF(env, "math/int32-min", janet_wrap_number(INT32_MIN),
                    "The minimum contiguous integer representable by a 32 bit signed integer");
     JANET_CORE_DEF(env, "math/int32-max", janet_wrap_number(INT32_MAX),
-                   "The maximum contiguous integer represtenable by a 32 bit signed integer");
+                   "The maximum contiguous integer representable by a 32 bit signed integer");
     JANET_CORE_DEF(env, "math/int-min", janet_wrap_number(JANET_INTMIN_DOUBLE),
                    "The minimum contiguous integer representable by a double (2^53)");
     JANET_CORE_DEF(env, "math/int-max", janet_wrap_number(JANET_INTMAX_DOUBLE),
-                   "The maximum contiguous integer represtenable by a double (-(2^53))");
+                   "The maximum contiguous integer representable by a double (-(2^53))");
 #ifdef NAN
     JANET_CORE_DEF(env, "math/nan", janet_wrap_number(NAN), "Not a number (IEEE-754 NaN)");
 #else

src/core/net.c

@@ -24,6 +24,7 @@
 #include "features.h"
 #include <janet.h>
 #include "util.h"
+#include "fiber.h"
 #endif
 
 #ifdef JANET_NET
@@ -114,26 +115,20 @@ static void janet_net_socknoblock(JSock s) {
 /* 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;
+void net_callback_connect(JanetFiber *fiber, JanetAsyncEvent event) {
+    JanetStream *stream = fiber->ev_stream;
+    NetStateConnect *state = (NetStateConnect *)fiber->ev_state;
     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;
+        case JANET_ASYNC_EVENT_CLOSE:
+            janet_cancel(fiber, janet_cstringv("stream closed"));
+            janet_async_end(fiber);
+            return;
     }
-    JanetStream *stream = s->stream;
 #ifdef JANET_WINDOWS
     int res = 0;
     int size = sizeof(res);
@@ -146,24 +141,24 @@ JanetAsyncStatus net_machine_connect(JanetListenerState *s, JanetAsyncEvent even
     if (r == 0) {
         if (res == 0) {
             state->did_connect = 1;
-            janet_schedule(s->fiber, janet_wrap_abstract(s->stream));
+            janet_schedule(fiber, janet_wrap_abstract(stream));
         } else {
-            janet_cancel(s->fiber, janet_cstringv(strerror(res)));
+            janet_cancel(fiber, janet_cstringv(strerror(res)));
             stream->flags |= JANET_STREAM_TOCLOSE;
         }
     } else {
-        janet_cancel(s->fiber, janet_ev_lasterr());
+        janet_cancel(fiber, janet_ev_lasterr());
         stream->flags |= JANET_STREAM_TOCLOSE;
     }
-    return JANET_ASYNC_STATUS_DONE;
+    janet_async_end(fiber);
 }
 
 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;
+    JanetFiber *f = janet_vm.root_fiber;
+    NetStateConnect *state = (NetStateConnect *) janet_async_start(f, stream, JANET_ASYNC_LISTEN_WRITE, net_callback_connect, sizeof(NetStateConnect));
     state->did_connect = 0;
 #ifdef JANET_WINDOWS
-    net_machine_connect(s, JANET_ASYNC_EVENT_USER);
+    net_callback_connect(f, JANET_ASYNC_EVENT_USER);
 #endif
 }
 
@@ -172,7 +167,6 @@ static void net_sched_connect(JanetStream *stream) {
 #ifdef JANET_WINDOWS
 
 typedef struct {
-    JanetListenerState head;
     WSAOVERLAPPED overlapped;
     JanetFunction *function;
     JanetStream *lstream;
@@ -180,10 +174,10 @@ typedef struct {
     char buf[1024];
 } NetStateAccept;
 
-static int net_sched_accept_impl(NetStateAccept *state, Janet *err);
+static int net_sched_accept_impl(NetStateAccept *state, JanetFiber *fiber, Janet *err);
 
-JanetAsyncStatus net_machine_accept(JanetListenerState *s, JanetAsyncEvent event) {
-    NetStateAccept *state = (NetStateAccept *)s;
+void net_callback_accept(JanetFiber *fiber, JanetAsyncEvent event) {
+    NetStateAccept *state = (NetStateAccept *)fiber->ev_state;
     switch (event) {
         default:
             break;
@@ -194,55 +188,58 @@ JanetAsyncStatus net_machine_accept(JanetListenerState *s, JanetAsyncEvent event
             break;
         }
         case JANET_ASYNC_EVENT_CLOSE:
-            janet_schedule(s->fiber, janet_wrap_nil());
-            return JANET_ASYNC_STATUS_DONE;
+            janet_schedule(fiber, janet_wrap_nil());
+            janet_async_end(fiber);
+            return;
         case JANET_ASYNC_EVENT_COMPLETE: {
             if (state->astream->flags & JANET_STREAM_CLOSED) {
-                janet_cancel(s->fiber, janet_cstringv("failed to accept connection"));
-                return JANET_ASYNC_STATUS_DONE;
+                janet_cancel(fiber, janet_cstringv("failed to accept connection"));
+                janet_async_end(fiber);
+                return;
             }
             SOCKET lsock = (SOCKET) state->lstream->handle;
             if (NO_ERROR != setsockopt((SOCKET) state->astream->handle, SOL_SOCKET, SO_UPDATE_ACCEPT_CONTEXT,
                                        (char *) &lsock, sizeof(lsock))) {
-                janet_cancel(s->fiber, janet_cstringv("failed to accept connection"));
-                return JANET_ASYNC_STATUS_DONE;
+                janet_cancel(fiber, janet_cstringv("failed to accept connection"));
+                janet_async_end(fiber);
+                return;
             }
 
             Janet streamv = janet_wrap_abstract(state->astream);
             if (state->function) {
                 /* Schedule worker */
-                JanetFiber *fiber = janet_fiber(state->function, 64, 1, &streamv);
-                fiber->supervisor_channel = s->fiber->supervisor_channel;
-                janet_schedule(fiber, janet_wrap_nil());
+                JanetFiber *sub_fiber = janet_fiber(state->function, 64, 1, &streamv);
+                sub_fiber->supervisor_channel = fiber->supervisor_channel;
+                janet_schedule(sub_fiber, janet_wrap_nil());
                 /* Now listen again for next connection */
                 Janet err;
-                if (net_sched_accept_impl(state, &err)) {
-                    janet_cancel(s->fiber, err);
-                    return JANET_ASYNC_STATUS_DONE;
+                if (net_sched_accept_impl(state, fiber, &err)) {
+                    janet_cancel(fiber, err);
+                    janet_async_end(fiber);
+                    return;
                 }
             } else {
-                janet_schedule(s->fiber, streamv);
-                return JANET_ASYNC_STATUS_DONE;
+                janet_schedule(fiber, streamv);
+                janet_async_end(fiber);
+                return;
             }
         }
     }
-    return JANET_ASYNC_STATUS_NOT_DONE;
 }
 
 JANET_NO_RETURN static void janet_sched_accept(JanetStream *stream, JanetFunction *fun) {
     Janet err;
-    JanetListenerState *s = janet_listen(stream, net_machine_accept, JANET_ASYNC_LISTEN_READ, sizeof(NetStateAccept), NULL);
-    NetStateAccept *state = (NetStateAccept *)s;
+    JanetFiber *f = janet_vm.root_fiber;
+    NetStateAccept *state = (NetStateAccept *) janet_async_start(f, stream, JANET_ASYNC_LISTEN_READ, net_callback_accept, sizeof(NetStateAccept));
     memset(&state->overlapped, 0, sizeof(WSAOVERLAPPED));
     memset(&state->buf, 0, 1024);
     state->function = fun;
     state->lstream = stream;
-    s->tag = &state->overlapped;
-    if (net_sched_accept_impl(state, &err)) janet_panicv(err);
+    if (net_sched_accept_impl(state, f, &err)) janet_panicv(err);
     janet_await();
 }
 
-static int net_sched_accept_impl(NetStateAccept *state, Janet *err) {
+static int net_sched_accept_impl(NetStateAccept *state, JanetFiber *fiber, 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) {
@@ -254,7 +251,11 @@ static int net_sched_accept_impl(NetStateAccept *state, Janet *err) {
     int socksize = sizeof(SOCKADDR_STORAGE) + 16;
     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 */
+        if (code == WSA_IO_PENDING) {
+            /* indicates io is happening async */
+            fiber->flags |= JANET_FIBER_EV_FLAG_IN_FLIGHT;
+            return 0;
+        }
         *err = janet_ev_lasterr();
         return 1;
     }
@@ -264,12 +265,12 @@ static int net_sched_accept_impl(NetStateAccept *state, Janet *err) {
 #else
 
 typedef struct {
-    JanetListenerState head;
     JanetFunction *function;
 } NetStateAccept;
 
-JanetAsyncStatus net_machine_accept(JanetListenerState *s, JanetAsyncEvent event) {
-    NetStateAccept *state = (NetStateAccept *)s;
+void net_callback_accept(JanetFiber *fiber, JanetAsyncEvent event) {
+    JanetStream *stream = fiber->ev_stream;
+    NetStateAccept *state = (NetStateAccept *)fiber->ev_state;
     switch (event) {
         default:
             break;
@@ -278,37 +279,41 @@ JanetAsyncStatus net_machine_accept(JanetListenerState *s, JanetAsyncEvent event
             break;
         }
         case JANET_ASYNC_EVENT_CLOSE:
-            janet_schedule(s->fiber, janet_wrap_nil());
-            return JANET_ASYNC_STATUS_DONE;
+            janet_schedule(fiber, janet_wrap_nil());
+            janet_async_end(fiber);
+            return;
+        case JANET_ASYNC_EVENT_USER:
         case JANET_ASYNC_EVENT_READ: {
 #if defined(JANET_LINUX)
-            JSock connfd = accept4(s->stream->handle, NULL, NULL, SOCK_CLOEXEC);
+            JSock connfd = accept4(stream->handle, NULL, NULL, SOCK_CLOEXEC);
 #else
             /* On BSDs, CLOEXEC should be inherited from server socket */
-            JSock connfd = accept(s->stream->handle, NULL, NULL);
+            JSock connfd = accept(stream->handle, NULL, NULL);
 #endif
             if (JSOCKVALID(connfd)) {
                 janet_net_socknoblock(connfd);
                 JanetStream *stream = make_stream(connfd, JANET_STREAM_READABLE | JANET_STREAM_WRITABLE);
                 Janet streamv = janet_wrap_abstract(stream);
                 if (state->function) {
-                    JanetFiber *fiber = janet_fiber(state->function, 64, 1, &streamv);
-                    fiber->supervisor_channel = s->fiber->supervisor_channel;
-                    janet_schedule(fiber, janet_wrap_nil());
+                    JanetFiber *sub_fiber = janet_fiber(state->function, 64, 1, &streamv);
+                    sub_fiber->supervisor_channel = fiber->supervisor_channel;
+                    janet_schedule(sub_fiber, janet_wrap_nil());
                 } else {
-                    janet_schedule(s->fiber, streamv);
-                    return JANET_ASYNC_STATUS_DONE;
+                    janet_schedule(fiber, streamv);
+                    janet_async_end(fiber);
+                    return;
                 }
             }
             break;
         }
     }
-    return JANET_ASYNC_STATUS_NOT_DONE;
 }
 
 JANET_NO_RETURN static void janet_sched_accept(JanetStream *stream, JanetFunction *fun) {
-    NetStateAccept *state = (NetStateAccept *) janet_listen(stream, net_machine_accept, JANET_ASYNC_LISTEN_READ, sizeof(NetStateAccept), NULL);
+    JanetFiber *f = janet_vm.root_fiber;
+    NetStateAccept *state = (NetStateAccept *) janet_async_start(f, stream, JANET_ASYNC_LISTEN_READ, net_callback_accept, sizeof(NetStateAccept));
     state->function = fun;
+    net_callback_accept(f, JANET_ASYNC_EVENT_USER);
     janet_await();
 }
 

src/core/os.c

@@ -517,7 +517,6 @@ static JanetEVGenericMessage janet_proc_wait_subr(JanetEVGenericMessage args) {
 
 /* Callback that is called in main thread when subroutine completes. */
 static void janet_proc_wait_cb(JanetEVGenericMessage args) {
-    janet_ev_dec_refcount();
     JanetProc *proc = (JanetProc *) args.argp;
     if (NULL != proc) {
         int status = args.tag;
@@ -827,7 +826,6 @@ static void janet_signal_callback(JanetEVGenericMessage msg) {
     JanetFunction *handler = janet_unwrap_function(handlerv);
     JanetFiber *fiber = janet_fiber(handler, 64, 0, NULL);
     janet_schedule_soon(fiber, janet_wrap_nil(), JANET_SIGNAL_OK);
-    janet_ev_dec_refcount();
 }
 
 static void janet_signal_trampoline_no_interrupt(int sig) {
@@ -836,7 +834,6 @@ 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) {
@@ -847,7 +844,6 @@ static void janet_signal_trampoline(int sig) {
     msg.argi = 1;
     janet_interpreter_interrupt(NULL);
     janet_ev_post_event(&janet_vm, janet_signal_callback, msg);
-    janet_ev_inc_refcount();
 }
 #endif
 
@@ -2340,6 +2336,34 @@ JANET_CORE_FN(os_permission_int,
     return janet_wrap_integer(os_get_unix_mode(argv, 0));
 }
 
+JANET_CORE_FN(os_posix_fork,
+              "(os/posix-fork)",
+              "Make a `fork` system call and create a new process. Return nil if in the new process, otherwise a core/process object (as returned by os/spawn). "
+              "Not supported on all systems (POSIX only).") {
+    janet_sandbox_assert(JANET_SANDBOX_SUBPROCESS);
+    janet_fixarity(argc, 0);
+    (void) argv;
+#ifdef JANET_WINDOWS
+    janet_panic("not supported");
+#else
+    pid_t result;
+    do {
+        result = fork();
+    } while (result == -1 && errno == EINTR);
+    if (result == -1) {
+        janet_panic(strerror(errno));
+    }
+    if (result) {
+        JanetProc *proc = janet_abstract(&ProcAT, sizeof(JanetProc));
+        memset(proc, 0, sizeof(JanetProc));
+        proc->pid = result;
+        proc->flags = JANET_PROC_ALLOW_ZOMBIE;
+        return janet_wrap_abstract(proc);
+    }
+    return janet_wrap_nil();
+#endif
+}
+
 #ifdef JANET_EV
 
 /*
@@ -2626,6 +2650,7 @@ void janet_lib_os(JanetTable *env) {
         JANET_CORE_REG("os/execute", os_execute),
         JANET_CORE_REG("os/spawn", os_spawn),
         JANET_CORE_REG("os/shell", os_shell),
+        JANET_CORE_REG("os/posix-fork", os_posix_fork),
         /* no need to sandbox process management if you can't create processes
          * (allows for limited functionality if use exposes C-functions to create specific processes) */
         JANET_CORE_REG("os/proc-wait", os_proc_wait),

src/core/specials.c

@@ -752,9 +752,8 @@ static JanetSlot janetc_break(JanetFopts opts, int32_t argn, const Janet *argv)
         if (!(scope->flags & JANET_SCOPE_WHILE) && argn) {
             /* Closure body with return argument */
             subopts.flags |= JANET_FOPTS_TAIL;
-            JanetSlot ret = janetc_value(subopts, argv[0]);
-            ret.flags |= JANET_SLOT_RETURNED;
-            return ret;
+            janetc_value(subopts, argv[0]);
+            return janetc_cslot(janet_wrap_nil());
         } else {
             /* while loop IIFE or no argument */
             if (argn) {
@@ -762,9 +761,7 @@ static JanetSlot janetc_break(JanetFopts opts, int32_t argn, const Janet *argv)
                 janetc_value(subopts, argv[0]);
             }
             janetc_emit(c, JOP_RETURN_NIL);
-            JanetSlot s = janetc_cslot(janet_wrap_nil());
-            s.flags |= JANET_SLOT_RETURNED;
-            return s;
+            return janetc_cslot(janet_wrap_nil());
         }
     } else {
         if (argn) {

src/core/state.c

@@ -62,18 +62,10 @@ 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;
-#ifdef JANET_WINDOWS
-    InterlockedIncrement(&vm->auto_suspend);
-#else
-    __atomic_add_fetch(&vm->auto_suspend, 1, __ATOMIC_RELAXED);
-#endif
+    janet_atomic_inc(&vm->auto_suspend);
 }
 
 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
+    janet_atomic_dec(&vm->auto_suspend);
 }

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 int32_t auto_suspend;
+    volatile JanetAtomicInt auto_suspend;
 
     /* The current running fiber on the current thread.
      * Set and unset by functions in vm.c */
@@ -121,6 +121,7 @@ struct JanetVM {
 
     /* Garbage collection */
     void *blocks;
+    void *weak_blocks;
     size_t gc_interval;
     size_t next_collection;
     size_t block_count;
@@ -155,10 +156,9 @@ struct JanetVM {
     JanetQueue spawn;
     JanetTimeout *tq;
     JanetRNG ev_rng;
-    volatile size_t extra_listeners; /* used in signal handler, must be volatile */
+    volatile JanetAtomicInt listener_count; /* 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;
@@ -175,6 +175,9 @@ struct JanetVM {
     int timer;
     int timer_enabled;
 #else
+    JanetStream **streams;
+    size_t stream_count;
+    size_t stream_capacity;
     pthread_attr_t new_thread_attr;
     JanetHandle selfpipe[2];
     struct pollfd *fds;

src/core/string.c

@@ -554,7 +554,7 @@ JANET_CORE_FN(cfun_string_format,
               "- `j`: format to jdn (Janet data notation).\n"
               "\n"
               "The following conversion specifiers are used for \"pretty-printing\", where the upper-case "
-              "variants generate colored output. These speficiers can take a precision "
+              "variants generate colored output. These specifiers can take a precision "
               "argument to specify the maximum nesting depth to print.\n"
               "- `p`, `P`: pretty format, truncating if necessary\n"
               "- `m`, `M`: pretty format without truncating.\n"

src/core/table.c

@@ -87,11 +87,27 @@ void janet_table_deinit(JanetTable *table) {
 }
 
 /* Create a new table */
+
 JanetTable *janet_table(int32_t capacity) {
     JanetTable *table = janet_gcalloc(JANET_MEMORY_TABLE, sizeof(JanetTable));
     return janet_table_init_impl(table, capacity, 0);
 }
 
+JanetTable *janet_table_weakk(int32_t capacity) {
+    JanetTable *table = janet_gcalloc(JANET_MEMORY_TABLE_WEAKK, sizeof(JanetTable));
+    return janet_table_init_impl(table, capacity, 0);
+}
+
+JanetTable *janet_table_weakv(int32_t capacity) {
+    JanetTable *table = janet_gcalloc(JANET_MEMORY_TABLE_WEAKV, sizeof(JanetTable));
+    return janet_table_init_impl(table, capacity, 0);
+}
+
+JanetTable *janet_table_weakkv(int32_t capacity) {
+    JanetTable *table = janet_gcalloc(JANET_MEMORY_TABLE_WEAKKV, sizeof(JanetTable));
+    return janet_table_init_impl(table, capacity, 0);
+}
+
 /* Find the bucket that contains the given key. Will also return
  * bucket where key should go if not in the table. */
 JanetKV *janet_table_find(JanetTable *t, Janet key) {
@@ -297,11 +313,46 @@ JANET_CORE_FN(cfun_table_new,
               "Creates a new empty table with pre-allocated memory "
               "for `capacity` entries. This means that if one knows the number of "
               "entries going into a table on creation, extra memory allocation "
-              "can be avoided. Returns the new table.") {
+              "can be avoided. "
+              "Returns the new table.") {
     janet_fixarity(argc, 1);
     int32_t cap = janet_getnat(argv, 0);
     return janet_wrap_table(janet_table(cap));
 }
+/*
+    uint32_t flags = janet_getflags(argv, 1, "kv");
+    if (flags == 0) return janet_wrap_table(janet_table(cap));
+    if (flags == 1) return janet_wrap_table(janet_table_weakk(cap));
+    if (flags == 2) return janet_wrap_table(janet_table_weakv(cap));
+    return janet_wrap_table(janet_table_weakkv(cap));
+    */
+
+JANET_CORE_FN(cfun_table_weak,
+              "(table/weak capacity)",
+              "Creates a new empty table with weak references to keys and values. Similar to `table/new`. "
+              "Returns the new table.") {
+    janet_fixarity(argc, 1);
+    int32_t cap = janet_getnat(argv, 0);
+    return janet_wrap_table(janet_table_weakkv(cap));
+}
+
+JANET_CORE_FN(cfun_table_weak_keys,
+              "(table/weak-keys capacity)",
+              "Creates a new empty table with weak references to keys and normal references to values. Similar to `table/new`. "
+              "Returns the new table.") {
+    janet_fixarity(argc, 1);
+    int32_t cap = janet_getnat(argv, 0);
+    return janet_wrap_table(janet_table_weakk(cap));
+}
+
+JANET_CORE_FN(cfun_table_weak_values,
+              "(table/weak-values capacity)",
+              "Creates a new empty table with normal references to keys and weak references to values. Similar to `table/new`. "
+              "Returns the new table.") {
+    janet_fixarity(argc, 1);
+    int32_t cap = janet_getnat(argv, 0);
+    return janet_wrap_table(janet_table_weakv(cap));
+}
 
 JANET_CORE_FN(cfun_table_getproto,
               "(table/getproto tab)",
@@ -376,6 +427,9 @@ JANET_CORE_FN(cfun_table_proto_flatten,
 void janet_lib_table(JanetTable *env) {
     JanetRegExt table_cfuns[] = {
         JANET_CORE_REG("table/new", cfun_table_new),
+        JANET_CORE_REG("table/weak", cfun_table_weak),
+        JANET_CORE_REG("table/weak-keys", cfun_table_weak_keys),
+        JANET_CORE_REG("table/weak-values", cfun_table_weak_values),
         JANET_CORE_REG("table/to-struct", cfun_table_tostruct),
         JANET_CORE_REG("table/getproto", cfun_table_getproto),
         JANET_CORE_REG("table/setproto", cfun_table_setproto),

src/core/vm.c

@@ -1585,6 +1585,7 @@ int janet_init(void) {
 
     /* Garbage collection */
     janet_vm.blocks = NULL;
+    janet_vm.weak_blocks = NULL;
     janet_vm.next_collection = 0;
     janet_vm.gc_interval = 0x400000;
     janet_vm.block_count = 0;

src/include/janet.h

@@ -234,6 +234,11 @@ extern "C" {
 #define JANET_EV_KQUEUE
 #endif
 
+/* Use poll as last resort */
+#if !defined(JANET_WINDOWS) && !defined(JANET_EV_EPOLL) && !defined(JANET_EV_KQUEUE)
+#define JANET_EV_POLL
+#endif
+
 /* How to export symbols */
 #ifndef JANET_EXPORT
 #ifdef JANET_WINDOWS
@@ -406,12 +411,11 @@ typedef enum {
     JANET_SIGNAL_USER6,
     JANET_SIGNAL_USER7,
     JANET_SIGNAL_USER8,
-    JANET_SIGNAL_USER9
+    JANET_SIGNAL_USER9,
+    JANET_SIGNAL_INTERRUPT = JANET_SIGNAL_USER8,
+    JANET_SIGNAL_EVENT = JANET_SIGNAL_USER9,
 } JanetSignal;
 
-#define JANET_SIGNAL_EVENT JANET_SIGNAL_USER9
-#define JANET_SIGNAL_INTERRUPT JANET_SIGNAL_USER8
-
 /* Fiber statuses - mostly corresponds to signals. */
 typedef enum {
     JANET_STATUS_DEAD,
@@ -575,7 +579,7 @@ typedef void *JanetAbstract;
 
 #define JANET_STREAM_CLOSED 0x1
 #define JANET_STREAM_SOCKET 0x2
-#define JANET_STREAM_IOCP 0x4
+#define JANET_STREAM_UNREGISTERED 0x4
 #define JANET_STREAM_READABLE 0x200
 #define JANET_STREAM_WRITABLE 0x400
 #define JANET_STREAM_ACCEPTABLE 0x800
@@ -583,55 +587,54 @@ typedef void *JanetAbstract;
 #define JANET_STREAM_TOCLOSE 0x10000
 
 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_COMPLETE, /* Used on windows for IOCP */
-    JANET_ASYNC_EVENT_USER
+    JANET_ASYNC_EVENT_INIT = 0,
+    JANET_ASYNC_EVENT_MARK = 1,
+    JANET_ASYNC_EVENT_DEINIT = 2,
+    JANET_ASYNC_EVENT_CLOSE = 3,
+    JANET_ASYNC_EVENT_ERR = 4,
+    JANET_ASYNC_EVENT_HUP = 5,
+    JANET_ASYNC_EVENT_READ = 6,
+    JANET_ASYNC_EVENT_WRITE = 7,
+    JANET_ASYNC_EVENT_COMPLETE = 8, /* Used on windows for IOCP */
+    JANET_ASYNC_EVENT_FAILED = 9, /* Used on windows for IOCP */
+    JANET_ASYNC_EVENT_USER = 10
 } JanetAsyncEvent;
 
-#define JANET_ASYNC_LISTEN_READ (1 << JANET_ASYNC_EVENT_READ)
-#define JANET_ASYNC_LISTEN_WRITE (1 << JANET_ASYNC_EVENT_WRITE)
-
 typedef enum {
-    JANET_ASYNC_STATUS_NOT_DONE,
-    JANET_ASYNC_STATUS_DONE
-} JanetAsyncStatus;
+    JANET_ASYNC_LISTEN_READ = 1,
+    JANET_ASYNC_LISTEN_WRITE,
+    JANET_ASYNC_LISTEN_BOTH
+} JanetAsyncMode;
 
 /* Typedefs */
-typedef struct JanetListenerState JanetListenerState;
 typedef struct JanetStream JanetStream;
-typedef JanetAsyncStatus(*JanetListener)(JanetListenerState *state, JanetAsyncEvent event);
+typedef void (*JanetEVCallback)(JanetFiber *fiber, JanetAsyncEvent event);
 
 /* Wrapper around file descriptors and HANDLEs that can be polled. */
 struct JanetStream {
     JanetHandle handle;
     uint32_t flags;
-    JanetListenerState *read_state;
-    JanetListenerState *write_state;
+    uint32_t index;
+    JanetFiber *read_fiber;
+    JanetFiber *write_fiber;
     const void *methods; /* Methods for this stream */
 };
 
-/* Interface for state machine based event loop */
-struct JanetListenerState {
-    JanetListener machine;
-    JanetFiber *fiber;
-    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;
+JANET_API void janet_async_end(JanetFiber *fiber);
+JANET_API void *janet_async_start(JanetFiber *fiber, JanetStream *stream,
+                                  JanetAsyncMode mode, JanetEVCallback callback, size_t data_size);
+
+#endif
+
+/* Janet uses atomic integers in several places for synchronization between threads and
+ * signals. Define them here */
 #ifdef JANET_WINDOWS
-    void *tag; /* Used to associate listeners with an overlapped structure */
-    int bytes; /* Used to track how many bytes were transfered. */
-#endif
-};
+typedef long JanetAtomicInt;
+#else
+typedef int32_t JanetAtomicInt;
 #endif
+JANET_API JanetAtomicInt janet_atomic_inc(JanetAtomicInt volatile *x);
+JANET_API JanetAtomicInt janet_atomic_dec(JanetAtomicInt volatile *x);
 
 /* We provide three possible implementations of Janets. The preferred
  * nanboxing approach, for 32 or 64 bits, and the standard C version. Code in the rest of the
@@ -896,7 +899,7 @@ struct JanetGCObject {
     int32_t flags;
     union {
         JanetGCObject *next;
-        int32_t refcount; /* For threaded abstract types */
+        volatile JanetAtomicInt refcount; /* For threaded abstract types */
     } data;
 };
 
@@ -920,7 +923,9 @@ struct JanetFiber {
      * 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. */
     uint32_t sched_id; /* Increment everytime fiber is scheduled by event loop */
-    JanetListenerState *waiting;
+    JanetEVCallback ev_callback; /* Call this before starting scheduled fibers */
+    JanetStream *ev_stream; /* which stream we are waiting on */
+    void *ev_state; /* Extra data for ev callback state. On windows, first element must be OVERLAPPED. */
     void *supervisor_channel; /* Channel to push self to when complete */
 #endif
 };
@@ -1397,9 +1402,6 @@ 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);
-
 /* Shorthand for yielding to event loop in C */
 JANET_NO_RETURN JANET_API void janet_await(void);
 JANET_NO_RETURN JANET_API void janet_sleep_await(double sec);
@@ -1590,6 +1592,7 @@ JANET_API double janet_rng_double(JanetRNG *rng);
 
 /* Array functions */
 JANET_API JanetArray *janet_array(int32_t capacity);
+JANET_API JanetArray *janet_array_weak(int32_t capacity);
 JANET_API JanetArray *janet_array_n(const Janet *elements, int32_t n);
 JANET_API void janet_array_ensure(JanetArray *array, int32_t capacity, int32_t growth);
 JANET_API void janet_array_setcount(JanetArray *array, int32_t count);

test/helper.janet

@@ -19,7 +19,7 @@
                               (frame :source) (frame :source-line)))
   (if x
     (when is-verbose (eprintf "\e[32m✔\e[0m %s: %s: %v" line-info (describe e) x))
-    (eprintf "\e[31m✘\e[0m %s: %s: %v" line-info (describe e) x))
+    (do (eprintf "\e[31m✘\e[0m %s: %s: %v" line-info (describe e) x) (eflush)))
   x)
 
 (defmacro assert-error
@@ -34,17 +34,17 @@
 
 (defmacro assert-no-error
   [msg & forms]
-  (def errsym (keyword (gensym)))
-  ~(assert (not= ,errsym (try (do ,;forms) ([_] ,errsym))) ,msg))
+  (def e (gensym))
+  (def f (gensym))
+  (if is-verbose
+  ~(try (do ,;forms (,assert true ,msg)) ([,e ,f] (,assert false ,msg) (,debug/stacktrace ,f ,e "\e[31m✘\e[0m ")))
+  ~(try (do ,;forms (,assert true ,msg)) ([_] (,assert false ,msg)))))
 
 (defn start-suite [&opt x]
   (default x (dyn :current-file))
   (set suite-name
        (cond
          (number? x) (string x)
-         (string? x) (string/slice x
-                                   (length "test/suite-")
-                                   (- (inc (length ".janet"))))
          (string x)))
   (set start-time (os/clock))
   (eprint "Starting suite " suite-name "..."))

test/suite-boot.janet

@@ -924,4 +924,31 @@
                        [:strict 3 4 "bar-oops"]])
         "maclintf 2")
 
+# Bad bytecode wrt. using result from break expression
+(defn bytecode-roundtrip
+  [f]
+  (assert-no-error "bytecode round-trip" (unmarshal (marshal f make-image-dict))))
+
+(defn case-1 [&] (def x (break 1)))
+(bytecode-roundtrip case-1)
+(defn foo [&])
+(defn case-2 [&]
+  (foo (break (foo)))
+  (foo))
+(bytecode-roundtrip case-2)
+(defn case-3 [&]
+  (def x (break (do (foo)))))
+(bytecode-roundtrip case-3)
+(defn case-4 [&]
+  (def x (break (break (foo)))))
+(bytecode-roundtrip case-4)
+(defn case-4 [&]
+  (def x (break (break (break)))))
+(bytecode-roundtrip case-4)
+
+# Debug bytecode of these functions
+# (pp (disasm case-1))
+# (pp (disasm case-2))
+# (pp (disasm case-3))
+
 (end-suite)

test/suite-ev.janet

@@ -23,7 +23,7 @@
 
 # Subprocess
 # 5e1a8c86f
-(def janet (dyn :executable))
+(def janet (dyn *executable*))
 
 # Subprocess should inherit the "RUN" parameter for fancy testing
 (def run (filter next (string/split " " (os/getenv "SUBRUN" ""))))
@@ -205,7 +205,8 @@
   (test-echo "world")
   (test-echo (string/repeat "abcd" 200))
 
-  (:close s))
+  (:close s)
+  (gccollect))
 
 # Test on both server and client
 # 504411e
@@ -344,5 +345,25 @@
 (ev/go |(ev/chan-close ch))
 (assert (= (ev/select [ch 1]) [:close ch]))
 
-(end-suite)
+# ev/gather check
+(defn exec-slurp
+  "Read stdout of subprocess and return it trimmed in a string."
+  [& args]
+  (def env (os/environ))
+  (put env :out :pipe)
+  (def proc (os/spawn args :epx env))
+  (def out (get proc :out))
+  (def buf @"")
+  (ev/gather
+    (:read out :all buf)
+    (:wait proc))
+  (string/trimr buf))
+(assert-no-error
+  "ev/with-deadline 1"
+  (assert (= "hi"
+             (ev/with-deadline
+               10
+               (exec-slurp ;run janet "-e" "(print :hi)")))
+          "exec-slurp 1"))
 
+(end-suite)

test/suite-inttypes.janet

@@ -126,7 +126,7 @@
   (assert (deep= (int/to-bytes (u64 300) :be buf2)
                  @"abcd\x00\x00\x00\x00\x00\x00\x01\x2C")))
 
-# int/s64 and int/u64 paramater type checking
+# int/s64 and int/u64 parameter type checking
 # 6aea7c7f7
 (assert-error
   "bad value passed to int/to-bytes"