Inconsistent results with deferred/onto

[kingpong]

I'm trying to ensure that my chained callbacks are always run on a specific thread pool, so I'm using d/onto followed by d/chain to set up the callbacks. What I'm seeing, though, is that sometimes the callbacks will be run directly on the derefing thread. I'm able to reproduce this behavior reliably (clojure 1.8, Java 1.8, manifold 0.1.6).

(require '[manifold.deferred :as d])
(import '(java.util.concurrent ForkJoinPool))

(defn tid []
  (let [t (Thread/currentThread)]
    (format "Thread %d: %s" (.getId t) (.getName t))))

(defn chained-onto-threads
  "Tries n times to run f (which should produce a deferred) then chains the
   result into a callback function that should run on the ForkJoinPool.
   Returns a map containing the frequencies of threads that were actually used."
  [n f]
  (let [forkjoin-pool (ForkJoinPool/commonPool)
        seen-threads  (atom {})]
    (run! deref
          (for [i (range n)]
            (-> (f)
                (d/onto forkjoin-pool)
                (d/chain
                  (fn [_]
                    ;; Expecting this to be ForkJoinPool, but isn't always.
                    (swap! seen-threads update (tid) #(inc (or % 0))))))))
    @seen-threads))

When I run chained-onto-threads with an f that produces a manifold future, sometimes the chained function runs on an nREPL thread, not a ForkJoinPool thread as expected:

user=> (require '[clojure.pprint :as pp])
nil
user=> (pp/pprint (chained-onto-threads 1000 #(d/future "any value")))
{"Thread 33: nREPL-worker-3" 7,
 "Thread 35: ForkJoinPool.commonPool-worker-0" 142,
 "Thread 37: ForkJoinPool.commonPool-worker-1" 123,
 "Thread 38: ForkJoinPool.commonPool-worker-2" 160,
 "Thread 39: ForkJoinPool.commonPool-worker-3" 144,
 "Thread 40: ForkJoinPool.commonPool-worker-4" 143,
 "Thread 43: ForkJoinPool.commonPool-worker-5" 163,
 "Thread 45: ForkJoinPool.commonPool-worker-6" 118}

With an f that produces a success-deferred, the chained function always runs on an nREPL thread, not on the executor that was specified:

user=>   (pp/pprint
  #_=>     (chained-onto-threads 1000 #(d/success-deferred "any value")))
{"Thread 15: nREPL-worker-1" 1000}

Can you shed any light on what's happening? Am I doing something wrong or is this a bug?

[dm3]

I'd say it's an inconsistency. This stems from

https://github.com/ztellman/manifold/blob/193f5f48972c8e20dd0a9fc41a1311566a9f7bdd/src/manifold/deferred.clj#L378 - when state of deferred is realized, the listener execution happens immediately on the thread which attaches the listener.

[kingpong]

Here's a more practical example of why this is a problem:

  (let [d  (-> 100
               (d/success-deferred)
               (d/onto (ForkJoinPool/commonPool)))
        t0 (System/currentTimeMillis)
        _  (d/chain d #(Thread/sleep %))  ;; want to run in parallel with ↓
        _  (d/chain d #(Thread/sleep %))  ;; want to run in parallel with ↑
        t1 (System/currentTimeMillis)]
    (println "Chaining took" (- t1 t0) "ms"))

; ==> Chaining took 209 ms

I would have expected the calls to d/chain to be async, and the sleeps to run in parallel, on the ForkJoinPool. But instead they run in sequence and block each other.

It's also a problem if the code in the chained callbacks requires thread-local context that is provided by the executor, for example MDC.

[kingpong]

@dm3 since I've explicitly set the executor with (d/onto ...), doesn't the anonymous function you mention run on the executor (line 384)?
https://github.com/ztellman/manifold/blob/eabe63e42af4ee9454751ad272fb418701e90f53/src/manifold/deferred.clj#L376-L385

[ztellman]

As @dm3 said, if you chain on a deferred which is already realized, it executes immediately and does not register any sort of callback. Since you are simply using (d/future "a value") without sleeping or anything, at least some of the time the future has executed before being passed into d/chain.

[kingpong]

@ztellman Is this the desired behavior, though? In my second example, code that I would expect not to block at all actually takes 200ms. Imagine that this is happening in a UI thread, and I want to ensure that the callbacks, which may block, are run in a separate thread to avoid hanging the UI. The initial deferred came from a library function, and it may (or may not) be realized by the time it gets to me.

[ztellman]

It's a fair question. This code is effectively using deferreds as a way to convey blocking work onto another thread, which I would accomplish with d/future-with, but I can see why you might expect that would be done for you. Let me give it some thought.

[kingpong]

Thanks for considering. The actual problem that is biting me is that the code in my callbacks requires thread-local context that is automatically provided by the executor I specified, and the thread-local context isn't always there. I think the unexpected blocking case is more compelling to most users of the library, though.

[ztellman]

Wrapping the chain call in future-with will fix your immediate issue, fwiw. I’ll update once I’ve mulled this over a bit.

…

On Fri, Feb 23, 2018 at 2:40 PM Philip Garrett ***@***.***> wrote: Thanks for considering. The actual problem that is biting me is that the code in my callbacks requires thread-local context that is automatically provided by the executor I specified, and the thread-local context isn't always there. I think the unexpected blocking case is more compelling to most users of the library, though. — You are receiving this because you were mentioned. Reply to this email directly, view it on GitHub <#151 (comment)>, or mute the thread <https://github.com/notifications/unsubscribe-auth/AAB6P0tVAcJ23ke213AXYO4dIdsHxyllks5tXxRGgaJpZM4SOehO> .

[mping]

@ztellman any update on this?

[ferdinand-beyer]

I've stumbled over this as well, trying to use an executor to set up a thread-local context.

On my machine, this snippet constantly yields the REPL thread, so the executor passed to onto is essentially ignored:

(-> (d/future 1)
    (d/onto (executor/fixed-thread-executor 2))
    (d/chain (fn [_] (.getName (Thread/currentThread)))))

This means that the onto examples in the documentation are mostly wrong (probably depending on the computer).

I guess this is not really an issue if all one cares about is fast execution, but this means that one cannot use custom executors for other purposes.

@KingMob -- if you decide this is not a bug, maybe we should document this, stating that there is no guarantee? The current docstring Returns a deferred whose callbacks will be run on executor. is definitely misleading.

[arnaudgeiser]

Yes, if the deferred is already realized, it will continue on the same thread and the specified Executor (onto) won't be used.
Here is a bit of context on Aleph regarding this behaviour (I'm not saying the documentation cannot be improved).

clj-commons/aleph#427 (comment)

[arnaudgeiser]

(-> (d/future 1)
    (d/onto (e/fixed-thread-executor 2))
    (d/chain (fn [_] (prn (.getName (Thread/currentThread))))))

=> "nREPL-session-8ba7d6cf-90f9-428b-9505-bba48850ff05"
(or sometimes if unlucky) => "manifold-execute-3"

(-> (d/future (Thread/sleep 100) 1)
    (d/onto (e/fixed-thread-executor 2))
    (d/chain (fn [_] (prn (.getName (Thread/currentThread))))))

=> "manifold-pool-37-1"