Revert "ASYNC-262: Added Java system property clojure.core.async.vthreads to control how core.async uses JDK 21+ virtual threads. When the user wishes to target vthreads then core.async will use a vthreads ExecutorService for :io workloads, including the io-thread macro. Also, the go macro will use io-thread as its basis instead of the IOC machinery. Also added a runtime check for vthreads in the case where AOT compiles to vthreads. Finally, added requiring-resolve use in the go macro to load the IOC machinery in the go macro when needed and also when supported by the CLJ version. As a backup to this, added a check at the c.a top level to load the ioc support if the CLJ version doesn't allow nested compilation patterns."

This reverts commit 634d6cb.

Author: puredanger

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

@@ -40,34 +40,7 @@ core.async. If not supplied the ExecutorService for :io will be
 used instead.
 
 The set of contexts may grow in the future so the function should
-return nil for unexpected contexts.
-
-Use the Java system property `clojure.core.async.vthreads` to control
-how core.async uses JDK 21+ virtual threads. The property can be one of
-the following values:
-
-unset - core.async will opportunistically use vthreads when available
-(≥ Java 21) and will otherwise use the old IOC impl. io-thread and :io
-thread pool will run on platform threads if vthreads are not available.
-If AOT compiling, go blocks will always use IOC so that the resulting
-bytecode works on all JVMs (so no change in compiled output)
-
-\"target\" - means that you are targeting virtual threads. At runtime
-from source, go blocks will throw if vthreads are not available.
-If AOT compiling, go blocks are always compiled as normal Clojure
-code to be run on vthreads and will throw at runtime if vthreads are
-not available (Java <21)
-
-\"avoid\" - means that vthreads will not be used by core.async - you can
-use this to minimize impacts if you are not yet ready to utilize vthreads
-in your app. If AOT compiling, go blocks will use IOC. At runtime, io-thread
-and the :io thread pool use platform threads
-
-Note: existing IOC compiled go blocks from older core.async versions continue
-to work (we retain and load the IOC state machine runtime - this does not
-require the analyzer), and you can interact with the same channels from both
-IOC and vthread code.
-"
+return nil for unexpected contexts."
   (:refer-clojure :exclude [reduce transduce into merge map take partition
                             partition-by bounded-count])
   (:require [clojure.core.async.impl.protocols :as impl]
@@ -76,18 +49,14 @@ IOC and vthread code.
             [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]))
 
-(def ^:private lazy-loading-supported? (dispatch/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)
@@ -169,21 +138,6 @@ IOC and vthread code.
   [^long msecs]
   (timers/timeout msecs))
 
-(defn- defparkingop* [op doc arglist]
-  (let [as (mapv #(list 'quote %) arglist)
-        blockingop (-> op name (str "!") symbol)]
-    `(def ~(with-meta op {:arglists `(list ~as) :doc doc})
-       (fn [~'& ~'args]
-         (if (dispatch/in-vthread?)
-           ~(list* apply blockingop '[args])
-           (assert nil ~(str op " used not in (go ...) block")))))))
-
-(defmacro defparkingop
-  "Emits a Var with a function that checks if it's running in a virtual thread. If so then
-  the related blocking op will be called, otherwise the function throws."
-  [op doc arglist]
-  (defparkingop* op doc arglist))
-
 (defmacro defblockingop
   [op doc arglist & body]
   (let [as (mapv #(list 'quote %) arglist)]
@@ -208,11 +162,11 @@ IOC and vthread code.
       @ret
       (deref p))))
 
-(defparkingop <!
-  "takes a val from port. Must be called inside a (go ...) block, or on
-  a virtual thread (no matter how it was started). Will return nil if
-  closed. Will park if nothing is available."
-  [port])
+(defn <!
+  "takes a val from port. Must be called inside a (go ...) block. Will
+  return nil if closed. Will park if nothing is available."
+  [port]
+  (assert nil "<! used not in (go ...) block"))
 
 (defn take!
   "Asynchronously takes a val from port, passing to fn1. Will pass nil
@@ -247,12 +201,12 @@ IOC and vthread code.
       @ret
       (deref p))))
 
-(defparkingop >!
+(defn >!
   "puts a val into port. nil values are not allowed. Must be called
-  inside a (go ...) block, or on a virtual thread (no matter how it
-  was started). Will park if no buffer space is available.
+  inside a (go ...) block. Will park if no buffer space is available.
   Returns true unless port is already closed."
-  [port val])
+  [port val]
+  (assert nil ">! used not in (go ...) block"))
 
 (def ^:private nop (on-caller (fn [_])))
 (def ^:private fhnop (fn-handler nop))
@@ -390,16 +344,16 @@ IOC and vthread code.
       @ret
       (deref p))))
 
