Skip to content

Always run chained callbacks on deferred's executor #258

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
DerGuteMoritz wants to merge 8 commits into
base: master
Choose a base branch
from
Open

Always run chained callbacks on deferred's executor #258

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
8 commits
Select commit Hold shift + click to select a range
96e699c
Centralize callback invocation
DerGuteMoritz Jan 8, 2026
b35d2cf
Also use executor in chained callbacks when deferred is already realized
DerGuteMoritz Jan 8, 2026
c4d084e
Don't re-enqueue callbacks when already on the same executor
DerGuteMoritz Jan 8, 2026
53ab721
Fix race in `test-window-streams`
DerGuteMoritz Jan 8, 2026
986c2e3
Simplify windowing streams
DerGuteMoritz Jan 8, 2026
04f64da
Introduce `manifold.executor/wrap-executor`
DerGuteMoritz Jan 11, 2026
a815909
Prevent redundant executor wrapping
DerGuteMoritz Jan 11, 2026
35eb48a
Fix reflection warnings in `manifold.executor`
DerGuteMoritz Jan 20, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
211 changes: 112 additions & 99 deletions src/manifold/deferred.clj
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -185,6 +185,10 @@
[x]
`(.realized ~(with-meta x {:tag "manifold.deferred.IDeferred"})))

(definline ^:no-doc executor
[x]
`(.executor ~(with-meta x {:tag "manifold.deferred.IDeferred"})))

