ASYNC-259 Implement go-checking with new executors

Author: puredanger

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

@@ -13,8 +13,6 @@
 
 (set! *warn-on-reflection* true)
 
-(defonce ^:private in-dispatch (ThreadLocal.))
-
 (defonce executor nil)
 
 (defn counted-thread-factory
@@ -48,15 +46,18 @@
    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

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)))