dispatch.clj

;;   Copyright (c) Rich Hickey and contributors. All rights reserved.
;;   The use and distribution terms for this software are covered by the
;;   Eclipse Public License 1.0 (http://opensource.org/licenses/eclipse-1.0.php)
;;   which can be found in the file epl-v10.html at the root of this distribution.
;;   By using this software in any fashion, you are agreeing to be bound by
;;   the terms of this license.
;;   You must not remove this notice, or any other, from this software.

(ns ^{:skip-wiki true}
  clojure.core.async.impl.dispatch
  (:import [java.util.concurrent Executors ExecutorService ThreadFactory]))

(set! *warn-on-reflection* true)

(defonce executor nil)

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

;; go blocking checking

(defonce in-go-dispatch (ThreadLocal.))

(defmacro with-dispatch-thread-marking
  [& body]
  (if (Boolean/getBoolean "clojure.core.async.go-checking")
    `(try
       (.set ^ThreadLocal in-go-dispatch true)
       ~@body
       (finally
         (.set ^ThreadLocal in-go-dispatch false)))
    `(do ~@body)))

(defn in-dispatch-thread?
  "Returns true if the current thread is used for go block dispatch"
  []
  (boolean (.get ^ThreadLocal in-go-dispatch)))

(defn check-blocking-in-dispatch
  "If the current thread is being used for go block dispatch, throw an exception"
  []
  (when (in-dispatch-thread?)
    (throw (IllegalStateException. "Invalid blocking call in dispatch thread"))))

(defn ex-handler
  "conveys given Exception to current thread's default uncaught handler. returns nil"
  [ex]
  (-> (Thread/currentThread)
      .getUncaughtExceptionHandler
      (.uncaughtException (Thread/currentThread) ex))
  nil)

(defn- make-ctp-named
  [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)
    :mixed   (make-ctp-named :mixed)))

(def executor-for
  "Given a workload tag, returns an ExecutorService instance and memoizes the result. By
  default, core.async will defer to a user factory (if provided via sys prop) or construct
  a specialized ExecutorService instance for each tag :io, :compute, and :mixed. When
  given the tag :core-async-dispatch it will default to the executor service for :io."
  (memoize
   (fn ^ExecutorService [workload]
     (let [sysprop-factory (when-let [esf (System/getProperty "clojure.core.async.executor-factory")]
                             (requiring-resolve (symbol esf)))
           sp-exec (and sysprop-factory (sysprop-factory workload))]
       (or sp-exec
           (if (= workload :core-async-dispatch)
             (executor-for :io)
             (create-default-executor workload)))))))

(defn exec
  [^Runnable r workload]
  (let [^ExecutorService e (executor-for workload)]
    (.execute e r)))

(defn run
  "Runs Runnable r on current thread when :on-caller? meta true, else in a thread pool thread."
  [^Runnable r]
  (if (-> r meta :on-caller?)
    (try (.run r) (catch Throwable t (ex-handler t)))
    (exec r :core-async-dispatch)))