(definline ^:no-doc success-value [x default-value]
`(.successValue ~(with-meta x {:tag "manifold.deferred.IDeferred"}) ~default-value))

Expand Down Expand Up @@ -283,6 +287,7 @@
(realized? this))
ADeferred
IDeferred
(executor [_])
(realized [_]
(or (.isDone x) (.isCancelled x)))
(onRealized [_ on-success on-error]
Expand Down Expand Up @@ -317,6 +322,7 @@
(.isRealized ^IPending x))
ADeferred
IDeferred
(executor [_])
(realized [_]
(.isRealized ^IPending x))
(onRealized [_ on-success on-error]
Expand Down Expand Up @@ -399,7 +405,19 @@
[^IMutableDeferred deferred listener]
(.cancelListener deferred listener))

(defn execute-callback [^Executor executor callback]
(if (or (nil? executor) (identical? executor (ex/current-executor)))
(callback)
(.execute executor
(fn []
(try
(callback)
(catch Throwable e
(log/error e "error in deferred handler")))))))
Comment thread
DerGuteMoritz marked this conversation as resolved.

(defmacro ^:private set-deferred [val token success? claimed? executor]
;; Relies on all arguments being safe against double-evaluation, i.e. they're local variables at
;; all callsites.
`(if (utils/with-lock* ~'lock
(when (and
(identical? ~(if claimed? ::claimed ::unset) ~'state)
Expand All @@ -412,17 +430,8 @@
(clojure.core/loop []
(when-let [^IDeferredListener l# (.poll ~'listeners)]
(try
(if (nil? ~executor)
(~(if success? `.onSuccess `.onError) ^IDeferredListener l# ~val)
(.execute ~(with-meta executor {:tag "java.util.concurrent.Executor"})
(fn []
(try
(~(if success? `.onSuccess `.onError) l# ~val)
(catch Throwable e#
#_(.printStackTrace e#)
(log/error e# "error in deferred handler"))))))
(execute-callback ~executor #(~(if success? `.onSuccess `.onError) l# ~val))
(catch Throwable e#
#_(.printStackTrace e#)
(log/error e# "error in deferred handler")))
(recur)))
true)
Expand Down Expand Up @@ -518,9 +527,7 @@
(do
(.add listeners listener)
nil)))]
(if executor
(.execute executor f)
(f)))
(execute-callback executor f))
true)
(cancelListener [_ listener]
(utils/with-lock* lock
Expand Down Expand Up @@ -595,9 +602,7 @@
IMutableDeferred
(claim [_] false)
(addListener [_ listener]
(if (nil? executor)
(.onSuccess ^IDeferredListener listener val)
(.execute executor #(.onSuccess ^IDeferredListener listener val)))
(execute-callback executor #(.onSuccess ^IDeferredListener listener val))
true)
(cancelListener [_ listener] false)
(success [_ x] false)
Expand All @@ -613,9 +618,7 @@
(executor [_] executor)
(realized [this] true)
(onRealized [this on-success on-error]
(if executor
(.execute executor #(on-success val))
(on-success val)))
(execute-callback executor #(on-success val)))
(successValue [_ default-value]
val)
(errorValue [_ default-value]
Expand Down Expand Up @@ -654,7 +657,7 @@
(claim [_] false)
(addListener [_ listener]
(set! consumed? true)
(.onError ^IDeferredListener listener error)
(execute-callback executor #(.onError ^IDeferredListener listener error))
true)
(cancelListener [_ listener] false)
(success [_ x] false)
Expand All @@ -671,9 +674,7 @@
(realized [_] true)
(onRealized [this on-success on-error]
(set! consumed? true)
(if (nil? executor)
(on-error error)
(.execute executor #(on-error error))))
(execute-callback executor #(on-error error)))
(successValue [_ default-value]
default-value)
(errorValue [_ default-value]
Expand Down Expand Up @@ -737,29 +738,34 @@
([error executor]
(ErrorDeferred. error nil false executor)))

(declare chain)
(defn- different-executor? [d]
(when-let [ex (executor d)]
(not (identical? ex (ex/current-executor)))))

(defn unwrap'
"Like unwrap, but does not coerce deferrable values."
[x]
(if (deferred? x)
(let [val (success-value x ::none)]
(if (identical? val ::none)
x
(recur val)))
(if (different-executor? x)
x
(let [val (success-value x ::none)]
(if (identical? val ::none)
x
(recur val))))
x))
Comment thread
DerGuteMoritz marked this conversation as resolved.

(defn unwrap
"Recursively unwraps a deferred or deferrable until either 1) a non-deferred
value is reached, or 2) an unrealized deferrable is reached."
"Recursively unwraps a deferred or deferrable until either 1) a non-deferred value is reached, or 2)
an unrealized deferrable is reached, or 3) a realized deferrable with a different executor than

@buhhh buhhh Jan 21, 2026

Copy link
Copy Markdown

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Suggested change
an unrealized deferrable is reached, or 3) a realized deferrable with a different executor than
an unrealized deferrable is reached, or 3) a deferrable with a different executor than

Maybe I am missing something, but it seems to me this would also stop when reaching an unrealized deferrable with a different executor than the current one. Same for the corresponding commit message – which additionally has 'explicit' executor rather than 'different' executor.

Side question: is it manifold parlance that a deferred in error state is considered 'unrealized' here?

the current one is reached."
[x]
(let [d (->deferred x nil)]
(if (nil? d)
x
(let [val (success-value d ::none)]
(if (identical? ::none val)
d
(recur val))))))
(cond (nil? d) x
(different-executor? d) d
:else (let [val (success-value d ::none)]
(if (identical? ::none val)
d
(recur val))))))

(defn connect
"Conveys the realized value of `a` into `b`."
Expand Down Expand Up @@ -1074,40 +1080,44 @@
([x f g & fs]
(apply chain- nil x f g fs)))

(defn catch'
"Like `catch`, but does not coerce deferrable values."
([x error-handler]
(catch' x nil error-handler))
([x error-class error-handler]
(let [x (chain' x)
catch? #(or (nil? error-class) (instance? error-class %))]
(if-not (deferred? x)

;; not a deferred value, skip over it
x

(success-error-unrealized x
val x

err (try
(if (catch? err)
(chain' (error-handler err))
(error-deferred err))
(catch Throwable e
(error-deferred e)))

(let [d' (deferred)]

(on-realized x
#(success! d' %)
#(try
(if (catch? %)
(chain'- d' (error-handler %))
(chain'- d' (error-deferred %)))
(catch Throwable e
(error! d' e))))
(let [subscribe (fn [x catch? error-handler]
(let [d' (deferred)]
(on-realized x
#(success! d' %)
#(try
(if (catch? %)
(chain'- d' (error-handler %))
(chain'- d' (error-deferred %)))
(catch Throwable e
(error! d' e))))

d'))]
(defn catch'
"Like `catch`, but does not coerce deferrable values."
([x error-handler]
(catch' x nil error-handler))
([x error-class error-handler]
(let [catch? #(or (nil? error-class) (instance? error-class %))]
(if (not (deferred? x))
;; not a deferred value, skip over it
x
(let [x' (unwrap' x)]
(if (not (deferred? x'))
;; successfully realized, skip over it
x
(success-error-unrealized x'
val x

err (try
(if (catch? err)
(if (different-executor? x')
(subscribe x catch? error-handler)
(chain' (error-handler err)))
(error-deferred err))
(catch Throwable e
(error-deferred e)))

d'))))))
(subscribe x catch? error-handler)))))))))

(defn catch
"An equivalent of the catch clause, which takes an `error-handler` function that will be invoked
Expand All @@ -1130,36 +1140,39 @@
chain)
x)))

(defn finally'
"Like `finally`, but doesn't coerce deferrable values."
[x f]
(success-error-unrealized x

val (try
(f)
x
(catch Throwable e
(error-deferred e)))

err (try
(f)
(error-deferred err)
(catch Throwable e
(error-deferred e)))

(let [d (deferred)]
(on-realized x
#(try
(f)
(success! d %)
(catch Throwable e
(error! d e)))
#(try
(f)
(error! d %)
(catch Throwable e
(error! d e))))
d)))
(let [subscribe (fn [x f]
(let [d (deferred)]
(on-realized x
#(try
(f)
(success! d %)
(catch Throwable e
(error! d e)))
#(try
(f)
(error! d %)
(catch Throwable e
(error! d e))))
d))]
(defn finally'
"Like `finally`, but doesn't coerce deferrable values."
[x f]
(if (different-executor? x)
(subscribe x f)
(success-error-unrealized x
val (try
(f)
x
(catch Throwable e
(error-deferred e)))

err (try
(f)
(error-deferred err)
(catch Throwable e
(error-deferred e)))

(subscribe x f)))))

(defn finally
"An equivalent of the finally clause, which takes a no-arg side-effecting function that executes
Expand Down
Loading