Fix dropped errors in CompletionStage's *EitherAsync methods

For `*Async` variants of `CompletionStage` methods, the (first) deferred argument was wrapped with
`onto` so that its callbacks are invoked on the given `executor`. This results in a new deferred
connected to the original one. In case of the `*EitherAsync` methods, that new deferred would then
leak when `alt` chooses the `other` one but the `onto` deferred ends up in an error state since
there was no way for the caller to attach an error handler to it.

Furthermore, for async variants of methods accepting two completion stages, `f` was never executed
on the given executor. For example, `then-combine` was defined like this:

    (defn- then-combine [d other ^BiFunction f]
      (assert-some other f)
      (fmap-deferred (zip d other)
                    (fn [[x y]] (.apply f x y))))

Now `then-combine-async` (defined via `def-async-for-dual`) would wrap `d` with `(onto d executor)`
before calling `then-combine`. However, the deferred passed to `fmap-deferred` was the one returned
from `(zip d other)` which would not have the given executor attached. Thus, `f` would have been
executed on a different executor than desired.

This patch addresses both issues by adding a `result` deferred as the final argument to every method
implementation. This deferred is bound to the desired `executor`. The operation `f` is then attached
as a callback to that `result` deferred and the deferred of the main operation is connected to
it. For synchronous implementations, `nil` can be passed as `result` in which case the callback is
attached directly to the main operation deferred to save some overhead. This means that now the
original deferred is passed to `alt` so that all error states can be handled by the caller.

It also removes the `def-async-for` and `def-async-for-dual` macros and opts to instead always
explicitly pass the `result` deferred in all method implementations. This isn't much more verbose
but hopefully serves to make the code bit easier to follow.

Finally, it renames `fmap-deferred` to `shallow-chain` for consistency with the established `chain`
naming and introduces the analogous `shallow-connect` (also private for now).

Author: DerGuteMoritz

src/manifold/deferred.clj

@@ -74,7 +74,9 @@
 
          to-completable-future
 
-         when-complete when-complete-async)
+         when-complete when-complete-async
+
+         deferred cs-default-executor)
 
 ;; The potemkin abstract type for
 ;; implementations such as CompletionStage
