Fix meson build.

Add bind option to net/connect

.github/workflows/release.yml

@@ -0,0 +1,35 @@
+name: Release
+
+on:
+  push:
+    tags:
+      - "v*.*.*"
+
+jobs:
+
+  release:
+    name: Build release binaries
+    runs-on: ${{ matrix.os }}
+    strategy:
+      matrix:
+        os: [ ubuntu-latest, macos-latest ]
+    steps:
+      - name: Checkout the repository
+        uses: actions/checkout@master
+      - name: Set the version
+        run: echo "version=${GITHUB_REF/refs\/tags\//}" >> $GITHUB_ENV
+      - name: Set the platform
+        run: echo "platform=$(tr '[A-Z]' '[a-z]' <<< $RUNNER_OS)" >> $GITHUB_ENV
+      - name: Compile the project
+        run: make clean && make
+      - name: Build the artifact
+        run: JANET_DIST_DIR=janet-${{ env.version }}-${{ env.platform }} make build/janet-${{ env.version }}-${{ env.platform }}-x64.tar.gz
+      - name: Draft the release
+        uses: softprops/action-gh-release@v1
+        with:
+          draft: true
+          files: |
+            build/*.gz
+            build/janet.h
+            build/c/janet.c
+            build/c/shell.c

.github/workflows/test.yml

@@ -0,0 +1,34 @@
+name: Test
+
+on: [push, pull_request]
+
+jobs:
+
+  test-posix:
+    name: Build and test on POSIX systems
+    runs-on: ${{ matrix.os }}
+    strategy:
+      matrix:
+        os: [ ubuntu-latest, macos-latest ]
+    steps:
+      - name: Checkout the repository
+        uses: actions/checkout@master
+      - name: Compile the project
+        run: make clean && make
+      - name: Test the project
+        run: make test
+
+  test-windows:
+    name: Build and test on Windows
+    runs-on: windows-latest
+    steps:
+      - name: Checkout the repository
+        uses: actions/checkout@master
+      - name: Setup MSVC
+        uses: ilammy/msvc-dev-cmd@v1
+      - name: Build the project
+        shell: cmd
+        run: build_win
+      - name: Test the project
+        shell: cmd
+        run: build_win test

CHANGELOG.md

@@ -1,6 +1,17 @@
 # Changelog
 All notable changes to this project will be documented in this file.
 
+## Unlreleased - ???
+- Supervisor channels in threads will no longer include a wastful copy of the fiber in every
+  message across a thread.
+- Allow passing a closure to `ev/thead` as well as a whole fiber.
+- Allow passing a closure directly to `ev/go` to spawn fibers on the event loop.
+
+## 1.17.1 - 2021-08-29
+- Fix docstring typos
+- Add `make install-jpm-git` to make jpm co-install simpler if using makefile.
+- Fix bugs with starting ev/threads and fiber marshling.
+
 ## 1.17.0 - 2021-08-21
 - Add the `-E` flag for one-liners with the `short-fn` syntax for argument passing.
 - Add support for threaded abstract types. Threaded abstract types can easily be shared between threads.

Makefile

@@ -281,6 +281,17 @@ install: $(JANET_TARGET) $(JANET_LIBRARY) $(JANET_STATIC_LIBRARY) build/janet.pc
 	cp build/janet.pc '$(DESTDIR)$(JANET_PKG_CONFIG_PATH)/janet.pc'
 	[ -z '$(DESTDIR)' ] && $(LDCONFIG) || true
 
+install-jpm-git: $(JANET_TARGET)
+	mkdir -p build
+	rm -rf build/jpm
+	git clone --depth=1 https://github.com/janet-lang/jpm.git build/jpm
+	cd build/jpm && PREFIX='$(DESTDIR)$(PREFIX)' \
+		JANET_MANPATH='$(DESTDIR)$(JANET_MANPATH)' \
+		JANET_HEADERPATH='$(DESTDIR)$(INCLUDEDIR)/janet' \
+		JANET_BINPATH='$(DESTDIR)$(BINDIR)' \
+		JANET_LIBPATH='$(DESTDIR)$(LIBDIR)' \
+		../../$(JANET_TARGET) ./bootstrap.janet
+
 uninstall:
 	-rm '$(DESTDIR)$(BINDIR)/janet'
 	-rm -rf '$(DESTDIR)$(INCLUDEDIR)/janet'

README.md

@@ -3,6 +3,7 @@
 [![Appveyor Status](https://ci.appveyor.com/api/projects/status/bjraxrxexmt3sxyv/branch/master?svg=true)](https://ci.appveyor.com/project/bakpakin/janet/branch/master)
 [![builds.sr.ht status](https://builds.sr.ht/~bakpakin/janet/commits/freebsd.yml.svg)](https://builds.sr.ht/~bakpakin/janet/commits/freebsd.yml?)
 [![builds.sr.ht status](https://builds.sr.ht/~bakpakin/janet/commits/openbsd.yml.svg)](https://builds.sr.ht/~bakpakin/janet/commits/openbsd.yml?)
+[![Actions Status](https://github.com/janet-lang/janet/actions/workflows/test.yml/badge.svg)](https://github.com/janet-lang/janet/actions/workflows/test.yml)
 
 <img src="https://raw.githubusercontent.com/janet-lang/janet/master/assets/janet-w200.png" alt="Janet logo" width=200 align="left">
 

examples/evsleep.janet

@@ -10,3 +10,13 @@
 (ev/call worker :b 5)
 (ev/sleep 0.3)
 (ev/call worker :c 12)
+
+(defn worker2
+  [name]
+  (repeat 10
+    (ev/sleep 0.2)
+    (print name " working")))
+
+(ev/go worker2 :bob)
+(ev/go worker2 :joe)
+(ev/go worker2 :sally)

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.17.0')
+  version : '1.17.1')
 
 # Global settings
 janet_path = join_paths(get_option('prefix'), get_option('libdir'), 'janet')
@@ -73,6 +73,7 @@ conf.set('JANET_NO_REALPATH', not get_option('realpath'))
 conf.set('JANET_NO_PROCESSES', not get_option('processes'))
 conf.set('JANET_SIMPLE_GETLINE', get_option('simple_getline'))
 conf.set('JANET_EV_NO_EPOLL', not get_option('epoll'))
+conf.set('JANET_EV_NO_KQUEUE', not get_option('kqueue'))
 conf.set('JANET_NO_THREADS', get_option('threads'))
 conf.set('JANET_NO_INTERPRETER_INTERRUPT', not get_option('interpreter_interrupt'))
 if get_option('os_name') != ''
@@ -171,7 +172,7 @@ janetc = custom_target('janetc',
   capture : true,
   command : [
     janet_boot, meson.current_source_dir(),
-    'JANET_PATH', janet_path, 'JANET_HEADERPATH', header_path
+    'JANET_PATH', janet_path
   ])
 
 janet_dependencies = [m_dep, dl_dep]

meson_options.txt

@@ -18,6 +18,7 @@ option('umask', type : 'boolean', value : true)
 option('realpath', type : 'boolean', value : true)
 option('simple_getline', type : 'boolean', value : false)
 option('epoll', type : 'boolean', value : false)
+option('kqueue', type : 'boolean', value : false)
 option('interpreter_interrupt', type : 'boolean', value : false)
 
 option('recursion_guard', type : 'integer', min : 10, max : 8000, value : 1024)

src/boot/boot.janet

@@ -1669,7 +1669,7 @@
   * tuple -- a tuple pattern will match if its first element matches, and the
     following elements are treated as predicates and are true.
 
-  * `_` symbol -- the last special case is the `_` symbol, which is a wildcard
+  * `\_` symbol -- the last special case is the `\_` symbol, which is a wildcard
     that will match any value without creating a binding.
 
   While a symbol pattern will ordinarily match any value, the pattern `(@ <sym>)`,
@@ -2124,11 +2124,10 @@
 ###
 ###
 
-# Initialize syspath and header path
+# Initialize syspath
 (each [k v] (partition 2 (tuple/slice boot/args 2))
   (case k
-    "JANET_PATH" (setdyn :syspath v)
-    "JANET_HEADERPATH" (setdyn :headerpath v)))
+    "JANET_PATH" (setdyn :syspath v)))
 
 (defn make-env
   `Create a new environment table. The new environment
@@ -3361,23 +3360,23 @@
     Returns a fiber that is scheduled to run the function.
     ```
     [f & args]
-    (ev/go (fiber/new (fn [&] (f ;args)) :tp)))
+    (ev/go (fn _call [&] (f ;args))))
 
   (defmacro ev/spawn
     "Run some code in a new fiber. This is shorthand for (ev/call (fn [] ;body))."
     [& body]
-    ~(,ev/go (fiber/new (fn _spawn [&] ,;body) :tp)))
+    ~(,ev/go (fn _spawn [&] ,;body)))
 
   (defmacro ev/do-thread
     ``Run some code in a new thread. Suspends the current fiber until the thread is complete, and
     evaluates to nil.``
     [& body]
-    ~(,ev/thread (fiber/new (fn _thread [&] ,;body) :t)))
+    ~(,ev/thread (fn _do-thread [&] ,;body)))
 
   (defmacro ev/spawn-thread
-    ``Run some code in a new thread. Like `ev/do-thread`, but returns immediately with a fiber.``
+    ``Run some code in a new thread. Like `ev/do-thread`, but returns nil immediately.``
     [& body]
-    ~(,ev/thread (fiber/new (fn _thread [&] ,;body) :t) nil :n))
+    ~(,ev/thread (fn _spawn-thread [&] ,;body) nil :n))
 
   (defmacro ev/with-deadline
     `Run a body of code with a deadline, such that if the code does not complete before
@@ -3408,7 +3407,7 @@
          (def ,res @[])
          (,wait-for-fibers ,chan
            ,(seq [[i body] :pairs bodies]
-              ~(,ev/go (,fiber/new (fn [] (put ,res ,i ,body)) :tp) nil ,chan)))
+              ~(,ev/go (fn [] (put ,res ,i ,body)) nil ,chan)))
          ,res))))
 
 (compwhen (dyn 'net/listen)
@@ -3521,7 +3520,6 @@
   (var *error-level* nil)
 
   (if-let [jp (getenv-alias "JANET_PATH")] (setdyn :syspath jp))
-  (if-let [jp (getenv-alias "JANET_HEADERPATH")] (setdyn :headerpath jp))
   (if-let [jprofile (getenv-alias "JANET_PROFILE")] (setdyn :profilepath jprofile))
 
   (defn- get-lint-level

src/conf/janetconf.h

@@ -5,9 +5,9 @@
 
 #define JANET_VERSION_MAJOR 1
 #define JANET_VERSION_MINOR 17
-#define JANET_VERSION_PATCH 0
+#define JANET_VERSION_PATCH 1
 #define JANET_VERSION_EXTRA ""
-#define JANET_VERSION "1.17.0"
+#define JANET_VERSION "1.17.1"
 
 /* #define JANET_BUILD "local" */
 
