Merge branch 'master' into dev-vthreads-exec

Author: fogus

README.md

@@ -65,6 +65,10 @@ Copyright © Rich Hickey and contributors
 
 ## Changelog
 
+* Release 1.8.718-beta2 on 2025.03.10
+  * [ASYNC-259](https://clojure.atlassian.net/browse/ASYNC-259) (CLJ) Implement `clojure.core.async.go-checking` with new executor pools
+  * [ASYNC-260](https://clojure.atlassian.net/browse/ASYNC-260) (CLJ) Remove now unused property `clojure.core.async.pool-size`
+  * [ASYNC-261](https://clojure.atlassian.net/browse/ASYNC-261) (CLJ) Put clojure.core.async/thread-macro-executor var back into place
 * Release 1.8.711-beta1 on 2025.02.19
   * [ASYNC-256](https://clojure.atlassian.net/browse/ASYNC-256) (CLJ) Add io-thread and System property clojure.core.async.executor-factory
   * [ASYNC-255](https://clojure.atlassian.net/browse/ASYNC-255) (CLJ) alts guards against put of nil message on entry

VERSION_TEMPLATE

@@ -1 +1 @@
-1.8.GENERATED_VERSION-beta1
+1.8.GENERATED_VERSION-beta2

pom.xml

@@ -3,7 +3,7 @@
   <groupId>org.clojure</groupId>
   <artifactId>core.async</artifactId>
   <!-- Don't set this manually! Call script/build/update_version -->
-  <version>1.8.712-SNAPSHOT</version>
+  <version>1.9.0-SNAPSHOT</version>
   <packaging>jar</packaging>
   <name>core.async</name>
   <description>Facilities for async programming and communication in Clojure</description>

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

@@ -9,10 +9,6 @@
 (ns clojure.core.async
   "Facilities for async programming and communication.
 
-go blocks are dispatched over an internal thread pool, which
-defaults to 8 threads. The size of this pool can be modified using
-the Java system property `clojure.core.async.pool-size`.
-
 Set Java system property `clojure.core.async.go-checking` to true
 to validate go blocks do not invoke core.async blocking operations.
 Property is read once, at namespace load time. Recommended for use
@@ -58,9 +54,8 @@ return nil for unexpected contexts."
             )
   (:import [java.util.concurrent.atomic AtomicLong]
            [java.util.concurrent.locks Lock]
-           [java.util.concurrent Executors Executor ThreadLocalRandom]
-           [java.util Arrays ArrayList]
-           [clojure.lang Var]))
+           [java.util.concurrent ThreadLocalRandom]
+           [java.util Arrays ArrayList]))
 
 (alias 'core 'clojure.core)
 
@@ -507,6 +502,8 @@ return nil for unexpected contexts."
     `(thread-call (^:once fn* [] ~@body) :io)
     (#'clojure.core.async.impl.go/go-impl &env body)))
 
+(defonce ^:private thread-macro-executor nil)
+
 (defn thread-call
   "Executes f in another thread, returning immediately to the calling
   thread. Returns a channel which will receive the result of calling

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

@@ -8,13 +8,10 @@
 
 (ns ^{:skip-wiki true}
   clojure.core.async.impl.dispatch
-  (:require [clojure.core.async.impl.protocols :as impl])
   (:import [java.util.concurrent Executors ExecutorService ThreadFactory]))
 
 (set! *warn-on-reflection* true)
 
-(defonce ^:private in-dispatch (ThreadLocal.))
-
 (defonce executor nil)
 
 (defn counted-thread-factory
@@ -38,25 +35,18 @@
              (.setName (format name-format (swap! counter inc)))
              (.setDaemon daemon))))))))
 
-(defonce
-  ^{:doc "Number of processors reported by the JVM"}
-  processors (.availableProcessors (Runtime/getRuntime)))
-
-(def ^:private pool-size
-  "Value is set via clojure.core.async.pool-size system property; defaults to 8; uses a
-   delay so property can be set from code after core.async namespace is loaded but before
-   any use of the async thread pool."
-  (delay (or (Long/getLong "clojure.core.async.pool-size") 8)))
+;; used only for implementing go-checking
+(def ^:private ^:dynamic *in-go-dispatch* false)
 
 (defn in-dispatch-thread?
-  "Returns true if the current thread is a go block dispatch pool thread"
+  "Returns true if the current thread is used for go block dispatch"
   []
-  (boolean (.get ^ThreadLocal in-dispatch)))
+  (boolean *in-go-dispatch*))
 
 (defn check-blocking-in-dispatch
-  "If the current thread is a dispatch pool thread, throw an exception"
+  "If the current thread is being used for go block dispatch, throw an exception"
   []
-  (when (.get ^ThreadLocal in-dispatch)
+  (when (in-dispatch-thread?)
     (throw (IllegalStateException. "Invalid blocking call in dispatch thread"))))
 
 (defn ex-handler
@@ -68,8 +58,8 @@
   nil)
 
 (defn- make-ctp-named
-  [workflow]
-  (Executors/newCachedThreadPool (counted-thread-factory (str "async-" (name workflow) "-%d") true)))
+  [workload]
+  (Executors/newCachedThreadPool (counted-thread-factory (str "async-" (name workload) "-%d") true)))
 
 (def virtual-threads-available?
   (try

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

@@ -1041,18 +1041,23 @@
       second
       (emit-state-machine num-user-params user-transitions))))
 
+(def ^:private go-checking (Boolean/getBoolean "clojure.core.async.go-checking"))
+
 (defn go-impl
   [env body]
-  (let [crossing-env (zipmap (keys env) (repeatedly gensym))]
-    `(let [c# (clojure.core.async/chan 1)
-           captured-bindings# (Var/getThreadBindingFrame)]
-       (dispatch/run
-         (^:once fn* []
-           (let [~@(mapcat (fn [[l sym]] [sym `(^:once fn* [] ~(vary-meta l dissoc :tag))]) crossing-env)
-                 f# ~(state-machine
-                       `(do ~@body) 1 [crossing-env env] rt/async-custom-terminators)
-                 state# (-> (f#)
-                          (rt/aset-all! rt/USER-START-IDX c#
-                            rt/BINDINGS-IDX captured-bindings#))]
-             (rt/run-state-machine-wrapped state#))))
-       c#)))
+  (let [crossing-env (zipmap (keys env) (repeatedly gensym))
+        run-body `(let [c# (clojure.core.async/chan 1)
+                        captured-bindings# (Var/getThreadBindingFrame)]
+                    (dispatch/run
+                      (^:once fn* []
+                        (let [~@(mapcat (fn [[l sym]] [sym `(^:once fn* [] ~(vary-meta l dissoc :tag))]) crossing-env)
+                              f# ~(state-machine
+                                    `(do ~@body) 1 [crossing-env env] rt/async-custom-terminators)
+                              state# (-> (f#)
+                                       (rt/aset-all! rt/USER-START-IDX c#
+                                         rt/BINDINGS-IDX captured-bindings#))]
+                          (rt/run-state-machine-wrapped state#))))
+                    c#)]
+    (if go-checking
+      (list 'binding ['clojure.core.async.impl.dispatch/*in-go-dispatch* true] run-body)
+      run-body)))