AYNC-272: Restricted vthreads use to io-thread. go again uses IOC. Removed sysprops around vthreads. go block dispatch uses :mixed instead of :io workload profile. ExecutorService -> Executor and fixed flow to use this. Removed use of explicit ExecutorService to create vthreads and use the one-shot in a reify instance instead. Removed ability to call parking ops in io-thread. Modified docs.

Author: fogus

src/main/clojure/clojure/core/async.clj

@@ -17,11 +17,11 @@ go block threads - use Thread.setDefaultUncaughtExceptionHandler()
 to catch and handle.
 
 Use the Java system property `clojure.core.async.executor-factory`
-to specify a function that will provide ExecutorServices for
+to specify a function that will provide an Executor instance for
 application-wide use by core.async in lieu of its defaults. The
 property value should name a fully qualified var. The function
 will be passed a keyword indicating the context of use of the
-executor, and should return either an ExecutorService, or nil to
+executor, and should return either an Executor, or nil to
 use the default. Results per keyword will be cached and used for
 the remainder of the application. Possible context arguments are:
 
@@ -36,7 +36,7 @@ flow/process
 
 :core-async-dispatch - used for completion fn handling (e.g. in put!
 and take!, as well as go block IOC thunk processing) throughout
-core.async. If not supplied the ExecutorService for :io will be
+core.async. If not supplied, the Executor for :io will be
 used instead.
 
 The set of contexts may grow in the future so the function should
@@ -49,14 +49,23 @@ return nil for unexpected contexts."
             [clojure.core.async.impl.timers :as timers]
             [clojure.core.async.impl.dispatch :as dispatch]
             [clojure.core.async.impl.ioc-macros :as ioc]
-            clojure.core.async.impl.go ;; TODO: make conditional
             [clojure.core.async.impl.mutex :as mutex]
             )
   (:import [java.util.concurrent.atomic AtomicLong]
            [java.util.concurrent.locks Lock]
            [java.util.concurrent ThreadLocalRandom]
            [java.util Arrays ArrayList]))
 
