How about

["test"
 ["" {:post ...}]
 ["/:param" {:get ...}]]

thanks!

(not saying that it'll work BTW - I myself don't use reitit, so it's just a guess based on other code that I've seen)

it worked like a charm 🙂 I still cannot get used to such tricks of clojure, too many java in my head and too few time spent coding in clojure still

From your initial code, /test will definitely run the POST request, same as /test/param cos you must run /test before /test/param But if you make it as @U2FRKM4TW suggested, /test”” will run separately from /test/param That’s how reitit routing works

I would add more queues 🙂

or sort of flip from push to pull

a big question is if the partition ids are all known statically or if partitions can be added dynamically

> a big question is if the partition ids are all known statically or if partitions can be added dynamically In this case it is dynamic! > I would add more queues > or sort of flip from push to pull Oo, I'm intrigued! Would you mind explaining a bit more? Thanks @U0NCTKEV8!

dynamic is much trickier

basically it is a kind of pubsub with a single consumer for each topic (a topic being a partition id)

if you have some pub/sub stuff, you can stick something that starts a single worker for a given topic between the input queue and the pubsub (copying from the input queue to pubsub, starting a worker for a partition if absent)

but removing workers from that is tricky (if you don't want idle workers hanging around because you saw some partition id once hours ago)

That is interesting! I could do something like start a go-loop dynamically for each partition. I could also have the go-loop "shut itself off" if it's queue is empty. Is something like that what you had in mind?

the thing with doing it via workers with multiple queues and locks is it is hard to keep all your workers running, and you end up doing things like having them poll each queue and then sleep for some amount of time if there is nothing nothing on the queues (instead of blocking on a single queue) which works, you just feel bad doing it

if you don't do a non-blocking polling from the worker they get stuck reading from Q1 with no work on it, when Q2 has some work to do on it

(def input (async/chan))
(def next-input (async/chan))
(def work-pub (async/pub next-input part-id))
(def shutting-down (async/chan))

(def start-worker [the-part-id]
  (let [ch (async/sub work-pub the-part-id (async/chan))]
    (async/go-loop [ack-ch nil]
      (let [to (async/timeout! 1000)
            [the-val the-chan] (async/alts! (concat (when-not ack-ch
                                                      [to])
                                                    [ch]
                                                    (when ack-ch
                                                      [ack-ch])))]
        (cond (= the-chan to)
              (let [ack (async/chan)]
                (async/>! shutting-down [part-id ack])
                (recur ack))
              (= the-chan ch)
              (do
                (do-stuff the-val)
                (recur ch (when ack-ch (async/close! ack-ch) nil)))
              (= the-chan ack-ch)
              nil)))))

;; worker launcher
(async/go-loop [workers {} buf nil]
  (when-some? buf
    (async/>! next-input buf))
  (async/alt!
    shutting-down ([[part-id ack]]
                   (if (async/>! ack true)
                     (recur (dissoc workers part-id) buf)
                     (recur workers buf)))
    input ([msg]
           (let [p (part-id msg)]
             (if (contains? workers p)
               (recur workers msg)
               (recur (assoc workers p (start-worker p)) msg))))))

oh, the worker should shutdown its pubsub channel when it exits

and there are resource management issues with async/pub leaking mults and not removing closed channels until a message comes in for the topic it is subscribed to

(we have an alternative async/pub at work that avoids all that)

I'll sit tomorrow morning with fresh eyes and think on this with full focus. Really appreciate the demo code, thanks @U0NCTKEV8!

This is essentially a group-by of streams. Both more.async and #CL85MBPEF have those building blocks Remember you also want to GC idle partitions after a while to avoid oom

I've been playing with both solutions today, and I learned a lot. Thank you @U0NCTKEV8 @UK0810AQ2!

@UK0810AQ2 I have a curious question for you, reading the implementation of group-by! in more.async https://github.com/bsless/more.async/blob/master/src/main/clojure/more/async.clj#L270-L272 Would there be a race condition of sorts, between delayed-death!, and get-chan. where the constraint of "one consumer is active for a given partition at a time" fails? My thinking is: • There may be a split second, when delayed-death closes a channel, though f is still running • get-chan would return a new channel, and run f, which would now mean two f s would run at once for a particular partition-id I realize group-by doesn't specifically support the constraint of "1 f running at a time for a partition id" -- Just wanted to ask to make sure I am understanding the code properly. I would guess I would need to do something more like what hiredman did with the ack-ch, is that right?

I think you're right regarding the race condition, it should be transactional, you could do some locking in the f level to avoid it, too

Gotcha, thanks Ben!

the ack channel in my example above is to avoid that race

Yes! Really cool.

but there is a potential for a deadlock, because the worker launcher is waiting forever for a write to the ack channel

ah, no, the worker closes the ack channel if it decides to keep going

What do you think about this:

(defn- start-worker! [n work-ch done-ch f]
  (loop []
    (let [[k item] (a/<!! work-ch)]
      (when-not (nil? item)
        (f item)
        (a/>!! done-ch [k item])
        (recur)))))

(defn- process-from [{:keys [running] :as state}
                     [k item :as v]]
  (if (running k)
    [nil (update-in state [:pending-work k] (fnil conj []) item)]
    [v (update state :running conj k)]))

(defn- process-done [{:keys [pending-work] :as state} [k item :as v]]
  (let [[next-item & remaining] (pending-work k)]
    (if next-item
      [[k next-item] (assoc-in state [:pending-work k] remaining)]
      [nil (update state :running disj k)])))

(defn- start-scheduler! [state-atom from-ch work-ch done-ch]
  (loop []
    (let [[v ch] (a/alts!! [from-ch done-ch])
          state @state-atom]
      (when-not (nil? v)
        (let [[to-work new-state] (condp = ch
                                    from-ch (process-from state v)
                                    done-ch (process-done state v))]
          (reset! state-atom new-state)
          (when to-work (a/>!! work-ch to-work))
          (recur))))))

(defn start!
  [num-workers from-ch f]
  (let [work-ch (a/chan num-workers)
        done-ch (a/chan)
        workers (->> (range num-workers)
                     (mapv #(future (start-worker! %1 work-ch done-ch f))))
        sched-state (atom {:pending-work {} :running #{}})
        scheduler (future (start-scheduler!
                           sched-state
                           from-ch
                           work-ch
                           done-ch))]
    {:workers workers
     :scheduler scheduler
     :schedule-state sched-state}))

(comment
  (def num-workers 4)
  (def from-ch (a/chan))
  (defn f [x] (println "got! %s" x))
  (def ret (start! num-workers from-ch f)))

I would go with an unrestricted set of go workers that queue work on some executor (possibly on a fixed pool) and wait for it to finish

possibly with additional feedback like watching the queue of the executor and if gets too large wait for some amount of time (using a timeout channel, to yield the go executor thread to service other gos) before trying again

Gotcha. For my own learning, one more curious question: what would be your reasoning? Is it because it makes things simpler down the road? (error handling etc)

it breaks the problem down into two smaller subproblems to solve(and those solutions can be reused) instead of trying to solve one big problem at once

That makes sense to me, thanks for all the help @U0NCTKEV8!

On the jvm clojure, you could muck around with intern, but I would just do:

(defmacro symbolate [a-set]
  `(do
     ~@(for [sym a-set]
         `(def ~sym (quote ~sym)))))

It took me (too much) trial-and-error to get it to work, but your idea sent me down the right path

(def my-set #{'Required 'Optional})

(defmacro symbolate [a-set-sym]
  (let [aset (eval a-set-sym)
        cmds (for [e aset] (list 'def e `'~e))]
    (do cmds)))

(defn -main
  "I don't do a whole lot ... yet."
  [& args]
  (symbolate my-set)
  (println Required)
  (println Optional))

I don’t think the macro is doing quite what you expect. In particular the call to do is not doing anything, as it just returns a sequence like ((def A 'A) (def B 'B)) which achieves the correct result because the seq and defs get evaluated. The last line could be (into '(do) cmds)

yes, you’re right. I modified it to (list ~@cmds)`

getting there drop-by-drop 🙂

That returns the same thing as it did before

well it works for bigger lists

(def my-set
  #{'Local 'In 'Out 'InOut 'ReturnVar 'Unspecified
    })

(defmacro symbolate [a-set-sym]
  (let [a-set (eval a-set-sym)
        cmds (for [e a-set] (list 'def e `'~e))]
    `(list ~@cmds)))

(defn -main
  "I don't do a whole lot ... yet."
  [& args]
  (symbolate my-set)

  (println Local)
  (println In)
  (println Out)
  (println InOut)
  (println ReturnVar)
  (println Unspecified)
  )

The code you want to return is (do (def A 'A) (def B 'B)), but that returns ((def A 'A) ,,,)

ok lemme try it

You can also check it with macroexpand

(defmacro symbolate [a-set-sym]
  (let [a-set (eval a-set-sym)
        cmds (for [e a-set] (list 'def e `'~e))]
    (into '(do) cmds)))

(defn -main
  "I don't do a whole lot ... yet."
  [& args]
  (symbolate my-set)

  (println Local)
  (println In)
  (println Out)
  (println InOut)
  (println ReturnVar)
  (println Unspecified))

=> Syntax error compiling at (macro_question/core.clj:11:12).
Unable to resolve symbol: do in this context

I’ll try macroexpand

Ah sorry, backtick instead of quote. (into (do) cmds)`

Also a correction on your previous example, it's returning (list (def A 'A)) which then evaluates to a list of vars -- which may be what you want!

same result : unable to resolve symbol: do …

yes, i want the side-effects of the defs as well!

so i think the println’s test that the defs have happened

wife calling … back later 🙂 thanks for your kind attention to this!

The list of vars is exactly what i want; this method survived integration with the rest of my compiler IR 🙂

I used YourKit to try to find low-hanging fruit, but all the reports are obfuscated by anonymous function calls without line numbers, and call-graphs are lost at thread boundaries

I sprinkle transducers everywhere just out of personal preference. As far as speeding things up, I've found https://github.com/clojure-goes-fast/clj-async-profiler to very helpful. Depending on the use case, benchmark tools like https://github.com/hugoduncan/criterium are also helpful. Transducers should be generally faster, but it's hard to say how much of a win it will be without profiling. You might also find https://github.com/johnmn3/injest to be helpful.

https://github.com/BrunoBonacci/mulog is another great logging library that lets you capture event-level logging data (including duration) while preserving more context than call graphs alone

In general, unless you have big collections or deep transformation stacks, this is unlikely to help much. Because of chunking, sequences can be faster than you expect, and if you’re not using the entire result of a lazy seq transformation, they can dona lot less work than an eager reduce with transducers. But you know your app - there are probably a small number of places that make the biggest difference, so just think hard or try both in those

In general, using your brain and your tools to find and solve actual problems is going to be more efficient and impactful than randomly doing anything

Thanks yall! It's not my code, just being tasked with optimizing someone else's, which is hard to approach without the context of having written it

In that case async profiler is great. Look for the meatiest stack

I started doing this and got frustrated when they were slower than the conventional sequence operations… even those with lots of steps… probably because I wasn’t dealing with very large sequences. However, it taught me how sometimes breaking up logic into transducers made things more composable… and so I could have greater reuse and legibility… so it wasn’t a wasted effort at all.