Агенты 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, поместили это в Ref, и рабочие агенты потребляли предметы из этого 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))))