handle backpressure

[ikitommi]

How to solve backpressure with Sieppari. Related:

• http://aleph.io/manifold/rationale.html
• https://funcool.github.io/catacumba/latest/#asynchronous
• Websocket handlers should handle flow control/backpressure asynchronously pedestal/pedestal#497

[ericnormand]

To do this, you definitely shouldn't use future. It potentially creates a new thread per invocation. The threads are reused, but there is no maximum.

One possibility is to use the same model as core.async: go processes are executed on a fixed size thread pool. They are intended for lightweight work, mostly taking something off one channel and putting it on another. If you need to do heavy CPU work or blocking IO, you shouldn't do it in a go process, but instead, use some thread you control.

As far as Sieppari is concerned, this would mean don't use future. Create your own thread pool for handling the async portion of the interceptor machinery. Then suggest a discipline to the users of the library that they can do lightweight work in the interceptors, but if they have to do heavy CPU work or blocking IO, they need to do it in another thread.

[ericnormand]

Well, maybe I'm wrong about the first paragraph:

These pools will typically improve the performance of programs that execute many short-lived asynchronous tasks.

From https://docs.oracle.com/javase/8/docs/api/java/util/concurrent/Executors.html#newCachedThreadPool--

That is what future uses.

I still believe the other two paragraphs are correct. Long-running tasks in the threadpool will be problematic.

The other thing that I've noticed in the docs is that it's an unbounded queue, which is generally a bad idea. An easy way to know that your threads are overloaded is to have a fixed size queue. When it fills up, you can't add anything more, and that's a signal that you're overloaded.

[ericnormand]

But here's the issue: (future @p) is not a short-lived task. If p takes a while to be realized, the future's thread will block the whole time. I think a better primitive is needed than future. Something like Manifold's deferred.

(continue [c f] (future (f @c)))

should be

(continue [c f] (d/chain c f))

[miikka]

@ericnormand, at the moment Sieppari supports choosing your own async library - there's support for Manifold, core.async and Promesa in addition to using future. Which of course make this issue a bit weird, because I don't think that you can "solve" backpressure once for all the libraries.

[ericnormand]

Yes, I agree. But what I was saying was that Sieppari uses future for all async calls, and I don't think it's possible to have backpressure using future. If you want backpressure, you may have to use something else.

However, I'm still researching it. It might not be as bad as I thought.

[ericnormand]

So, I've done a bit more research:

Currently, on the develop branch, Sieppari creates 4 threads per async request, with no bounds, using future. This is worse than creating a single thread per request and having that thread block.

You probably already know this, but just so we're all on the same page, I'll bring it up. The whole point of asynchronous operation is to not require a thread per request. Asynchrony allows you to handle many requests with a relatively small, finite number of threads.

[miikka]

Right, the async implementation is even more broken than I realized.

W.r.t to using future for all async calls: How it's supposed to work is that if you return e.g. a Manifold deferred, the async calls are done with manifold.deferred/chain. This seems like the right thing to do. The support for Promesa and core.async are similar. However, it can go wrong in at least two ways:

• If you haven't required sieppari.async.manifold, Sieppari will quietly use the future-using AsyncContext implementation for clojure.lang.IDeref (which Deferred is an instance of).
• Even if you require sieppari.async.manifold, you can't tell Clojure which protocol implementation it should prefer so you may still get the future-using implementation for clojure.lang.IDeref. See e.g. CLJ-1807. Edit: Okay, this is not problem for Manifold since the AsyncContext is implemented for manifold.deferred.Deferred which is a concrete deftype. It could be a problem if it was an interface as is the case for Promesa and core.async. I think? Did not try it in practice.

I don't think it is in Sieppari's scope to create yet another async computation library, so maybe the right thing to do is to remove the future support. (Personally I consider the use of future in production code as a code smell, because like you say, it's so hard to enough have control over them.)

[ikitommi]

Both Protocols and MultiMethods are mutable global state and should not be used here. I propose a separate creation/compilation stage, where you explicitely define what kind of runner want to be created. Things to configure, at least:

• support for dynamic queues -> static chains can be optimized for stellar perf. e.g. a chain of just :enters with Manifold could be evaluated with just d/chain'. Only enters with CompletionStage would be just a chain of java functions, e.g. "java-fast".
• which async-libraries are supported, either just one or a set of those

default the support on JVM only to java.util.concurrent.CompletionStage. Both pohjavirta and porsas lean on these and you can use libs like promesa and auspex with it.

could look something like this (with fn->enter mapping and syntax ideas from #15):

(def runner1 (s/runner)) ;; defaults
(def runner2 (s/runner {:dynamic-queue false, :chain sieppari.core_async/chain}))

(defn async-interceptor [k f n]
  #(a/go (update % k (fnil f 0) n)))

(-> (s/context runner2)
    (s/enqueue [(async-interceptor :n + 10) (async-interceptor :n * 2)])
    (s/run)
    (s/await)
    :n)
; 20

[ericnormand]

I really like the idea of a compilation stage, but it doesn't work with the interceptor model of computation. Interceptors allow you to add things to the interceptor queue within the interceptors. I've often thought this was a mistake. As far as I can tell, that was built into Pedestal as a clever way to do routing. A statically known interceptor chain which can be compiled and otherwise optimized would be much faster and easier to understand. It might also have exploitable algebraic properties, like being a monoid. That said, I think the implementation would get hairy. There's quite a lot of logic about what to do next. It's not quite as simple as having a success callback and an error callback.

…

On Sun, Sep 22, 2019 at 2:30 PM Tommi Reiman ***@***.***> wrote: Both Protocols and MultiMethods are mutable global state and should not be used here. I propose a separate creation/compilation stage, where you explicitely define what kind of runner want to be created. Things to configure, at least: - support for dynamic queues -> static chains can be optimized for stellar perf. e.g. a chain of just :enters with Manifold could be evaluated with just d/chain'. Only enters with CompletionStage would be just a chain of java functions, e.g. "java-fast". - which async-libraries are supported, either just one or a set of those - default the support on JVM only to java.util.concurrent.CompletionStage. Both pohjavirta <https://github.com/metosin/pohjavirta> and porsas <https://github.com/metosin/porsas> lean on these and you can use libs like promesa <https://github.com/funcool/promesa> and auspex <https://github.com/mpenet/auspex> with it. could look something like this (with fn->enter mapping and syntax ideas from #15 <#15>): (def runner1 (s/runner)) ;; defaults (def runner2 (s/runner {:dynamic-queue false, :chain sieppari.core_async/chain})) (defn async-interceptor [k f n] #(a/go (update % k (fnil f 0) n))) (-> (s/context runner2) (s/enqueue [(async-interceptor :n + 10) (async-interceptor :n * 2)]) (s/run) (s/await) :n); 20 — You are receiving this because you were mentioned. Reply to this email directly, view it on GitHub <#9?email_source=notifications&email_token=AAA4LTG7X4L7H3EOTJWGEFDQK7BVTA5CNFSM4FS5QNU2YY3PNVWWK3TUL52HS4DFVREXG43VMVBW63LNMVXHJKTDN5WW2ZLOORPWSZGOD7JNITI#issuecomment-533910605>, or mute the thread <https://github.com/notifications/unsubscribe-auth/AAA4LTCN7KBVV6IRIVK365LQK7BVTANCNFSM4FS5QNUQ> .

[nilern]

If you haven't required sieppari.async.manifold, Sieppari will quietly use the future-using AsyncContext implementation for clojure.lang.IDeref (which Deferred is an instance of).

This is terrible.