@@ -48,6 +48,7 @@
 /* #define JANET_OS_NAME my-custom-os */
 /* #define JANET_ARCH_NAME pdp-8 */
 /* #define JANET_EV_NO_EPOLL */
+/* #define JANET_EV_NO_KQUEUE */
 /* #define JANET_NO_INTERPRETER_INTERRUPT */
 
 /* Custom vm allocator support */

src/core/ev.c

@@ -43,6 +43,9 @@
 #include <sys/epoll.h>
 #include <sys/timerfd.h>
 #endif
+#ifdef JANET_EV_KQUEUE
+#include <sys/event.h>
+#endif
 #endif
 
 typedef struct {
@@ -510,10 +513,10 @@ void janet_ev_mark(void) {
 static int janet_channel_push(JanetChannel *channel, Janet x, int mode);
 static int janet_channel_pop(JanetChannel *channel, Janet *item, int is_choice);
 
-static Janet make_supervisor_event(const char *name, JanetFiber *fiber) {
+static Janet make_supervisor_event(const char *name, JanetFiber *fiber, int threaded) {
     Janet tup[2];
     tup[0] = janet_ckeywordv(name);
-    tup[1] = janet_wrap_fiber(fiber);
+    tup[1] = threaded ? fiber->last_value : janet_wrap_fiber(fiber) ;
     return janet_wrap_tuple(janet_tuple_n(tup, 2));
 }
 
@@ -943,7 +946,7 @@ JANET_CORE_FN(cfun_channel_pop,
 JANET_CORE_FN(cfun_channel_choice,
               "(ev/select & clauses)",
               "Block until the first of several channel operations occur. Returns a tuple of the form [:give chan], [:take chan x], or [:close chan], where "
-              "a :give tuple is the result of a write and :take tuple is the result of a write. Each clause must be either a channel (for "
+              "a :give tuple is the result of a write and :take tuple is the result of a read. Each clause must be either a channel (for "
               "a channel take operation) or a tuple [channel x] for a channel give operation. Operations are tried in order, such that the first "
               "clauses will take precedence over later clauses. Both and give and take operations can return a [:close chan] tuple, which indicates that "
               "the specified channel was closed while waiting, or that the channel was already closed.") {
@@ -1209,13 +1212,17 @@ JanetFiber *janet_loop1(void) {
         Janet res;
         JanetSignal sig = janet_continue_signal(task.fiber, task.value, &res, task.sig);
         void *sv = task.fiber->supervisor_channel;
+        int is_suspended = sig == JANET_SIGNAL_EVENT || sig == JANET_SIGNAL_YIELD || sig == JANET_SIGNAL_INTERRUPT;
         if (NULL == sv) {
-            if (sig != JANET_SIGNAL_EVENT && sig != JANET_SIGNAL_YIELD && sig != JANET_SIGNAL_INTERRUPT) {
+            if (!is_suspended) {
                 janet_stacktrace(task.fiber, res);
             }
         } else if (sig == JANET_SIGNAL_OK || (task.fiber->flags & (1 << sig))) {
             JanetChannel *chan = janet_channel_unwrap(sv);
-            janet_channel_push(chan, make_supervisor_event(janet_signal_names[sig], task.fiber), 2);
+            janet_channel_push(chan, make_supervisor_event(janet_signal_names[sig],
+                               task.fiber, chan->is_threaded), 2);
+        } else if (!is_suspended) {
+            janet_stacktrace(task.fiber, res);
         }
         if (sig == JANET_SIGNAL_INTERRUPT) {
             /* On interrupts, return the interrupted fiber immediately */
@@ -1563,6 +1570,208 @@ void janet_ev_deinit(void) {
  * End epoll implementation
  */
 
+#elif defined(JANET_EV_KQUEUE)
+/* Definition from:
+ *   https://github.com/wahern/cqueues/blob/master/src/lib/kpoll.c
+ * NetBSD uses intptr_t while others use void * for .udata */
+#define EV_SETx(ev, a, b, c, d, e, f) EV_SET((ev), (a), (b), (c), (d), (e), ((__typeof__((ev)->udata))(f)))
+#define JANET_KQUEUE_TF (EV_ADD | EV_ENABLE | EV_CLEAR | EV_ONESHOT)
+
+/* NOTE:
+ * NetBSD doesn't like intervals less than 1 millisecond so lets make that the
+ * default anywhere JANET_KQUEUE_TS will be used. */
+#ifdef __NetBSD__
+#define JANET_KQUEUE_MIN_INTERVAL 1
+#else
+#define JANET_KQUEUE_MIN_INTERVAL 0
+#endif
+
+#ifdef __FreeBSD__
+#define JANET_KQUEUE_TS(timestamp) (timestamp)
+#else
+/* NOTE:
+ * NetBSD and OpenBSD expect things are always intervals, so fake that we have
+ * abstime capability by changing how a timestamp is used in all kqueue calls
+ * and defining absent macros. Additionally NetBSD expects intervals be
+ * greater than 1 millisecond, so correct all intervals to be at least 1
+ * millisecond under NetBSD. */
+JanetTimestamp fix_interval(const JanetTimestamp ts) {
+    return ts >= JANET_KQUEUE_MIN_INTERVAL ? ts : JANET_KQUEUE_MIN_INTERVAL;
+}
+#define JANET_KQUEUE_TS(timestamp) (fix_interval((timestamp - ts_now())))
+#define NOTE_MSECONDS 0
+#define NOTE_ABSTIME 0
+#endif
+
+
+/* TODO: make this available be we using kqueue or epoll, instead of
+ * redefinining it for kqueue and epoll separately? */
+static JanetTimestamp ts_now(void) {
+    struct timespec now;
+    janet_assert(-1 != clock_gettime(CLOCK_MONOTONIC, &now), "failed to get time");
+    uint64_t res = 1000 * now.tv_sec;
+    res += now.tv_nsec / 1000000;
+    return res;
+}
+
+void add_kqueue_events(const struct kevent *events, int length) {
+    /* NOTE: Status should be equal to the amount of events added, which isn't
+     * always known since deletions or modifications occur. Can't use the
+     * eventlist argument for it to report to us what failed otherwise we may
+     * poll in events to handle! This code assumes atomicity, that kqueue can
+     * either succeed or fail, but never partially (which is seemingly how it
+     * works in practice). When encountering an "inbetween" state we currently
+     * just panic!
+     *
+     * The FreeBSD man page kqueue(2) shows a check through the change list to
+     * check if kqueue had an error with any of the events being pushed to
+     * change. Maybe we should do this, even tho the man page also doesn't
+     * note that kqueue actually does this. We do not do this at this time.  */
+    int status;
+    status = kevent(janet_vm.kq, events, length, NULL, 0, NULL);
+    if (status == -1 && errno != EINTR)
+        janet_panicv(janet_ev_lasterr());
+}
+
+JanetListenerState *janet_listen(JanetStream *stream, JanetListener behavior, int mask, size_t size, void *user) {
+    JanetListenerState *state = janet_listen_impl(stream, behavior, mask, size, user);
+    struct kevent kev[2];
+
+    int length = 0;
+    if (state->stream->_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) {
+        EV_SETx(&kev[length], stream->handle, EVFILT_WRITE, EV_ADD | EV_ENABLE, 0, 0, stream);
+        length++;
+    }
+
+    if (length > 0) {
+        add_kqueue_events(kev, length);
+    }
+
+    return state;
+}
+
+static void janet_unlisten(JanetListenerState *state, int is_gc) {
+    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);
+
+            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);
+        }
+    }
+    janet_unlisten_impl(state, is_gc);
+}
+
+#define JANET_KQUEUE_TIMER_IDENT 1
+#define JANET_KQUEUE_MAX_EVENTS 64
+
+void janet_loop1_impl(int has_timeout, JanetTimestamp timeout) {
+    /* Construct our timer which is a definite time on the clock, not an
+     * interval, in milliseconds as that is `JanetTimestamp`'s precision. */
+    struct kevent timer;
+    if (janet_vm.timer_enabled || has_timeout) {
+        EV_SETx(&timer,
+               JANET_KQUEUE_TIMER_IDENT,
+               EVFILT_TIMER,
+               JANET_KQUEUE_TF,
+               NOTE_MSECONDS | NOTE_ABSTIME,
+               JANET_KQUEUE_TS(timeout), &janet_vm.timer);
+        add_kqueue_events(&timer, 1);
+    }
+    janet_vm.timer_enabled = has_timeout;
+
+    /* Poll for events */
+    int status;
+    struct kevent events[JANET_KQUEUE_MAX_EVENTS];
+    do {
+        status = kevent(janet_vm.kq, NULL, 0, events, JANET_KQUEUE_MAX_EVENTS, NULL);
+    } while (status == -1 && errno == EINTR);
+    if (status == -1)
+        JANET_EXIT("failed to poll events");
+
+    /* Step state machines */
+    for (int i = 0; i < status; i++) {
+        void *p = (void*) events[i].udata;
+        if (&janet_vm.timer == p) {
+            /* Timer expired, ignore */;
+        } else if (janet_vm.selfpipe == p) {
+            /* Self-pipe handling */
+            janet_ev_handle_selfpipe();
+        } else {
+            JanetStream *stream = p;
+            JanetListenerState *state = stream->state;
+            if (NULL != state) {
+                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);
+                    if (events[i].filter == EVFILT_READ)
+                        statuses[1] = state->machine(state, JANET_ASYNC_EVENT_READ);
+                    if ((events[i].flags & EV_EOF) && !(events[i].data > 0))
+                        statuses[3] = state->machine(state, JANET_ASYNC_EVENT_HUP);
+                } else {
+                    statuses[2] = state->machine(state, JANET_ASYNC_EVENT_ERR);
+                }
+                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);
+            }
+        }
+    }
+}
+
+void janet_ev_init(void) {
+    janet_ev_init_common();
+    janet_ev_setup_selfpipe();
+    janet_vm.kq = kqueue();
+    janet_vm.timer_enabled = 0;
+    if (janet_vm.kq == -1) goto error;
+    struct kevent events[2];
+    /* Don't use JANET_KQUEUE_TS here, as even under FreeBSD we use intervals
+     * here. */
+    EV_SETx(&events[0],
+            JANET_KQUEUE_TIMER_IDENT,
+            EVFILT_TIMER,
+            JANET_KQUEUE_TF,
+            NOTE_MSECONDS, JANET_KQUEUE_MIN_INTERVAL, &janet_vm.timer);
+    EV_SETx(&events[1], janet_vm.selfpipe[0], EVFILT_READ, EV_ADD | EV_ENABLE, 0, 0, janet_vm.selfpipe);
+    add_kqueue_events(events, 2);
+    return;
+error:
+    JANET_EXIT("failed to initialize event loop");
+}
+
+void janet_ev_deinit(void) {
+    janet_ev_deinit_common();
+    close(janet_vm.kq);
+    janet_ev_cleanup_selfpipe();
+    janet_vm.kq = 0;
+}
+
 #else
 
 #include <poll.h>
@@ -2438,16 +2647,38 @@ JANET_CORE_FN(cfun_ev_go,
               "(ev/go fiber &opt value supervisor)",
               "Put a fiber on the event loop to be resumed later. Optionally pass "
               "a value to resume with, otherwise resumes with nil. Returns the fiber. "
-              "An optional `core/channel` can be provided as well as a supervisor. When various "
+              "An optional `core/channel` can be provided as a supervisor. When various "
               "events occur in the newly scheduled fiber, an event will be pushed to the supervisor. "
               "If not provided, the new fiber will inherit the current supervisor.") {
     janet_arity(argc, 1, 3);
-    JanetFiber *fiber = janet_getfiber(argv, 0);
     Janet value = argc >= 2 ? argv[1] : janet_wrap_nil();
     void *supervisor = janet_optabstract(argv, argc, 2, &janet_channel_type, janet_vm.root_fiber->supervisor_channel);
+    JanetFiber *fiber;
+    if (janet_checktype(argv[0], JANET_FUNCTION)) {
+        /* Create a fiber for the user */
+        JanetFunction *func = janet_unwrap_function(argv[0]);
+        if (func->def->min_arity > 1) {
+            janet_panicf("task function must accept 0 or 1 arguments");
+        }
+        fiber = janet_fiber(func, 64, func->def->min_arity, &value);
+        fiber->flags |=
+            JANET_FIBER_MASK_ERROR |
+            JANET_FIBER_MASK_USER0 |
+            JANET_FIBER_MASK_USER1 |
+            JANET_FIBER_MASK_USER2 |
+            JANET_FIBER_MASK_USER3 |
+            JANET_FIBER_MASK_USER4;
+        if (!janet_vm.fiber->env) {
+            janet_vm.fiber->env = janet_table(0);
+        }
+        fiber->env = janet_table(0);
+        fiber->env->proto = janet_vm.fiber->env;
+    } else {
+        fiber = janet_getfiber(argv, 0);
+    }
     fiber->supervisor_channel = supervisor;
     janet_schedule(fiber, value);
-    return argv[0];
+    return janet_wrap_fiber(fiber);
 }
 
 /* For ev/thread - Run an interpreter in the new thread. */
@@ -2472,12 +2703,12 @@ static JanetEVGenericMessage janet_go_thread_subr(JanetEVGenericMessage args) {
         }
 
         /* Get supervsior */
-        void *supervisor = NULL;
         if (flags & 0x8) {
             Janet sup =
                 janet_unmarshal(nextbytes, endbytes - nextbytes,
                                 JANET_MARSHAL_UNSAFE, NULL, &nextbytes);
-            supervisor = janet_unwrap_pointer(sup);
+            /* Hack - use a global variable to avoid longjmp clobber */
+            janet_vm.user = janet_unwrap_pointer(sup);
         }
 
         /* Set cfunction registry */
@@ -2504,19 +2735,52 @@ static JanetEVGenericMessage janet_go_thread_subr(JanetEVGenericMessage args) {
                                        JANET_MARSHAL_UNSAFE, NULL, &nextbytes);
         Janet value = janet_unmarshal(nextbytes, endbytes - nextbytes,
                                       JANET_MARSHAL_UNSAFE, NULL, &nextbytes);
-        if (!janet_checktype(fiberv, JANET_FIBER)) janet_panicf("expected fiber, got %v", fiberv);
-        JanetFiber *fiber = janet_unwrap_fiber(fiberv);
-        fiber->supervisor_channel = supervisor;
+        JanetFiber *fiber;
+        if (!janet_checktype(fiberv, JANET_FIBER)) {
+            if (!janet_checktype(fiberv, JANET_FUNCTION)) {
+                janet_panicf("expected function|fiber, got %v", fiberv);
+            }
+            JanetFunction *func = janet_unwrap_function(fiberv);
+            if (func->def->min_arity > 1) {
+                janet_panicf("thread function must accept 0 or 1 arguments");
+            }
+            fiber = janet_fiber(func, 64, func->def->min_arity, &value);
+            fiber->flags |=
+                JANET_FIBER_MASK_ERROR |
+                JANET_FIBER_MASK_USER0 |
+                JANET_FIBER_MASK_USER1 |
+                JANET_FIBER_MASK_USER2 |
+                JANET_FIBER_MASK_USER3 |
+                JANET_FIBER_MASK_USER4;
+        } else {
+            fiber = janet_unwrap_fiber(fiberv);
+        }
+        fiber->supervisor_channel = janet_vm.user;
         janet_schedule(fiber, value);
         janet_loop();
         args.tag = JANET_EV_TCTAG_NIL;
     } else {
-        if (janet_checktype(tstate.payload, JANET_STRING)) {
-            args.tag = JANET_EV_TCTAG_ERR_STRINGF;
-            args.argp = strdup((const char *) janet_unwrap_string(tstate.payload));
+        void *supervisor = janet_vm.user;
+        if (NULL != supervisor) {
+            /* Got a supervisor, write error there */
+            Janet pair[] = {
+                janet_ckeywordv("error"),
+                tstate.payload
+            };
+            janet_channel_push((JanetChannel *)supervisor,
+                               janet_wrap_tuple(janet_tuple_n(pair, 2)), 2);
+        } else if (flags & 0x1) {
+            /* No wait, just print to stderr */
+            janet_eprintf("thread start failure: %v\n", tstate.payload);
         } else {
-            args.tag = JANET_EV_TCTAG_ERR_STRING;
-            args.argp = "failed to start thread";
+            /* Make ev/thread call from parent thread error */
+            if (janet_checktype(tstate.payload, JANET_STRING)) {
+                args.tag = JANET_EV_TCTAG_ERR_STRINGF;
+                args.argp = strdup((const char *) janet_unwrap_string(tstate.payload));
+            } else {
+                args.tag = JANET_EV_TCTAG_ERR_STRING;
+                args.argp = "failed to start thread";
+            }
         }
     }
     janet_restore(&tstate);
@@ -2527,18 +2791,19 @@ static JanetEVGenericMessage janet_go_thread_subr(JanetEVGenericMessage args) {
 }
 
 JANET_CORE_FN(cfun_ev_thread,
-              "(ev/thread fiber &opt value flags supervisor)",
-              "Resume a (copy of a) `fiber` in a new operating system thread, optionally passing `value` "
-              "to resume with. "
+              "(ev/thread main &opt value flags supervisor)",
+              "Run `main` in a new operating system thread, optionally passing `value` "
+              "to resume with. The parameter `main` can either be a fiber, or a function that accepts "
+              "0 or 1 arguments. "
               "Unlike `ev/go`, this function will suspend the current fiber until the thread is complete. "
               "If you want to run the thread without waiting for a result, pass the `:n` flag to return nil immediately. "
-              "Otherwise, returns (a copy of) the final result from the fiber on the new thread. Available flags:\n\n"
+              "Otherwise, returns nil. Available flags:\n\n"
               "* `:n` - return immediately\n"
               "* `:a` - don't copy abstract registry to new thread (performance optimization)\n"
               "* `:c` - don't copy cfunction registry to new thread (performance optimization)") {
     janet_arity(argc, 1, 4);
-    janet_getfiber(argv, 0);
     Janet value = argc >= 2 ? argv[1] : janet_wrap_nil();
+    if (!janet_checktype(argv[0], JANET_FUNCTION)) janet_getfiber(argv, 0);
     uint64_t flags = 0;
     if (argc >= 3) {
         flags = janet_getflags(argv, 2, "nac");
@@ -2559,7 +2824,7 @@ JANET_CORE_FN(cfun_ev_thread,
         janet_marshal(buffer, janet_wrap_abstract(supervisor), NULL, JANET_MARSHAL_UNSAFE);
     }
     if (!(flags & 0x4)) {
-        janet_assert(janet_vm.registry_count <= UINT32_MAX, "assert failed size check");
+        janet_assert(janet_vm.registry_count <= INT32_MAX, "assert failed size check");
         uint32_t temp = (uint32_t) janet_vm.registry_count;
         janet_buffer_push_bytes(buffer, (uint8_t *) &temp, sizeof(temp));
         janet_buffer_push_bytes(buffer, (uint8_t *) janet_vm.registry, (int32_t) janet_vm.registry_count * sizeof(JanetCFunRegistry));

src/core/fiber.c

@@ -642,7 +642,7 @@ JANET_CORE_FN(cfun_fiber_can_resume,
 }
 
 JANET_CORE_FN(cfun_fiber_last_value,
-              "(fiber/last-value",
+              "(fiber/last-value)",
               "Get the last value returned or signaled from the fiber.") {
     janet_fixarity(argc, 1);
     JanetFiber *fiber = janet_getfiber(argv, 0);

src/core/marsh.c

@@ -560,9 +560,9 @@ static void marshal_one(MarshalState *st, Janet x, int flags) {
         case JANET_FUNCTION: {
             pushbyte(st, LB_FUNCTION);
             JanetFunction *func = janet_unwrap_function(x);
+            pushint(st, func->def->environments_length);
             /* Mark seen before reading def */
             MARK_SEEN();
-            pushint(st, func->def->environments_length);
             marshal_one_def(st, func->def, flags);
             for (int32_t i = 0; i < func->def->environments_length; i++)
                 marshal_one_env(st, func->envs[i], flags + 1);
@@ -953,6 +953,7 @@ static const uint8_t *unmarshal_one_fiber(
     fiber->data = NULL;
     fiber->child = NULL;
     fiber->env = NULL;
+    fiber->last_value = janet_wrap_nil();
 #ifdef JANET_EV
     fiber->waiting = NULL;
     fiber->sched_id = 0;
@@ -1257,18 +1258,16 @@ static const uint8_t *unmarshal_one(
             data++;
             int32_t len = readnat(st, &data);
             if (len > 255) {
-                janet_panicf("invalid function");
+                janet_panicf("invalid function - too many environments (%d)", len);
             }
             func = janet_gcalloc(JANET_MEMORY_FUNCTION, sizeof(JanetFunction) +
                                  len * sizeof(JanetFuncEnv));
+            func->def = NULL;
             *out = janet_wrap_function(func);
             janet_v_push(st->lookup, *out);
             data = unmarshal_one_def(st, data, &def, flags + 1);
-            if (def->environments_length != len) {
-                janet_panicf("invalid function");
-            }
             func->def = def;
-            for (int32_t i = 0; i < def->environments_length; i++) {
+            for (int32_t i = 0; i < len; i++) {
                 data = unmarshal_one_env(st, data, &(func->envs[i]), flags + 1);
             }
             return data;

src/core/net.c

@@ -259,7 +259,8 @@ static int janet_get_sockettype(Janet *argv, int32_t argc, int32_t n) {
 }
 
 /* Needs argc >= offset + 2 */
-/* For unix paths, just rertuns a single sockaddr and sets *is_unix to 1, otherwise 0 */
+/* For unix paths, just rertuns a single sockaddr and sets *is_unix to 1,
+ * otherwise 0. Also, ignores is_bind when is a unix socket. */
 static struct addrinfo *janet_get_addrinfo(Janet *argv, int32_t offset, int socktype, int passive, int *is_unix) {
     /* Unix socket support - not yet supported on windows. */
 #ifndef JANET_WINDOWS
@@ -285,12 +286,12 @@ static struct addrinfo *janet_get_addrinfo(Janet *argv, int32_t offset, int sock
     }
 #endif
     /* Get host and port */
-    const char *host = janet_getcstring(argv, offset);
-    const char *port;
+    char *host = (char *)janet_getcstring(argv, offset);
+    char *port = NULL;
     if (janet_checkint(argv[offset + 1])) {
-        port = (const char *)janet_to_string(argv[offset + 1]);
+        port = (char *)janet_to_string(argv[offset + 1]);
     } else {
-        port = janet_optcstring(argv, offset + 2, offset + 1, NULL);
+        port = (char *)janet_optcstring(argv, offset + 2, offset + 1, NULL);
     }
     /* getaddrinfo */
     struct addrinfo *ai = NULL;
@@ -357,16 +358,48 @@ JANET_CORE_FN(cfun_net_sockaddr,
 }
 
 JANET_CORE_FN(cfun_net_connect,
-              "(net/connect host port &opt type)",
+              "(net/connect host port &opt type bindhost bindport)",
               "Open a connection to communicate with a server. Returns a duplex stream "
               "that can be used to communicate with the server. Type is an optional keyword "
-              "to specify a connection type, either :stream or :datagram. The default is :stream. ") {
-    janet_arity(argc, 2, 3);
+              "to specify a connection type, either :stream or :datagram. The default is :stream. "
+              "Bindhost is an optional string to select from what address to make the outgoing "
+              "connection, with the default being the same as using the OS's preferred address. ") {
+    janet_arity(argc, 2, 5);
 
+    /* Check arguments */
     int socktype = janet_get_sockettype(argv, argc, 2);
     int is_unix = 0;
+    char *bindhost = (char *) janet_optcstring(argv, argc, 3, NULL);
+    char *bindport = NULL;
+    if (janet_checkint(argv[4])) {
+        bindport = (char *)janet_to_string(argv[4]);
+    } else {
+        bindport = (char *)janet_optcstring(argv, argc, 4, NULL);
+    }
+
+    /* Where we're connecting to */
     struct addrinfo *ai = janet_get_addrinfo(argv, 0, socktype, 0, &is_unix);
 
+    /* Check if we're binding address */
+    struct addrinfo *binding = NULL;
+    if (bindhost != NULL) {
+        if (is_unix) {
+            freeaddrinfo(ai);
+            janet_panic("bindhost not supported for unix domain sockets");
+        }
+        /* getaddrinfo */
+        struct addrinfo hints;
+        memset(&hints, 0, sizeof(hints));
+        hints.ai_family = AF_UNSPEC;
+        hints.ai_socktype = socktype;
+        hints.ai_flags = 0;
+        int status = getaddrinfo(bindhost, bindport, &hints, &binding);
+        if (status) {
+            freeaddrinfo(ai);
+            janet_panicf("could not get address info for bindhost: %s", gai_strerror(status));
+        }
+    }
+
     /* Create socket */
     JSock sock = JSOCKDEFAULT;
     void *addr = NULL;
@@ -375,7 +408,9 @@ JANET_CORE_FN(cfun_net_connect,
     if (is_unix) {
         sock = socket(AF_UNIX, socktype | JSOCKFLAGS, 0);
         if (!JSOCKVALID(sock)) {
-            janet_panicf("could not create socket: %V", janet_ev_lasterr());
+            Janet v = janet_ev_lasterr();
+            janet_free(ai);
+            janet_panicf("could not create socket: %V", v);
         }
         addr = (void *) ai;
         addrlen = sizeof(struct sockaddr_un);
@@ -396,17 +431,42 @@ JANET_CORE_FN(cfun_net_connect,
             }
         }
         if (NULL == addr) {
+            Janet v = janet_ev_lasterr();
+            if (binding) freeaddrinfo(binding);
             freeaddrinfo(ai);
-            janet_panicf("could not create socket: %V", janet_ev_lasterr());
+            janet_panicf("could not create socket: %V", v);
+        }
+    }
+
+    /* Bind to bindhost and bindport if given */
+    if (binding) {
+        struct addrinfo *rp = NULL;
+        int did_bind = 0;
+        for (rp = ai; rp != NULL; rp = rp->ai_next) {
+            if (bind(sock, rp->ai_addr, (int) rp->ai_addrlen) == 0) {
+                did_bind = 1;
+                break;
+            }
+        }
+        if (!did_bind) {
+            Janet v = janet_ev_lasterr();
+            freeaddrinfo(binding);
+            freeaddrinfo(ai);
+            JSOCKCLOSE(sock);
+            janet_panicf("could not bind outgoing address: %V", v);
+        } else {
+            freeaddrinfo(binding);
         }
     }
 
     /* Connect to socket */
 #ifdef JANET_WINDOWS
     int status = WSAConnect(sock, addr, addrlen, NULL, NULL, NULL, NULL);
+    Janet lasterr = janet_ev_lasterr();
     freeaddrinfo(ai);
 #else
     int status = connect(sock, addr, addrlen);
+    Janet lasterr = janet_ev_lasterr();
     if (is_unix) {
         janet_free(ai);
     } else {
@@ -416,7 +476,7 @@ JANET_CORE_FN(cfun_net_connect,
 
     if (status == -1) {
         JSOCKCLOSE(sock);
-        janet_panicf("could not connect to socket: %V", janet_ev_lasterr());
+        janet_panicf("could not connect socket: %V", lasterr);
     }
 
     /* Set up the socket for non-blocking IO after connect - TODO - non-blocking connect? */

src/core/os.c

@@ -158,7 +158,7 @@ JANET_CORE_FN(os_arch,
               "(os/arch)",
               "Check the ISA that janet was compiled for. Returns one of:\n\n"
               "* :x86\n\n"
-              "* :x86-64\n\n"
+              "* :x64\n\n"
               "* :arm\n\n"
               "* :aarch64\n\n"
               "* :sparc\n\n"

src/core/state.h

@@ -76,6 +76,9 @@ typedef struct {
 } JanetCFunRegistry;
 
 struct JanetVM {
+    /* Place for user data */
+    void *user;
+
     /* Top level dynamic bindings */
     JanetTable *top_dyns;
 
@@ -168,6 +171,11 @@ struct JanetVM {
     int epoll;
     int timerfd;
     int timer_enabled;
+#elif defined(JANET_EV_KQUEUE)
+    JanetHandle selfpipe[2];
+    int kq;
+    int timer;
+    int timer_enabled;
 #else
     JanetHandle selfpipe[2];
     struct pollfd *fds;

src/core/util.c

@@ -545,6 +545,11 @@ void janet_cfuns_ext_prefix(JanetTable *env, const char *regprefix, const JanetR
     if (env) namebuf_deinit(&nb);
 }
 
+/* Register a value in the global registry */
+void janet_register(const char *name, JanetCFunction cfun) {
+    janet_registry_put(cfun, name, NULL, NULL, 0);
+}
+
 /* Abstract type introspection */
 
 void janet_register_abstract_type(const JanetAbstractType *at) {

src/core/vm.c

@@ -1520,6 +1520,7 @@ int janet_init(void) {
     janet_vm.root_capacity = 0;
 
     /* Scratch memory */
+    janet_vm.user = NULL;
     janet_vm.scratch_mem = NULL;
     janet_vm.scratch_len = 0;
     janet_vm.scratch_cap = 0;
@@ -1579,6 +1580,7 @@ void janet_deinit(void) {
     janet_vm.abstract_registry = NULL;
     janet_vm.core_env = NULL;
     janet_vm.top_dyns = NULL;
+    janet_vm.user = NULL;
     janet_free(janet_vm.traversal_base);
     janet_vm.fiber = NULL;
     janet_vm.root_fiber = NULL;

src/include/janet.h

@@ -198,6 +198,16 @@ extern "C" {
 #define JANET_EV_EPOLL
 #endif
 
+/* Enable or disable kqueue on BSD */
+#if defined(JANET_BSD) && !defined(JANET_EV_NO_KQUEUE)
+#define JANET_EV_KQUEUE
+#endif
+
+/* Enable or disable kqueue on Apple */
+#if defined(JANET_APPLE) && !defined(JANET_EV_NO_KQUEUE)
+#define JANET_EV_KQUEUE
+#endif
+
 /* How to export symbols */
 #ifndef JANET_API
 #ifdef JANET_WINDOWS
@@ -1852,6 +1862,9 @@ JANET_API void janet_cfuns_ext_prefix(JanetTable *env, const char *regprefix, co
 JANET_API void janet_def_sm(JanetTable *env, const char *name, Janet val, const char *documentation, const char *source_file, int32_t source_line);
 JANET_API void janet_var_sm(JanetTable *env, const char *name, Janet val, const char *documentation, const char *source_file, int32_t source_line);
 
+/* Legacy definition of C functions */
+JANET_API void janet_register(const char *name, JanetCFunction cfun);
+
 /* Allow setting entry name for static libraries */
 #ifdef __cplusplus
 #define JANET_MODULE_PREFIX extern "C"