puniverse/pulsar

`Deref clojure.core/promise` in fiber outputs stream of warnings

andreasthoelke opened this issue · 3 comments

andreasthoelke commented

Derefing a clojure.core/promise in a fiber

(def p1 (clojure.core/promise))
(def f1 (fiber (deref p1)))

starts outputting this warning repeatedly:

WARNING: fiber Fiber@10000050:fiber-10000050[task: ParkableForkJoinTask@71e54873(Fiber@10000050), target: co.paralleluniverse.pulsar.ClojureHelper$4@4010beaf, scheduler: co.paralleluniverse.fibers.FiberForkJoinScheduler@23eede1b] is blocking a thread (Thread[ForkJoinPool-default-fiber-pool-worker-5,5,main]).
        at sun.misc.Unsafe.park(Native Method)
        at java.util.concurrent.locks.LockSupport.park(LockSupport.java:175)
        at java.util.concurrent.locks.AbstractQueuedSynchronizer.parkAndCheckInterrupt(AbstractQueuedSynchronizer.java:836)
        at java.util.concurrent.locks.AbstractQueuedSynchronizer.doAcquireSharedInterruptibly(AbstractQueuedSynchronizer.java:997)
        at java.util.concurrent.locks.AbstractQueuedSynchronizer.acquireSharedInterruptibly(AbstractQueuedSynchronizer.java:1304)
        at java.util.concurrent.CountDownLatch.await(CountDownLatch.java:231)
        at clojure.core$promise$reify__7005.deref(core.clj:6823)
        at clojure.core$deref.invokeStatic(core.clj:2228)
        at clojure.core$deref.invoke(core.clj:2214)
        at rt_comm.connect_auth$fn__44050$fn__44051.invoke(connect_auth.clj:175)
        at co.paralleluniverse.pulsar.InstrumentedIFn.invoke(InstrumentedIFn.java:32)
        at co.paralleluniverse.pulsar.ClojureHelper.suspendableInvoke(ClojureHelper.java:218)
        at co.paralleluniverse.pulsar.ClojureHelper$4.run(ClojureHelper.java:205)
        at co.paralleluniverse.fibers.Fiber.run(Fiber.java:1072)
...

until the promise is resolved:

(deliver p1 123)
(join f1)
> 123

I have the same situation when using a manifold/deferred:

(def d1 (manifold.deferred/deferred))
(def f1 (fiber (deref d1)))

This starts printing a stream of similar warnings:

WARNING: fiber Fiber@10000034:fiber-10000034[task: ParkableForkJoinTask@5398e323(Fiber@10000034), target: co.paralleluniverse.pulsar.ClojureHelper$4@726f386b, scheduler: co.paralleluniverse.fibers.FiberForkJoinScheduler@23eede1b] is blocking a thread (Thread[ForkJoinPool-default-fiber-pool-worker-5,5,main]).
        at sun.misc.Unsafe.park(Native Method)
        at java.util.concurrent.locks.LockSupport.park(LockSupport.java:175)
        at java.util.concurrent.locks.AbstractQueuedSynchronizer.parkAndCheckInterrupt(AbstractQueuedSynchronizer.java:836)
        at java.util.concurrent.locks.AbstractQueuedSynchronizer.doAcquireSharedInterruptibly(AbstractQueuedSynchronizer.java:997)
        at java.util.concurrent.locks.AbstractQueuedSynchronizer.acquireSharedInterruptibly(AbstractQueuedSynchronizer.java:1304)
        at java.util.concurrent.CountDownLatch.await(CountDownLatch.java:231)
        at manifold.deferred.Deferred.deref(deferred.clj:439)
        at clojure.core$deref.invokeStatic(core.clj:2228)
        at clojure.core$deref.invoke(core.clj:2214)
        at rt_comm.connect_auth$eval43600$fn__43601.invoke(connect_auth.clj:175)
        at co.paralleluniverse.pulsar.InstrumentedIFn.invoke(InstrumentedIFn.java:32)
        at co.paralleluniverse.pulsar.ClojureHelper.suspendableInvoke(ClojureHelper.java:218)
        at co.paralleluniverse.pulsar.ClojureHelper$4.run(ClojureHelper.java:205)
        at co.paralleluniverse.fibers.Fiber.run(Fiber.java:1072)

While the fiber still returns the resolved value just fine:

