wip

deps.edn

@@ -25,7 +25,7 @@
         babashka/babashka.curl {:local/root "babashka.curl"}
         babashka/fs {:local/root "fs"}
         babashka/babashka.core {:local/root "babashka.core"}
-        org.clojure/core.async {:mvn/version "1.7.701"},
+        org.clojure/core.async {:mvn/version "1.8.711-beta1"},
         org.clojure/tools.cli {:mvn/version "1.0.214"},
         org.clojure/data.csv {:mvn/version "1.0.0"},
         cheshire/cheshire {:mvn/version "5.13.0"}

project.clj

@@ -29,7 +29,7 @@
                  [borkdude/sci.impl.reflector "0.0.3"]
                  [org.babashka/sci.impl.types "0.0.2"]
                  [org.babashka/babashka.impl.java "0.1.10"]
-                 [org.clojure/core.async "1.7.701"]
+                 [org.clojure/core.async "1.8.711-beta1"]
                  [org.clojure/test.check "1.1.1"]
                  [com.github.clj-easy/graal-build-time "0.1.0"]
                  [rewrite-clj/rewrite-clj "1.1.49"]

resources/META-INF/babashka/deps.edn

@@ -25,7 +25,7 @@
         babashka/babashka.curl {:local/root "babashka.curl"}
         babashka/fs {:local/root "fs"}
         babashka/babashka.core {:local/root "babashka.core"}
-        org.clojure/core.async {:mvn/version "1.7.701"},
+        org.clojure/core.async {:mvn/version "1.8.711-beta1"},
         org.clojure/tools.cli {:mvn/version "1.0.214"},
         org.clojure/data.csv {:mvn/version "1.0.0"},
         cheshire/cheshire {:mvn/version "5.13.0"}

src/aaaa_this_has_to_be_first/because_patches.clj

@@ -1,7 +1,8 @@
 (ns aaaa-this-has-to-be-first.because-patches
   ;; we need pprint loaded first, it patches pprint to not bloat the GraalVM binary
   (:require [babashka.impl.patches.datafy]
-            [babashka.impl.pprint]))
+            [babashka.impl.pprint]
+            ))
 
 ;; Enable this for scanning requiring usage:
 (def enable-require-scan

src/babashka/impl/clojure/core/async.clj

@@ -2,13 +2,38 @@
   {:no-doc true}
   (:require [clojure.core.async :as async]
             [clojure.core.async.impl.protocols :as protocols]
+            [clojure.core.async.impl.dispatch :as dispatch]
             [sci.core :as sci :refer [copy-var]]
             [sci.impl.copy-vars :refer [macrofy]]
-            [sci.impl.vars :as vars]))
+            [sci.impl.vars :as vars])
+  (:import [java.util.concurrent Executors ExecutorService ThreadFactory]))
 
 (set! *warn-on-reflection* true)
 
-(def ^java.util.concurrent.Executor executor @#'async/thread-macro-executor)
+#_(def ^java.util.concurrent.Executor executor @#'async/thread-macro-executor)
+(def executor-for
+  "Given a workload tag, returns an ExecutorService instance and memoizes the result. By
+  default, core.async will defer to a user factory (if provided via sys prop) or construct
+  a specialized ExecutorService instance for each tag :io, :compute, and :mixed. When
+  given the tag :core-async-dispatch it will default to the executor service for :io."
+  (memoize
+   (fn ^ExecutorService [workload]
+     (let [sysprop-factory nil #_(when-let [esf (System/getProperty "clojure.core.async.executor-factory")]
+                             (requiring-resolve (symbol esf)))
+           sp-exec (and sysprop-factory (sysprop-factory workload))]
+       (or sp-exec
+           (if (= workload :core-async-dispatch)
+             (executor-for :io)
+             (@#'dispatch/create-default-executor workload)))))))
+
+(alter-var-root #'dispatch/executor-for (constantly executor-for))
+
+(defn exec
+  [^Runnable r workload]
+  (let [^ExecutorService e (executor-for workload)]
+    (.execute e r)))
+
+(alter-var-root #'dispatch/exec (constantly exec))
 
 (def ^java.util.concurrent.Executor virtual-executor
   (try (eval '(java.util.concurrent.Executors/newVirtualThreadPerTaskExecutor))
@@ -17,20 +42,24 @@
 (defn thread-call
   "Executes f in another thread, returning immediately to the calling
   thread. Returns a channel which will receive the result of calling
-  f when completed, then close."
+  f when completed, then close. workload is a keyword that describes
-  [f]
+  the work performed by f, where:
-  (let [c (async/chan 1)]
+
-    (let [binds (vars/get-thread-binding-frame)]
+  :io - may do blocking I/O but must not do extended computation
-      (.execute executor
+  :compute - must not ever block
-                (fn []
+  :mixed - anything else (default)
-                  (vars/reset-thread-binding-frame binds)
+
-                  (try
+  when workload not supplied, defaults to :mixed"
-                    (let [ret (f)]
+  ([f] (thread-call f :mixed))
-                      (when-not (nil? ret)
+  ([f workload]
-                        (async/>!! c ret)))
+   (let [c (async/chan 1)
-                    (finally
+         returning-to-chan (fn [bf]
-                      (async/close! c))))))
+                             #(try
-    c))
+                                (when-some [ret (bf)]
+                                  (async/>!! c ret))
+                                (finally (async/close! c))))]
+     (-> f bound-fn* returning-to-chan (exec workload))
+     c)))
 
 (defn -vthread-call
   "Executes f in another virtual thread, returning immediately to the calling

src/babashka/impl/pprint.clj

@@ -2,7 +2,8 @@
   {:no-doc true}
   (:require [clojure.pprint :as pprint]
             [sci.core :as sci]
-            [sci.pprint]))
+            [sci.pprint]
+            [babashka.impl.clojure.core.async]))
 
 (defonce patched? (volatile! false))