merged concurrent and threadpool into dispatch

Author: fogus

build.clj

@@ -10,10 +10,8 @@
   (b/delete {:path "target"})
   (b/compile-clj {:basis basis, :src-dirs ["src/main/clojure"], :class-dir class-dir,
                   :filter-nses '[clojure.core.async]
-                  :ns-compile '[clojure.core.async.impl.exec.threadpool
-                                clojure.core.async.impl.protocols
+                  :ns-compile '[clojure.core.async.impl.protocols
                                 clojure.core.async.impl.mutex
-                                clojure.core.async.impl.concurrent
                                 clojure.core.async.impl.dispatch
                                 clojure.core.async.impl.ioc-macros
                                 clojure.core.async.impl.buffers

deps.edn

@@ -1,6 +1,6 @@
 {:paths ["src/main/clojure"]
  :deps
- {org.clojure/tools.analyzer.jvm {:mvn/version "1.3.2"}}
+ {org.clojure/tools.analyzer.jvm {:mvn/version "1.3.1"}}
  :aliases
  {:cljs-test {:extra-deps {org.clojure/clojurescript {:mvn/version "1.11.132"}}
               :extra-paths ["src/main/clojure/cljs" "src/test/cljs"]}

pom.xml

@@ -44,7 +44,7 @@
     <dependency>
       <groupId>org.clojure</groupId>
       <artifactId>tools.analyzer.jvm</artifactId>
-      <version>1.3.2</version>
+      <version>1.3.1</version>
     </dependency>
   </dependencies>
 

project.clj

@@ -5,7 +5,7 @@
             :url "http://www.eclipse.org/legal/epl-v10.html"}
   :parent [org.clojure/pom.contrib "1.2.0"]
   :dependencies [[org.clojure/clojure "1.11.4"]
-                 [org.clojure/tools.analyzer.jvm "1.3.2"]
+                 [org.clojure/tools.analyzer.jvm "1.3.1"]
                  [org.clojure/clojurescript "1.11.132" :scope "provided"]]
   :global-vars {*warn-on-reflection* true}
   :source-paths ["src/main/clojure"]

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

@@ -29,7 +29,6 @@ to catch and handle."
             [clojure.core.async.impl.ioc-macros :as ioc]
             clojure.core.async.impl.go ;; TODO: make conditional
             [clojure.core.async.impl.mutex :as mutex]
-            [clojure.core.async.impl.concurrent :as conc]
             )
   (:import [java.util.concurrent.atomic AtomicLong]
            [java.util.concurrent.locks Lock]

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

@@ -8,16 +8,58 @@
 
 (ns ^{:skip-wiki true}
   clojure.core.async.impl.dispatch
-  (:require [clojure.core.async.impl.protocols :as impl]
-            [clojure.core.async.impl.exec.threadpool :as tp])
-  (:import [java.util.concurrent Executors ExecutorService]))
+  (:require [clojure.core.async.impl.protocols :as impl])
+  (:import [java.util.concurrent Executors ExecutorService ThreadFactory]))
 
 (set! *warn-on-reflection* true)
 
+(defn counted-thread-factory
+  "Create a ThreadFactory that maintains a counter for naming Threads.
+     name-format specifies thread names - use %d to include counter
+     daemon is a flag for whether threads are daemons or not
+     opts is an options map:
+       init-fn - function to run when thread is created"
+  ([name-format daemon]
+    (counted-thread-factory name-format daemon nil))
+  ([name-format daemon {:keys [init-fn] :as opts}]
+   (let [counter (atom 0)]
+     (reify
+       ThreadFactory
+       (newThread [_this runnable]
+         (let [body (if init-fn
+                      (fn [] (init-fn) (.run ^Runnable runnable))
+                      runnable)
+               t (Thread. ^Runnable body)]
+           (doto t
+             (.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)))
+
+(defn thread-pool-executor
+  ([]
+    (thread-pool-executor nil))
+  ([init-fn]
+   (let [executor-svc (Executors/newFixedThreadPool
+                        @pool-size
+                        (counted-thread-factory "async-dispatch-%d" true
+                          {:init-fn init-fn}))]
+     (reify impl/Executor
+       (impl/exec [_ r]
+         (.execute executor-svc ^Runnable r))))))
+
 (defonce ^:private in-dispatch (ThreadLocal.))
 
 (defonce executor
-  (delay (tp/thread-pool-executor #(.set ^ThreadLocal in-dispatch true))))
+  (delay (thread-pool-executor #(.set ^ThreadLocal in-dispatch true))))
 
 (defn in-dispatch-thread?
   "Returns true if the current thread is a go block dispatch pool thread"
@@ -39,13 +81,13 @@
   nil)
 
 (defonce ^ExecutorService mixed-executor
-  (Executors/newCachedThreadPool (conc/counted-thread-factory "async-mixed-%d" true)))
+  (Executors/newCachedThreadPool (counted-thread-factory "async-mixed-%d" true)))
 
 (defonce ^ExecutorService io-executor
-  (Executors/newCachedThreadPool (conc/counted-thread-factory "async-io-%d" true)))
+  (Executors/newCachedThreadPool (counted-thread-factory "async-io-%d" true)))
 
 (defonce ^ExecutorService compute-executor
-  (Executors/newCachedThreadPool (conc/counted-thread-factory "async-compute-%d" true)))
+  (Executors/newCachedThreadPool (counted-thread-factory "async-compute-%d" true)))
 
 (defn run
   "Runs Runnable r on current thread when :on-caller? meta true, else in a thread pool thread."
@@ -57,7 +99,7 @@
 (defn exec
   [f exec]
   (let [^ExecutorService e (case exec
-                             :compute tp/compute-executor
-                             :io tp/io-executor
-                             tp/mixed-executor)]
+                             :compute compute-executor
+                             :io io-executor
+                             mixed-executor)]
     (.execute e f)))

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

@@ -8,12 +8,12 @@
 
 (ns clojure.core.async.concurrent-test
   (:require [clojure.test :refer :all]
-            [clojure.core.async.impl.concurrent :as conc])
+            [clojure.core.async.impl.dispatch :as dispatch])
   (:import [java.util.concurrent ThreadFactory]))
 
 (deftest test-counted-thread-factory
   (testing "Creates numbered threads"
-    (let [^ThreadFactory factory (conc/counted-thread-factory "foo-%d" true)
+    (let [^ThreadFactory factory (dispatch/counted-thread-factory "foo-%d" true)
           threads (repeatedly 3 #(.newThread factory (constantly nil)))]
       (is (= ["foo-1" "foo-2" "foo-3"] (map #(.getName ^Thread %) threads))))))