(d/success! d1 123)
(join f1)
> 123

Using Pulsar and Manifold in conjunction would allow to use Pulsar with Aleph effectively and would allow to combine the strengths of Pulsars promises with Manifolds features.

E.g. this works fine already:

(def p1 (pulsar.core/promise))

(def r1 (-> (p/promise #(inc @p1))
            (d/chain inc #(/ 1 %))
            (d/catch (fn [e] :err))))

(deliver p1 -2)
(deref r1)
> :err

But using (def p1 (clojure.core/promise)) or (def p1 (manifold.dererred/dererred)) instead of (def p1 (pulsar.core/promise)) will produce a stream of warnings, e.g. (def p1 (clojure.core/promise)) produces this warning:

WARNING: fiber Fiber@10000055:fiber-10000055[task: ParkableForkJoinTask@75d03a22(Fiber@10000055), target: co.paralleluniverse.strands.SuspendableUtils$VoidSuspendableCallable@39043fc3, scheduler: co.paralleluniverse.fibers.FiberForkJoinScheduler@23eede1b] is blocking a thread (Thread[ForkJoinPool-default-fiber-pool-worker-5,5,main]).
        at sun.misc.Unsafe.park(Native Method)
        at java.util.concurrent.locks.LockSupport.park(LockSupport.java:175)
        at java.util.concurrent.locks.AbstractQueuedSynchronizer.parkAndCheckInterrupt(AbstractQueuedSynchronizer.java:836)
        at java.util.concurrent.locks.AbstractQueuedSynchronizer.doAcquireSharedInterruptibly(AbstractQueuedSynchronizer.java:997)
        at java.util.concurrent.locks.AbstractQueuedSynchronizer.acquireSharedInterruptibly(AbstractQueuedSynchronizer.java:1304)
        at java.util.concurrent.CountDownLatch.await(CountDownLatch.java:231)
        at clojure.core$promise$reify__7005.deref(core.clj:6823)
        at clojure.core$deref.invokeStatic(core.clj:2228)
        at clojure.core$deref.invoke(core.clj:2214)
        at rt_comm.connect_auth$fn__44181.invokeStatic(connect_auth.clj:195)
        at rt_comm.connect_auth$fn__44181.invoke(connect_auth.clj:195)
        at co.paralleluniverse.pulsar.InstrumentedIFn.invoke(InstrumentedIFn.java:32)
        at co.paralleluniverse.pulsar.ClojureHelper.suspendableInvoke(ClojureHelper.java:218)
        at co.paralleluniverse.pulsar.ClojureHelper$4.run(ClojureHelper.java:205)
        at co.paralleluniverse.strands.dataflow.Val$1.run(Val.java:68)
        at co.paralleluniverse.strands.SuspendableUtils$VoidSuspendableCallable.run(SuspendableUtils.java:44)
        at co.paralleluniverse.strands.SuspendableUtils$VoidSuspendableCallable.run(SuspendableUtils.java:32)
        at co.paralleluniverse.fibers.Fiber.run(Fiber.java:1072)
pron commented

Those warnings are true, though: you really are blocking a thread. This is why we have pulsar.core/promise. What behavior would you like to see? The only solution, AFAICT, is to have that other library (manifold?) integrate Pulsar, so that, if running in a fiber, would only block the fiber.

andreasthoelke commented

Ah thanks, it makes sense now. I guess I was somehow thinking that derefing even non-pulsar things in a fiber would still only block that fiber/Strand! So this warning is in fact super useful – I just overlooked that critical part “..is blocking a thread” in the long string.
In any case I think I can just link Aleph/Manifold to Pulsar on the callback level, e.g.

(defn websocket-handler [request]
  (-> (aleph.http/websocket-connection request)
      (manifold.deferred/on-realized (fn [user-socket]
                                       (,,,)))))

it will just not be as smooth. Will explore this a bit further and potentially propose an integration at the Manifold repo.

pron commented

I should note that it's trivial to transform an asynchronous API to a fiber-blocking one using await. Transforming a thread-blocking API to a fiber-blocking one requires a more work, and usually the cooperation of the library's author.

When integrating third-party libraries, you then have two options:

  1. Use await to recreate a convenient synchronous API on top of the library's async API.
  2. Use channels (that work on threads and fibers alike) to communicate between a thread and a fiber.