+(defn- at-least-clojure-version?
+  [[maj min incr]]
+  (let [{:keys [major minor incremental]} *clojure-version*]
+    (not (neg? (compare [major minor incremental] [maj min incr])))))
+
+(def ^:private lazy-loading-supported? (at-least-clojure-version? [1 12 3]))
+
+(when-not lazy-loading-supported?
+  (require 'clojure.core.async.impl.go))
+
 (alias 'core 'clojure.core)
 
 (set! *warn-on-reflection* false)
@@ -485,9 +494,13 @@ return nil for unexpected contexts."
   core.async blocking ops (those ending in !!) and other blocking IO.
 
   Returns a channel which will receive the result of the body when
-  completed"
+  completed.
+
+  The resources associated with a go block may be reclaimed, and the block
+  never resumed, when the channels with which it interacts are no longer
+  referenced (outside of the go block)."
   [& body]
-  (#'clojure.core.async.impl.go/go-impl &env body))
+  ((requiring-resolve 'clojure.core.async.impl.go/go-impl) &env body))
 
 (defonce ^:private thread-macro-executor nil)
 
@@ -524,7 +537,12 @@ return nil for unexpected contexts."
   "Executes the body in a thread, returning immediately to the calling
   thread. The body may do blocking I/O but must not do extended computation.
   Returns a channel which will receive the result of the body when completed,
-  then close."
+  then close.
+
+  Will dispatch to a virtual thread if available in the runtime JVM, and an
+  ordinary thread if not. Like 'thread's (and unlike 'go' blocks) 'io-thread's
+  must terminate ordinarily, and will keep referenced resources alive
+  until they do."
   [& body]
   `(thread-call (^:once fn* [] ~@body) :io))
 

src/main/clojure/clojure/core/async/flow.clj

@@ -98,8 +98,8 @@
   async/sliding-buffer of size 100, thus signals not handled in a
   timely manner will be dropped in favor of later arriving signals.
   
-  :mixed-exec/:io-exec/:compute-exec -> ExecutorService
-  These can be used to specify the ExecutorService to use for the
+  :mixed-exec/:io-exec/:compute-exec -> Executor
+  These can be used to specify the Executor to use for the
   corresonding workload, in lieu of the lib defaults.
 
   N.B. The flow is not started. See 'start'"
@@ -334,7 +334,7 @@
 
   futurize accepts kwarg options:
   :exec - one of the workloads :mixed, :io, :compute
-          or a j.u.c.ExecutorService object,
+          or a j.u.c.Executor object,
           default :mixed"
   [f & {:keys [exec]
         :or {exec :mixed} :as opts}]

src/main/clojure/clojure/core/async/flow/impl.clj

@@ -14,24 +14,24 @@
             [clojure.core.async.impl.dispatch :as disp]
             [clojure.walk :as walk]
             [clojure.datafy :as datafy])
-  (:import [java.util.concurrent Future Executors ExecutorService TimeUnit]
+  (:import [java.util.concurrent Future Executors Executor TimeUnit]
            [java.util.concurrent.locks ReentrantLock]))
 
 (set! *warn-on-reflection* true)
 
 (defn datafy [x]
   (condp instance? x
     clojure.lang.Fn (-> x str symbol)
-    ExecutorService (str x)
+    Executor (str x)
     clojure.lang.Var (symbol x)
     (datafy/datafy x)))
 
 (defn futurize [f {:keys [exec]}]
   (fn [& args]
-    (let [^ExecutorService e (if (instance? ExecutorService exec)
-                               exec
-                               (disp/executor-for exec))]
-      (.submit e ^Callable #(apply f args)))))
+    (let [^Executor e (if (instance? Executor exec)
+                        exec
+                        (disp/executor-for exec))]
+      (.execute e ^Runnable #(apply f args)))))
 
 (defn prep-proc [ret pid {:keys [proc, args, chan-opts] :or {chan-opts {}}}]
   (let [{:keys [ins outs signal-select]} (spi/describe proc)
@@ -51,8 +51,8 @@
   (let [lock (ReentrantLock.)
         chans (atom nil)
         execs {:mixed mixed-exec :io io-exec :compute compute-exec}
-        _ (assert (every? #(or (nil? %) (instance? ExecutorService %)) (vals execs))
-                  "mixed-exe, io-exec and compute-exec must be ExecutorServices")
+        _ (assert (every? #(or (nil? %) (instance? Executor %)) (vals execs))
+                  "mixed-exe, io-exec and compute-exec must be Executors")
         pdescs (reduce-kv prep-proc {} procs)
         allopts (fn [iok] (into {} (mapcat #(map (fn [[k opts]] [[(:pid %) k] opts]) (iok %)) (vals pdescs))))
         inopts (allopts :ins)

src/main/clojure/clojure/core/async/flow/spi.clj

@@ -82,7 +82,7 @@
           N.B. outputs may be nil if not connected
   :resolver - an impl of spi/Resolver, which can be used to find
               channels given their logical [pid cid] coordinates, as well as to
-              obtain ExecutorServices corresponding to the
+              obtain Executors corresponding to the
               logical :mixed/:io/:compute contexts"))
 
 (defprotocol Resolver
@@ -91,5 +91,5 @@
     write to or nil (in which case the output should be dropped,
     e.g. nothing is connected).")
   (get-exec [_ context]
-    "returns the ExecutorService for the given context, one
+    "returns the Executor for the given context, one
      of :mixed, :io, :compute"))

src/main/clojure/clojure/core/async/impl/dispatch.clj

@@ -8,7 +8,7 @@
 
 (ns ^{:skip-wiki true}
   clojure.core.async.impl.dispatch
-  (:import [java.util.concurrent Executors ExecutorService ThreadFactory]))
+  (:import [java.util.concurrent Executors Executor ThreadFactory]))
 
 (set! *warn-on-reflection* true)
 
@@ -72,31 +72,47 @@
   [workload]
   (Executors/newCachedThreadPool (counted-thread-factory (str "async-" (name workload) "-%d") true)))
 
+(def virtual-threads-available?
+  (try
+    (Class/forName "java.lang.Thread$Builder$OfVirtual")
+    true
+    (catch ClassNotFoundException _
+      false)))
+
+(defn- make-io-executor
+  []
+  (if virtual-threads-available?
+    (let [svt (.getDeclaredMethod Thread "startVirtualThread" (into-array Class [Runnable]))]
+      (reify Executor
+        (execute [_ r]
+          (.invoke svt nil (object-array [r])))))
+    (make-ctp-named :io)))
+
 (defn ^:private create-default-executor
   [workload]
   (case workload
     :compute (make-ctp-named :compute)
-    :io      (make-ctp-named :io)
+    :io      (make-io-executor)
     :mixed   (make-ctp-named :mixed)))
 
 (def executor-for
-  "Given a workload tag, returns an ExecutorService instance and memoizes the result. By
+  "Given a workload tag, returns an Executor 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
+  a specialized Executor 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]
+   (fn ^Executor [workload]
      (let [sysprop-factory (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)
+             (executor-for :mixed)
              (create-default-executor workload)))))))
 
 (defn exec
   [^Runnable r workload]
-  (let [^ExecutorService e (executor-for workload)]
+  (let [^Executor e (executor-for workload)]
     (.execute e r)))
 
 (defn run