rearranged to avoid internal require

Author: fogus

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

@@ -34,7 +34,8 @@ to catch and handle."
   (:import [java.util.concurrent.atomic AtomicLong]
            [java.util.concurrent.locks Lock]
            [java.util.concurrent Executors Executor ThreadLocalRandom]
-           [java.util Arrays ArrayList]))
+           [java.util Arrays ArrayList]
+           [clojure.lang Var]))
 
 (alias 'core 'clojure.core)
 
@@ -462,29 +463,43 @@ to catch and handle."
   [& body]
   (#'clojure.core.async.impl.go/go-impl &env body))
 
-(require '[clojure.core.async.impl.exec.services :as exec-services])
+(defn thread-impl
+  [f workload]
+  (let [c (chan 1)]
+    (let [binds (Var/getThreadBindingFrame)]
+      (dispatch/executor-service-call
+       (fn []
+         (Var/resetThreadBindingFrame binds)
+         (try
+           (let [ret (f)]
+             (when-not (nil? ret)
+               (>!! c ret)))
+           (finally
+             (close! c))))
+       workload))
+    c))
 
 (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]
-  (exec-services/thread-call f :mixed))
+  (thread-impl f :mixed))
 
 (defmacro thread
   "Executes the body in another thread, returning immediately to the
   calling thread. Returns a channel which will receive the result of
   the body when completed, then close."
   [& body]
-  `(thread-call (^:once fn* [] ~@body)))
+  `(thread-impl (^:once fn* [] ~@body) :mixed))
 
 (defmacro io-thread
   "Executes the body in a thread intended for blocking I/O workloads,
   returning immediately to the calling thread. The body must not do
   extended computation (if so, use 'thread' instead). Returns a channel
   which will receive the result of the body when completed, then close."
   [& body]
-  `(exec-services/thread-call (^:once fn* [] ~@body) :io))
+  `(thread-impl (^:once fn* [] ~@body) :io))
 
 ;;;;;;;;;;;;;;;;;;;; ops ;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;
 

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

@@ -9,7 +9,8 @@
 (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]))
+            [clojure.core.async.impl.exec.threadpool :as tp])
+  (:import [java.util.concurrent ExecutorService]))
 
 (set! *warn-on-reflection* true)
 
@@ -43,3 +44,11 @@
   (if (-> r meta :on-caller?)
     (try (.run r) (catch Throwable t (ex-handler t)))
     (impl/exec @executor r)))
+
+(defn executor-service-call
+  [f exec]
+  (let [^ExecutorService e (case exec
+                             :compute tp/compute-executor
+                             :io tp/io-executor
+                             tp/mixed-executor)]
+    (.execute e f)))

src/main/clojure/clojure/core/async/impl/exec/services.clj

@@ -9,7 +9,7 @@
 (ns clojure.core.async.impl.exec.threadpool
   (:require [clojure.core.async.impl.protocols :as impl]
             [clojure.core.async.impl.concurrent :as conc])
-  (:import [java.util.concurrent Executors]))
+  (:import [java.util.concurrent Executors ExecutorService]))
 
 (set! *warn-on-reflection* true)
 
@@ -30,3 +30,12 @@
      (reify impl/Executor
        (impl/exec [_ r]
          (.execute executor-svc ^Runnable r))))))
+
+(defonce ^ExecutorService mixed-executor
+  (Executors/newCachedThreadPool (conc/counted-thread-factory "async-mixed-%d" true)))
+
+(defonce ^ExecutorService io-executor
+  (Executors/newCachedThreadPool (conc/counted-thread-factory "async-io-%d" true)))
+
+(defonce ^ExecutorService compute-executor
+  (Executors/newCachedThreadPool (conc/counted-thread-factory "async-compute-%d" true)))