Update os/shell to be non-blocking as well.

Author: bakpakin

src/boot/boot.janet

@@ -3146,10 +3146,17 @@
 
 (compwhen (dyn 'ev/go)
   (defn net/close "Alias for ev/close." [stream] (ev/close stream))
+
+  (defn ev/call
+    "Call a function asynchronously. Returns a fiber that is scheduled to "
+    "run the function."
+    [f & args]
+    (ev/go (fiber/new (fn [&] (f ;args)) :tp)))
+
   (defmacro ev/spawn
     "Run some code in a new fiber. This is shorthand for (ev/call (fn [] ;body))."
     [& body]
-    ~(,ev/call (fn [] ,;body)))
+    ~(,ev/go (fiber/new (fn _spawn [&] ,;body) :tp)))
 
   (defmacro ev/with-deadline
     `Run a body of code with a deadline, such that if the code does not complete before

src/core/ev.c

@@ -1435,12 +1435,16 @@ void janet_ev_default_threaded_callback(JanetEVGenericMessage return_value) {
         case JANET_EV_TCTAG_ERR_KEYWORD:
             janet_cancel(return_value.fiber, janet_ckeywordv((const char *) return_value.argp));
             break;
+        case JANET_EV_TCTAG_BOOLEAN:
+            janet_schedule(return_value.fiber, janet_wrap_boolean(return_value.argi));
+            break;
     }
     janet_gcunroot(janet_wrap_fiber(return_value.fiber));
 }
 
 
 /* Convenience method for common case */
+JANET_NO_RETURN
 void janet_ev_threaded_await(JanetThreadedSubroutine fp, int tag, int argi, void *argp) {
     JanetEVGenericMessage arguments;
     arguments.tag = tag;
@@ -2016,11 +2020,6 @@ int janet_make_pipe(JanetHandle handles[2], int mode) {
 #endif
 }
 
-static void janet_ev_go(JanetFiber *fiber, Janet value, JanetChannel *supervisor_channel) {
-    fiber->supervisor_channel = supervisor_channel;
-    janet_schedule(fiber, value);
-}
-
 /* C functions */
 
 static Janet cfun_ev_go(int32_t argc, Janet *argv) {
@@ -2029,21 +2028,11 @@ static Janet cfun_ev_go(int32_t argc, Janet *argv) {
     Janet value = argc == 2 ? argv[1] : janet_wrap_nil();
     JanetChannel *supervisor_channel = janet_optabstract(argv, argc, 2, &ChannelAT,
                                        janet_vm_root_fiber->supervisor_channel);
-    janet_ev_go(fiber, value, supervisor_channel);
+    fiber->supervisor_channel = supervisor_channel;
+    janet_schedule(fiber, value);
     return argv[0];
 }
 
-static Janet cfun_ev_call(int32_t argc, Janet *argv) {
-    janet_arity(argc, 1, -1);
-    JanetFunction *fn = janet_getfunction(argv, 0);
-    JanetFiber *fiber = janet_fiber(fn, 64, argc - 1, argv + 1);
-    if (NULL == fiber) janet_panicf("invalid arity to function %v", argv[0]);
-    fiber->env = janet_table(0);
-    fiber->env->proto = janet_current_fiber()->env;
-    janet_ev_go(fiber, janet_wrap_nil(), (JanetChannel *)(janet_vm_root_fiber->supervisor_channel));
-    return janet_wrap_fiber(fiber);
-}
-
 static Janet cfun_ev_give_supervisor(int32_t argc, Janet *argv) {
     janet_arity(argc, 1, -1);
     JanetChannel *chan = janet_vm_root_fiber->supervisor_channel;
@@ -2148,12 +2137,6 @@ Janet janet_cfun_stream_write(int32_t argc, Janet *argv) {
 }
 
 static const JanetReg ev_cfuns[] = {
-    {
-        "ev/call", cfun_ev_call,
-        JDOC("(ev/call fn & args)\n\n"
-             "Call a function asynchronously. Returns a fiber that is scheduled to "
-             "run the function.")
-    },
     {
         "ev/go", cfun_ev_go,
         JDOC("(ev/go fiber &opt value supervisor)\n\n"

src/core/os.c

@@ -973,15 +973,34 @@ static Janet os_spawn(int32_t argc, Janet *argv) {
     return os_execute_impl(argc, argv, 1);
 }
 
+#ifdef JANET_EV
+/* Runs in a separate thread */
+static JanetEVGenericMessage os_shell_subr(JanetEVGenericMessage args) {
+    int stat = system((const char *) args.argp);
+    free(args.argp);
+    if (args.argi) {
+        args.tag = JANET_EV_TCTAG_INTEGER;
+    } else {
+        args.tag = JANET_EV_TCTAG_BOOLEAN;
+    }
+    args.argi = stat;
+    return args;
+}
+#endif
+
 static Janet os_shell(int32_t argc, Janet *argv) {
     janet_arity(argc, 0, 1);
     const char *cmd = argc
                       ? janet_getcstring(argv, 0)
                       : NULL;
+#ifdef JANET_EV
+    janet_ev_threaded_await(os_shell_subr, 0, argc, cmd ? strdup(cmd) : NULL);
+#else
     int stat = system(cmd);
     return argc
            ? janet_wrap_integer(stat)
            : janet_wrap_boolean(stat);
+#endif
 }
 
 #endif /* JANET_NO_PROCESSES */

src/include/janet.h

@@ -1324,6 +1324,7 @@ typedef struct {
 #define JANET_EV_TCTAG_ERR_STRING 5   /* cancel with janet_cstringv((const char *) argp) */
 #define JANET_EV_TCTAG_ERR_STRINGF 6  /* cancel with janet_cstringv((const char *) argp), then call free on argp. */
 #define JANET_EV_TCTAG_ERR_KEYWORD 7  /* cancel with janet_ckeywordv((const char *) argp) */
+#define JANET_EV_TCTAG_BOOLEAN 8      /* resume with janet_wrap_boolean(argi) */
 
 /* Function pointer that is run in the thread pool */
 typedef JanetEVGenericMessage(*JanetThreadedSubroutine)(JanetEVGenericMessage arguments);
@@ -1333,7 +1334,7 @@ typedef void (*JanetThreadedCallback)(JanetEVGenericMessage return_value);
 
 /* API calls for quickly offloading some work in C to a new thread or thread pool. */
 JANET_API void janet_ev_threaded_call(JanetThreadedSubroutine fp, JanetEVGenericMessage arguments, JanetThreadedCallback cb);
-JANET_API void janet_ev_threaded_await(JanetThreadedSubroutine fp, int tag, int argi, void *argp);
+JANET_NO_RETURN JANET_API void janet_ev_threaded_await(JanetThreadedSubroutine fp, int tag, int argi, void *argp);
 
 /* Callback used by janet_ev_threaded_await */
 JANET_API void janet_ev_default_threaded_callback(JanetEVGenericMessage return_value);

test/suite0009.janet

@@ -161,8 +161,6 @@
 (ev/sleep 0)
 (ev/cancel fiber "boop")
 
-(assert-error "bad arity to ev/call" (ev/call inc 1 2 3))
-
 (assert (os/execute [janet "-e" `(+ 1 2 3)`] :xp) "os/execute self")
 
 (end-suite)