@@ -84,94 +86,99 @@
   ADeferred
   CompletionStage
   (thenApply [d f]
-    (then-apply d f))
+    (then-apply d f nil))
   (thenApplyAsync [d f]
-    (then-apply-async d f))
+    (then-apply d f (deferred (cs-default-executor))))
   (thenApplyAsync [d f executor]
-    (then-apply-async d f executor))
+    (then-apply d f (deferred executor)))
 
   (thenAccept [d f]
-    (then-accept d f))
+    (then-accept d f nil))
   (thenAcceptAsync [d f]
-    (then-accept-async d f))
+    (then-accept d f (deferred (cs-default-executor))))
   (thenAcceptAsync [d f executor]
-    (then-accept-async d f executor))
+    (then-accept d f (deferred executor)))
 
   (thenRun [d f]
-    (then-run d f))
+    (then-run d f nil))
   (thenRunAsync [d f]
-    (then-run-async d f))
+    (then-run d f (deferred (cs-default-executor))))
   (thenRunAsync [d f executor]
-    (then-run-async d f executor))
+    (then-run d f (deferred executor)))
 
   (thenCombine [d other f]
-    (then-combine d other f))
+    (then-combine d other f nil))
   (thenCombineAsync [d other f]
-    (then-combine-async d other f))
+    (then-combine d other f (deferred (cs-default-executor))))
   (thenCombineAsync [d other f executor]
-    (then-combine-async d other f executor))
+    (then-combine d other f (deferred executor)))
 
   (thenAcceptBoth [d other f]
-    (then-accept-both d other f))
+    (then-accept-both d other f nil))
   (thenAcceptBothAsync [d other f]
-    (then-accept-both-async d other f))
+    (then-accept-both d other f (deferred (cs-default-executor))))
   (thenAcceptBothAsync [d other f executor]
-    (then-accept-both-async d other f executor))
+    (then-accept-both d other f (deferred executor)))
 
   (runAfterBoth [d other f]
-    (run-after-both d other f))
+    (run-after-both d other f nil))
   (runAfterBothAsync [d other f]
-    (run-after-both-async d other f))
+    (run-after-both d other f (deferred (cs-default-executor))))
   (runAfterBothAsync [d other f executor]
-    (run-after-both-async d other f executor))
+    (run-after-both d other f (deferred executor)))
 
   (applyToEither [d other f]
-    (apply-to-either d other f))
+    (apply-to-either d other f nil))
   (applyToEitherAsync [d other f]
-    (apply-to-either-async d other f))
+    (apply-to-either d other f (deferred (cs-default-executor))))
   (applyToEitherAsync [d other f executor]
-    (apply-to-either-async d other f executor))
+    (apply-to-either d other f (deferred executor)))
 
   (acceptEither [d other f]
-    (accept-either d other f))
+    (accept-either d other f nil))
   (acceptEitherAsync [d other f]
-    (accept-either-async d other f))
+    (accept-either d other f (deferred (cs-default-executor))))
   (acceptEitherAsync [d other f executor]
-    (accept-either-async d other f executor))
+    (accept-either d other f (deferred executor)))
 
   (runAfterEither [d other f]
-    (run-after-either d other f))
+    (run-after-either d other f nil))
   (runAfterEitherAsync [d other f]
-    (run-after-either-async d other f))
+    (run-after-either d other f (deferred (cs-default-executor))))
   (runAfterEitherAsync [d other f executor]
-    (run-after-either-async d other f executor))
+    (run-after-either d other f (deferred executor)))
 
   (thenCompose [d f]
-    (then-compose d f))
+    (then-compose d f nil))
   (thenComposeAsync [d f]
-    (then-compose-async d f))
+    (then-compose d f (deferred (cs-default-executor))))
   (thenComposeAsync [d f executor]
-    (then-compose-async d f executor))
+    (then-compose d f (deferred executor)))
 
   (handle [d f]
-    (then-handle d f))
+    (then-handle d f nil))
   (handleAsync [d f]
-    (then-handle-async d f))
+    (then-handle d f (deferred (cs-default-executor))))
   (handleAsync [d f executor]
-    (then-handle-async d f executor))
+    (then-handle d f (deferred executor)))
 
   (exceptionally [d f]
-    (then-exceptionally d f))
-
-  (toCompletableFuture [d]
-    (to-completable-future d))
+    (then-exceptionally d f nil))
+  ;; Only available since Java 12
+  ;; (exceptionallyAsync [d f]
+  ;;   (then-exceptionally d f (deferred (cs-default-executor))))
+  ;; (exceptionallyAsync [d f executor]
+  ;;   (then-exceptionally d (deferred executor)))
 
   (whenComplete [d f]
-    (when-complete d f))
+    (when-complete d f nil))
   (whenCompleteAsync [d f]
-    (when-complete-async d f))
+    (when-complete d f (deferred (cs-default-executor))))
   (whenCompleteAsync [d f executor]
-    (when-complete-async d f executor)))
+    (when-complete d f (deferred executor)))
+
+  (toCompletableFuture [d]
+    (to-completable-future d)))
 
 (definline realized?
   "Returns true if the manifold deferred is realized."
@@ -1538,157 +1545,129 @@
 ;; CompletionStage helper fns
 ;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;
 
-(defmacro ^:no-doc def-async-for
-  "Defines a CompletionStage async version of the function associated with
-   the given symbol, with '-async' appended."
-  [fn-name]
-  (let [async-name (symbol (str (name fn-name) "-async"))]
-    `(defn- ~async-name
-       ([d# f#]
-        (~async-name d# f# (or (ex/executor) (ex/execute-pool))))
-       ([d# f# executor#]
-        (~fn-name (onto d# executor#) f#)))))
-
-(defmacro ^:no-doc def-async-for-dual
-  "Defines a CompletionStage async version of the two-deferred
-   function associated with the given symbol, with '-async' appended."
-  [fn-name]
-  (let [async-name (symbol (str (name fn-name) "-async"))]
-    `(defn- ~async-name
-       ([d# d2# f#]
-        (~async-name d# d2# f# (or (ex/executor) (ex/execute-pool))))
-       ([d# d2# f# executor#]
-        (~fn-name (onto d# executor#) d2# f#)))))
-
-(defn- fmap-deferred
-  "Returns a new deferred with function `f` applies to realized value of `d`.
-   (Like fmap but for deferreds.)
-
-   This function does not unwrap the result of f; it will only be applied to
-   the immediate value of `d`. This is for mimicking CompletionStage's
-   behavior."
+(defn- cs-default-executor []
+  (or (ex/executor) (ex/execute-pool)))
+
+(defn- shallow-connect
+  "Like `connect` but without implicit unwrapping of conveyed value."
+  [from to]
+  (on-realized from
+               (fn [val] (success! to val))
+               (fn [error] (error! to error))))
+
+(defn- shallow-chain
+  "Returns a new deferred with function `f` applied to realized value of `d`.
+
+   Unlike `chain`, this function does not unwrap the result of `f`; it will only be applied to the
+   immediate value of `d`. This is for mimicking `CompletionStage`'s behavior."
   [d f]
   (let [d' (deferred)]
     (on-realized d
                  (fn [val] (success! d' (f val)))
                  (fn [error] (error! d' error)))
     d'))
 
-(defn- then-apply [d ^Function f]
-  (assert-some f)
-  (fmap-deferred d #(.apply f %)))
+(defn- completion-stage-result [d f to]
+  (when to
+    (shallow-connect d to))
+  (shallow-chain (or to d) f))
 
-(def-async-for then-apply)
+(defn- then-apply [d ^Function f to]
+  (assert-some f)
+  (completion-stage-result d #(.apply f %) to))
 
-(defn- then-accept [d ^Consumer c]
+(defn- then-accept [d ^Consumer c to]
   (assert-some c)
-  (fmap-deferred d #(.accept c %)))
-
-(def-async-for then-accept)
+  (completion-stage-result d #(.accept c %) to))
 
-(defn- then-run [d ^Runnable f]
+(defn- then-run [d ^Runnable f to]
   (assert-some f)
-  (fmap-deferred d (fn [_] (.run f))))
-
-(def-async-for then-run)
+  (completion-stage-result d (fn [_] (.run f)) to))
 
-
-(defn- then-combine [d other ^BiFunction f]
+(defn- then-combine [d other ^BiFunction f to]
   (assert-some other f)
-  (fmap-deferred (zip d other)
-                (fn [[x y]] (.apply f x y))))
-
-(def-async-for-dual then-combine)
+  (completion-stage-result (zip d other)
+                           (fn [[x y]] (.apply f x y))
+                           to))
 
-
-(defn- then-accept-both [d other ^BiConsumer f]
+(defn- then-accept-both [d other ^BiConsumer f to]
   (assert-some other f)
-  (fmap-deferred (zip d other)
-                (fn [[x y]] (.accept f x y))))
-
-(def-async-for-dual then-accept-both)
+  (completion-stage-result (zip d other)
+                           (fn [[x y]] (.accept f x y))
+                           to))
 
-
-(defn- run-after-both [d other ^Runnable f]
+(defn- run-after-both [d other ^Runnable f to]
   (assert-some other f)
-  (fmap-deferred (zip d other)
-                (fn [[_ _]] (.run f))))
-
+  (completion-stage-result (zip d other)
+                           (fn [[_ _]] (.run f))
+                           to))
 
-(def-async-for-dual run-after-both)
-
-
-(defn- apply-to-either [d other ^Function f]
+(defn- apply-to-either [d other ^Function f to]
   (assert-some other f)
-  (then-apply (alt d other) f))
-
-(def-async-for-dual apply-to-either)
+  (then-apply (alt d other) f to))
 
-
-(defn- accept-either [d other ^Function f]
+(defn- accept-either [d other ^Function f to]
   (assert-some other f)
-  (then-accept (alt d other) f))
-
-(def-async-for-dual accept-either)
+  (then-accept (alt d other) f to))
 
-
-(defn- run-after-either [d other ^Function f]
+(defn- run-after-either [d other ^Function f to]
   (assert-some other f)
-  (then-run (alt d other) f))
-
-(def-async-for-dual run-after-either)
+  (then-run (alt d other) f to))
 
-
-(defn- then-compose [d ^Function f]
+(defn- then-compose [d ^Function f to]
   (assert-some f)
   (let [d' (deferred)]
-    (on-realized d
-                 (fn [val]
-                   (on-realized (->deferred (.apply f val))
-                                #(success! d' %)
-                                #(error! d' %)))
-                 (fn [error] (error! d' error)))
+    (-> (completion-stage-result d #(->deferred (.apply f %)) to)
+        (on-realized (fn [fd]
+                       (shallow-connect fd d'))
+                     (fn [error]
+                       (error! d' error))))
     d'))
 
-(def-async-for then-compose)
-
-
-(defn- then-handle [d ^BiFunction f]
+(defn- then-handle [d ^BiFunction f to]
   (assert-some f)
+  ;; Can't use `completion-stage-result` here because it only covers
+  ;; the success case.
+  (when to
+    (shallow-connect d to))
   (let [d' (deferred)]
     (on-realized
-     d
+     (or to d)
      (fn [val] (success! d' (.apply f val nil)))
      (fn [error] (success! d' (.apply f nil error))))
     d'))
 
-
-(def-async-for then-handle)
-
-
-(defn- then-exceptionally [d ^Function f]
+(defn- then-exceptionally [d ^Function f to]
   (assert-some f)
+  ;; Can't use `completion-stage-result` here because it only covers
+  ;; the success case.
+  (when to
+    (shallow-connect d to))
   (let [d' (deferred)]
     (on-realized
-      d
-      (fn [val] (success! d' val))
-      (fn [error] (success! d' (.apply f error))))
+     (or to d)
+     (fn [val] (success! d' val))
+     (fn [error] (success! d' (.apply f error))))
     d'))
 
 (defn- to-completable-future [d]
 
-  (let [result (CompletableFuture.)]
+  (let [to (CompletableFuture.)]
 
     (on-realized d
-                 #(.complete result %)
-                 #(.completeExceptionally result %))
+                 #(.complete to %)
+                 #(.completeExceptionally to %))
 
-    result))
+    to))
 
-(defn- when-complete [d ^BiConsumer f]
+(defn- when-complete [d ^BiConsumer f to]
   (assert-some f)
+  ;; Can't use `completion-stage-result` here because it only covers
+  ;; the success case.
+  (when to
+    (shallow-connect d to))
   (let [d' (deferred)]
-    (on-realized d
+    (on-realized (or to d)
                  (fn [val]
                    (try (.accept f val nil)
                         (success! d' val)
@@ -1701,8 +1680,6 @@
                           (error! d' err)))))
     d'))
 
-(def-async-for when-complete)
-
 ;;;
 
 (alter-meta! #'->Deferred assoc :private true)