luciano/babashka
1
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

Go macro uses virtual threads (#1517)

Browse source
This commit is contained in:
Michiel Borkent 2023-03-10 19:22:53 +01:00 committed by GitHub
parent 4438f3bbcf
commit 7aba75564e
No known key found for this signature in database
GPG key ID: 4AEE18F83AFDEB23
2 changed files with 42 additions and 4 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

2
resources/META-INF/babashka/deps.edn
View file

@ -50,7 +50,7 @@
org.clojure/data.priority-map {:mvn/version "1.1.0"}
insn/insn {:mvn/version "0.5.2"}
org.clojure/core.rrb-vector {:mvn/version "0.1.2"}
org.babashka/cli {:mvn/version "0.6.46"}
org.babashka/cli {:mvn/version "0.6.49"}
org.babashka/http-client {:mvn/version "0.1.6"}}
:aliases {:babashka/dev
{:main-opts ["-m" "babashka.main"]}

44
src/babashka/impl/clojure/core/async.clj
View file

@ -8,6 +8,10 @@
(def ^java.util.concurrent.Executor executor @#'async/thread-macro-executor)
(def ^java.util.concurrent.Executor virtual-executor
(try (eval '(java.util.concurrent.Executors/newVirtualThreadPerTaskExecutor))
(catch Exception _ 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
@ -26,10 +30,32 @@
(async/close! c))))))
c))
(defn -vthread-call
"Executes f in another virtual thread, returning immediately to the calling
thread. Returns a channel which will receive the result of calling
f when completed, then close."
[f]
(let [c (async/chan 1)]
(let [binds (vars/get-thread-binding-frame)]
(.execute virtual-executor
(fn []
(vars/reset-thread-binding-frame binds)
(try
(let [ret (f)]
(when-not (nil? ret)
(async/>!! c ret)))
(finally
(async/close! c))))))
c))
(defn thread
[_ _ & body]
`(~'clojure.core.async/thread-call (fn [] ~@body)))
(defn -vthread
[_ _ & body]
`(~'clojure.core.async/-vthread-call (fn [] ~@body)))
(defn alt!!
"Like alt!, except as if by alts!!, will block until completed, and
not intended for use in (go ...) blocks."
@ -38,10 +64,19 @@
(defn go-loop
[_ _ bindings & body]
(list 'clojure.core.async/thread (list* 'loop bindings body)))
(list 'clojure.core.async/go (list* 'loop bindings body)))
(def core-async-namespace (sci/create-ns 'clojure.core.async nil))
(defn timeout [ms]
(if virtual-executor
(let [chan (async/chan nil)]
(.execute virtual-executor (fn []
(Thread/sleep ms)
(async/close! chan)))
chan)
(async/timeout ms)))
(def async-namespace
{:obj core-async-namespace
'<!! (copy-var async/<!! core-async-namespace)
@ -92,7 +127,8 @@
'tap (copy-var async/tap core-async-namespace)
'thread (macrofy 'thread thread core-async-namespace)
'thread-call (copy-var thread-call core-async-namespace)
'timeout (copy-var async/timeout core-async-namespace)
'-vthread-call (copy-var -vthread-call core-async-namespace)
'timeout (copy-var timeout core-async-namespace)
'to-chan (copy-var async/to-chan core-async-namespace)
'to-chan! (copy-var async/to-chan! core-async-namespace)
'to-chan!! (copy-var async/to-chan!! core-async-namespace)
@ -107,7 +143,9 @@
'untap (copy-var async/untap core-async-namespace)
'untap-all (copy-var async/untap-all core-async-namespace)
;; polyfill
'go (macrofy 'go thread core-async-namespace)
'go (if virtual-executor
(macrofy 'go -vthread core-async-namespace)
(macrofy 'go thread core-async-namespace))
'<! (copy-var async/<!! core-async-namespace {:name '<!})
'>! (copy-var async/>!! core-async-namespace {:name '>!})
'alt! (macrofy 'alt! alt!! core-async-namespace)