Clojure агенти консумират от опашка

Това, което искате, е начин да продължите да раздавате задачи, но с някаква горна граница. Един прост подход към това е да се използва семафор за координиране на ограничението. Ето как бих подходил:

(let [limit (.availableProcessors (Runtime/getRuntime))
      ; note: you might choose limit 20 based upon your problem description
      sem (java.util.concurrent.Semaphore. limit)]
  (defn submit-future-call
    "Takes a function of no args and yields a future object that will
    invoke the function in another thread, and will cache the result and
    return it on all subsequent calls to deref/@. If the computation has
    not yet finished, calls to deref/@ will block. 
    If n futures have already been submitted, then submit-future blocks
    until the completion of another future, where n is the number of
    available processors."  
    [#^Callable task]
    ; take a slot (or block until a slot is free)
    (.acquire sem)
    (try
      ; create a future that will free a slot on completion
      (future (try (task) (finally (.release sem))))
      (catch java.util.concurrent.RejectedExecutionException e
        ; no task was actually submitted
        (.release sem)
        (throw e)))))

(defmacro submit-future
  "Takes a body of expressions and yields a future object that will
  invoke the body in another thread, and will cache the result and
  return it on all subsequent calls to deref/@. If the computation has
  not yet finished, calls to deref/@ will block.
  If n futures have already been submitted, then submit-future blocks
  until the completion of another future, where n is the number of
  available processors."  
  [& body] `(submit-future-call (fn [] ~@body)))

#_(example
    user=> (submit-future (reduce + (range 100000000)))
    #<core$future_call$reify__5782@6c69d02b: :pending>
    user=> (submit-future (reduce + (range 100000000)))
    #<core$future_call$reify__5782@38827968: :pending>
    user=> (submit-future (reduce + (range 100000000)))
    ;; blocks at this point for a 2 processor PC until the previous
    ;; two futures complete
    #<core$future_call$reify__5782@214c4ac9: :pending>
    ;; then submits the job

С това сега просто трябва да координирате как се изпълняват самите задачи. Изглежда, че вече разполагате с механизмите за това. Цикъл (изпращане-бъдеще (процес-опашка-елемент))

Може би бихте могли да използвате функцията seque? Цитирам (doc seque):

clojure.core/seque
([s] [n-or-q s])
  Creates a queued seq on another (presumably lazy) seq s. The queued
  seq will produce a concrete seq in the background, and can get up to
  n items ahead of the consumer. n-or-q can be an integer n buffer
  size, or an instance of java.util.concurrent BlockingQueue. Note
  that reading from a seque can block if the reader gets ahead of the
  producer.

Това, което имам предвид, е мързелива последователност, получаваща елементи от опашката по мрежата; бихте опаковали това в seque, поставили това в реф и агентите-работници да консумират елементи от това seque. seque връща нещо, което изглежда точно като обикновен сек от гледна точка на вашия код, като магията на опашката се случва по прозрачен начин. Обърнете внимание, че ако последователността, която поставите вътре, е разделена на парчета, тогава тя все още ще бъде принудена парче по едно. Също така имайте предвид, че самото първоначално извикване на seque изглежда блокира, докато не бъдат получени първоначален елемент или два (или парче, в зависимост от случая; мисля, че това е по-скоро свързано с начина, по който работят мързеливите последователности, отколкото със самия seque).

Скица на кода (наистина схематичен, изобщо не е тестван):

(defn get-queue-items-seq []
  (lazy-seq
   (cons (get-queue-item)
         (get-queue-items-seq))))

(def task-source (ref (seque (get-queue-items-seq))))

(defn do-stuff []
  (let [worker (agent nil)]
    (if-let [result
             (dosync
               (when-let [task (first @task-source)]
                (send worker (fn [_] (do-stuff-with task)))))]
      (do (await worker)
          ;; maybe do something with worker's state
          (do-stuff))))) ;; continue working

(defn do-lots-of-stuff []
  (let [fs (doall (repeatedly 20 #(future (do-stuff))))]
    fs)))

Всъщност вероятно бихте искали по-сложен производител на последователността на елемент от опашката, така че да можете да го помолите да спре да произвежда нови елементи (необходимост, ако цялото нещо трябва да може да бъде затворено елегантно; фючърсите ще умрат, когато задачата източникът работи на сухо, използвайте future-done?, за да видите дали вече са го направили). И това е нещо, което виждам на пръв поглед... Сигурен съм, че има още неща за излъскване тук. Мисля обаче, че общият подход ще работи.

Не съм сигурен колко идиоматично е това, тъй като все още съм начинаещ с езика, но следното решение работи за мен:

(let [number-of-messages-per-time 2
      await-timeout 1000]
  (doseq [p-messages (partition number-of-messages-per-time messages)]
    (let [agents (map agent p-messages)]
      (doseq [a agents] (send-off a process))
      (apply await-for await-timeout agents)
      (map deref agents))))