-(defparkingop alts!
+(defn alts!
   "Completes at most one of several channel operations. Must be called
-  inside a (go ...) block, or on a virtual thread (no matter how it was
-  started). ports is a vector of channel endpoints, which can be either
-  a channel to take from or a vector of [channel-to-put-to val-to-put],
-  in any combination. Takes will be made as if by <!, and puts will be
-  made as if by >!. Unless the :priority option is true, if more than one
-  port operation is ready a non-deterministic choice will be made. If no
-  operation is ready and a :default value is supplied, [default-val :default]
-  will be returned, otherwise alts! will park until the first operation to
+  inside a (go ...) block. ports is a vector of channel endpoints,
+  which can be either a channel to take from or a vector of
+  [channel-to-put-to val-to-put], in any combination. Takes will be
+  made as if by <!, and puts will be made as if by >!. Unless
+  the :priority option is true, if more than one port operation is
+  ready a non-deterministic choice will be made. If no operation is
+  ready and a :default value is supplied, [default-val :default] will
+  be returned, otherwise alts! will park until the first operation to
   become ready completes. Returns [val port] of the completed
   operation, where val is the value taken for takes, and a
   boolean (true unless already closed, as per put!) for puts.
@@ -413,7 +367,8 @@ IOC and vthread code.
   used, nor in what order should they be, so they should not be
   depended upon for side effects."
 
-  [ports & {:as opts}])
+  [ports & {:as opts}]
+  (assert nil "alts! used not in (go ...) block"))
 
 (defn do-alt [alts clauses]
   (assert (even? (count clauses)) "unbalanced clauses")
@@ -516,22 +471,6 @@ IOC and vthread code.
   (let [ret (impl/take! port (fn-handler nop false))]
     (when ret @ret)))
 
-(defn- go* [body env]
-  (cond (and (not dispatch/virtual-threads-available?)
-             dispatch/target-vthreads?
-             (not clojure.core/*compile-files*))
-        (dispatch/report-vthreads-not-available-error!)
-
-        (or dispatch/target-vthreads?
-            (and dispatch/unset-vthreads?
-                 dispatch/virtual-threads-available?
-                 (not clojure.core/*compile-files*)))
-        `(do (dispatch/ensure-runtime-vthreads!)
-             (thread-call (^:once fn* [] ~@body) :io))
-
-        :else
-        ((requiring-resolve 'clojure.core.async.impl.go/go-impl) env body)))
-
 (defmacro go
   "Asynchronously executes the body, returning immediately to the
   calling thread. Additionally, any visible calls to <!, >! and alt!/alts!
@@ -548,7 +487,7 @@ IOC and vthread code.
   Returns a channel which will receive the result of the body when
   completed"
   [& body]
-  (go* body &env))
+  (#'clojure.core.async.impl.go/go-impl &env body))
 
 (defonce ^:private thread-macro-executor nil)
 

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

@@ -72,63 +72,11 @@
   [workload]
   (Executors/newCachedThreadPool (counted-thread-factory (str "async-" (name workload) "-%d") true)))
 
-(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 virtual-threads-available?
-  (try
-    (Class/forName "java.lang.Thread$Builder$OfVirtual")
-    true
-    (catch ClassNotFoundException _
-      false)))
-
-(defn- vthreads-directive
-  "Retrieves the value of the sysprop clojure.core.async.vthreads."
-  []
-  (System/getProperty "clojure.core.async.vthreads"))
-
-(def target-vthreads?
-  (= (vthreads-directive) "target"))
-
-(def unset-vthreads?
-  (nil? (vthreads-directive)))
-
-(def vthreads-available-and-allowed?
-  (and virtual-threads-available?
-       (not= (vthreads-directive) "avoid")))
-
-(def ^:private virtual-thread?
-  (if virtual-threads-available?
-    (eval `(fn [^Thread t#] (~'.isVirtual t#)))
-    (constantly false)))
-
-(defn in-vthread? []
-  (and virtual-threads-available?
-       (virtual-thread? (Thread/currentThread))))
-
-(defn report-vthreads-not-available-error! []
-  (throw (ex-info "Code compiled to target virtual threads, but is running without vthread support."
-                  {:runtime-jvm-version (System/getProperty "java.version")
-                   :vthreads-directive (vthreads-directive)})))
-
-(defn ensure-runtime-vthreads! []
-  (when (not vthreads-available-and-allowed?)
-    (report-vthreads-not-available-error!)))
-
-(defn- make-io-executor
-  []
-  (if vthreads-available-and-allowed?
-    (-> (.getDeclaredMethod Executors "newVirtualThreadPerTaskExecutor" (make-array Class 0))
-        (.invoke nil (make-array Class 0)))
-    (make-ctp-named :io)))
-
 (defn ^:private create-default-executor
   [workload]
   (case workload
     :compute (make-ctp-named :compute)
-    :io      (make-io-executor)
+    :io      (make-ctp-named :io)
     :mixed   (make-ctp-named :mixed)))
 
 (def executor-for

src/test/clojure/clojure/core/async_test.clj

@@ -193,7 +193,15 @@
       (io-thread (>!! c2 (clojure.string/upper-case (<!! c1))))
       (io-thread (>!! c3 (clojure.string/reverse (<!! c2))))
       (>!! c1 "loop")
-      (is (= "POOL" (<!! c3))))))
+      (is (= "POOL" (<!! c3)))))
+  (testing "io-thread parking op should fail"
+    (let [c1 (chan)]
+      (io-thread
+       (try
+         (>! c1 :no)
+         (catch AssertionError _
+           (>!! c1 :yes))))
+      (is (= :yes (<!! c1))))))
 
 (deftest ops-tests
   (testing "map<"
@@ -480,4 +488,4 @@
     (thrown? AssertionError
       (let [c1 (a/chan)
             c2 (a/chan)]
-        (a/alts!! [c1 [c2 nil]])))))
+        (a/alts!! [c1 [c2 nil]])))))