How does one clearly structure dependencies between core.async channels?

Question

Let's say I have a corpus of computations that I want to run asynchronously using core.async, but unfortunately a few of the functions depend on the output of other functions. How do I go about structuring this cleanly in my code, while also getting the best performance?

A few potential solutions I've come across are

• Prismatic's Graph - seems reasonable, although I haven't tested it with core.async channels; the fact that it requires the use of fnk is a little off-putting for me because it requires buying into their DSL for function definitions, but if that's the best solution then I don't mind.
• Javelin cells - only for ClojureScript (currently) and uses FRP instead of CSP as the implementation, but it does a very good job of modeling dependencies among computations via formula cells.
• Onyx - made for distributed computation (as a competitor to Apache Storm, etc) but has a "workflow" abstraction that handles dependencies between computations and works with core.async. This seems like the closest fit to my problem domain, but I'm not sure if I need the overhead of all the cluster management features.

What's the canonical solution for this problem?

Edit: added Onyx

Answer 1

This question is kind of hard to answer because your question lacks specifics about your use case. Libraries like Graph, Javelin and Onyx all have different use cases that go beyond just making computations depend on each other.

If you would just like to have a thread or go block depend on results generated in another part of your system, I would suggest just using the core.async primitives without any additional libraries.

The most basic solution to making execution wait for another thread of activity is using blocking takes when taking values from channels. This will halt the thread (or go block) when no values are available on that channel.

As you can see in the following example, making a computation depend upon an activity done in another thread is very easy.

(let [c (chan)]
  (thread (>!! c “hello”))
  (assert (= “hello” (<!! c)))
  (close! c)

There are also more elaborate mechanisms available. The Alts!! function provides the ability to wait on many channels at the same time. Several different flavours of the pipeline function allow you to model concurrency in a dataflow like manner.

Are there any specific problems you run into that can not be expressed clearly using the build-in functions?

Answer 2

I don't think there is a canonical way to solve it, core.async is so new that few people have given it a shot. If I were to choose between your three options I'd go with Graph, it's been deployed and tested in production for a while, and you don't need Clojurescript to run it. If you're interested in a FRP solution take a look at Java Reactive Extensions, Clojure bindings for it exist in RxClojure.