Is there a function in core.async that lets me put n values in n channels? Something like

(defn multi->!! [channel-val-pairs]
  (loop [cvs channel-val-pairs]
    (when (seq cvs)
      (let [[_ c] (async/alts!! cvs)]
        (recur (filter #(not= c (first %)) cvs))))))

If you squint at it right, that's a mult

Or not, I guess a mult would be the same message

something like

(defn multi [channel-val-pairs]
  (let [c (async/chan (count channel-val-pairs))]
    (doseq [[c v] channel-val-pairs]
      (async/put! c v (fn [_] (async/put! c true))))
    (async/go
      (dotimes [i (count channel-val-pairs)]
        (async/<! c)))))

I think one reason multi might not exist is that there are many different options for backpressure and timeouts when you're putting on multiple channels with multiple values that will likely depend on the use case

yeah, if you sort of flip it and use a pub those choices can be easier

each subscription can do their own timeoutes/buffer policies, etc

core.async, and a lot of csp inspired libraries are actually kind of primitive about building operations like this

where alt is a disjunction of operations on channels, what you want is a conjunction of operations on channels

and core.async doesn't have a way to build atomic conjunctions of operations

thanks, my case is actually just 2 channels, so I wonder if the extra "overhead" of multi is actually warranted or if I should just do alts! followed by >!

> atomic conjunctions of operations intuitively, that seems like it would be easy to encourage deadlock

there is a great thesis from a few years ago (https://www.ccs.neu.edu/home/turon/thesis.pdf) which gives you transactions over both refs (stm) and channels (csp like)

ie. what happens if one or both channels fall behind?

should that block progress for both channels or just one?

should there be timeouts?

The easy version is just to async/put! for each channel, but that's just kicking the can down the road with respect to thinking about how the system responds to load and errors

Yes put! is not an option and both >! need to succeed.

If that's the case, then I think your implementation looks pretty good. It might not matter for your specific use case, but your multi->!! implementation will have unexpected results if someone ever pass multiple operations on the same channel.

I was thinking something like:

(defn multi->!! [channel-val-pairs]
  (loop [ops (set channel-val-pairs)]
    (when (seq ops)
      (let [op (async/alts!! ops)]
        (recur (disj ops op))))))

should be a map not a set

it should be a bag

a map will guarantee unique channels not depending on the value to be put

{ch value}

well, in my head, the abstraction of "do these operations and return when all of them complete" is more useful than "do these operations with the restriction that at most one value can be written to each channel and return when all of them are complete"

but you could do a version with the map, but I would probably call it something else

you could even pass a map

or a set

the set version is broken for that then (which I guess is why you suggested a bag)

or someone could write their own async library that includes atomic conjunctions of operations

More of a general question, what are your go-to tools for figuring out if and where your system is experiencing too much backpressure? My way often just comes down to lots of logging and doesn't seem very efficient, so I wonder what other people are doing. Is there a library to visualize things? I also wonder if there has been any work done in the direction of autoscaling the different parts of a system when it happens?

it isn't so much "too much backpressure" the backpressure is always right, it is a question of where are my bottlenecks

the backpressure is a signal from the slower parts of the system to slow down the faster parts so the slower parts are not overwhelmed. so the thing is identifying the slower parts

a lot of which are kind of a "no duh" like io to disk

I guess the questions I would have are: • where is the backpressure coming from? I/O, other systems, CPU? • is this a distributed system?

For general performance stuff, http://clojure-goes-fast.com/blog/ has some good info

I really like https://github.com/clojure-goes-fast/clj-async-profiler

searches for bottlenecks are usually iterative, at any given moment there is one bottleneck, and if you make that not the bottleneck something else is now the bottleneck

@finn.volkel you may find this talk interesting - https://www.youtube.com/watch?v=r-TLSBdHe1A - and if that peaks your interest, you may want to look into the coz and jcoz causal profilers. I think in the future as a community we could try to get something more Clojure-specific (similar to the clojure-goes-fast tooling) that can do this kind of profiler analysis on Clojure code with more friendly REPL tooling (e.g. based on forms and/or functions, and not lines of code). Well, at least, one can dream :)

there is a jvm variant of the impl in that talk, so should theoretically work on Clojure

at least one of the problems he describes was actually found on the jvm with an initial env that was sensitive to a java system property

but in practice, I have not found any of that to be necessary to understand or diagnose problem with the tools we have

there is no substitute for having a model (mental at least) of your system under load with tools to validate your expectations

the stupidest technique I know is also one of the most effective - load your system, then take 10 thread dumps. whatever threads show up as RUNNING the most is the bottleneck