You can make a little util like this:

(defn await-io
  [input-coll io-fn n]
  (let [output-chan (a/chan)]
    (a/pipeline-blocking n
                         output-chan
                         (map io-fn)
                         (a/to-chan! input-coll))
    (a/"
  ""
  ""]
 slurp
 10)

this looks cool thank you so much

clojure is just awesome one of the best lisps i ever used and one of the nicest communities thank you !!!

btw, another reason to beware of pmap is it's lazy, so you'll have double trouble when mixing it with side effects, I even wrote https://bsless.github.io/side-effects/ about all the ways this could melt your code

If it’s doing IO, and you’re on JDK 21, the by far easiest thing to do would be to spawn virtual threads and constrain parallelism with a semaphore.

@U06B8J0AJ That still fails to saturate the concurrency though.

By the way, the output chan doesn't need to be buffered, I forgot that into consumes one element at a time. So no buffer needed

Why would it fail to saturate concurrency? @U0K064KQV

@U06B8J0AJ Well, it depends on the detail I guess of exactly how you orchestrate it. But to saturate it, you need a mechanism like alt!! or poll or you need async completion notifications. So that you're not blocked on all the spawned thread completing, or on them completing in the submitted order. But instead that you are blocked on the first one to complete, so that as soon as one of them completes, you can queue up another.

In "Java" style, I would use the ExecutorCompletionService, you can use vthreads with it. And it lets you poll or take which will return the next completed task if any. And at that moment you can submit another, and process the competed one.

pipeline-blocking does this under the hood for us. I have not yet found something simpler than that. In terms of how short the implementation is.

Oh, or maybe you implied, say you have 100k elements to process at 100 concurrency, that you spawn 100k vthread and have them all block behind the semaphore? I hadn't really thought of that. Are vthread that free that you can just use them as a queue?

@U06B8J0AJ Hum , ok so actually, your idea of the semaphore approach I think might be amazing haha. I guess I never really thought of using semaphore before. I don't even think it matters if you use vthreads or real threads.

(defn await-io4
  [input-coll io-fn n]
  (let [sem (Semaphore. n true)
        res (for [e input-coll]
              (do (.acquire sem)
                  (future (try (io-fn e) (finally (.release sem))))))]
    (mapv deref res)))

Pipeline maintains order and requires using channels

This one maintains order as well (I did just edit it, got inspired by the code of pipeline and realized I could maintain order in a more straightforward way)

Some helpers to give virtual threads an interface similar to futures.

It’s true that none of this requires virtual threads, but in the case of IO, there’s no reason not to use them. They’re cheap and plentiful.

Btw, using semaphores is not my idea, but the official recommendation from the JVM dev team. You can have millions of virtual threads with no problem, so when there’s no upper bound set by the thread pool, a semaphore is required.

Promesa has a core.async inspired implementation of CSP for channels etc. on top of virtual threads: https://funcool.github.io/promesa/latest/promesa.exec.csp.html. Experimental though.

Not quite sure if i have virtual threads on babashka. Maybe i've read something wrong

virtual threads and semaphores sounded familiar: https://andersmurphy.com/2024/05/06/clojure-managing-throughput-with-virtual-threads.html

That's the one I should have linked 😛

Or https://inside.java/2024/02/04/sip094/#rate-limiting-virtual-threads

Btw, note cloneThreadBindingFrame . There might be surprises if it’s missing.

I'm still not sure if it's ideal to use them as a queue. Which you're doing in your example. Because you spawn one even for waiting tasks, and have them blocked. I tried to search for info and couldn't really see. Probably for small number of tasks it's fine. But it still seems they consume more memory than queuing up tasks would. And I also couldn't tell if it impacts scheduling in any way if you have thousands or more of them.

That makes a ton of sense. What doesn't though, is if I reverse the order, I can see via my logging that the requests don't make it to the handlers in most cases. They do for /test which is a simple return, but don't for /conversation. Note that I didn't change anything about the handler, and it has a log line on the first sexp just saying that the handler has been hit. I can tell that the authentication step is done for all of them. If I remove the public routes, the private routes all work.

I would double check your logging, and what routes are making requests to

I refactored to this:

(defn app
  [& config]
  (-> (routes public-routes (authenticate-request private-routes))
      (common-middleware config)))

Depending on what your middleware is doing, if it is internally stateful (the session middleware is the most common mw that is) two calls would result in an entirely separate state each, which can break things in odd ways

Nothing is stateful, I am not sophisticated enough to use sessions. I have the following mw: 1. parse url query params and wrap in the body 2. embed context/global variables 3. log out the request 4. handle cors