first cut of broadcast system

Author: richhickey

src/main/clojure/clojure/core/async/flow.clj

@@ -33,6 +33,17 @@
   details. The flow configuration provides a centralized place for
   policy decisions regarding process settings, threading, buffering etc.
 
+  Flow also provides a subsystem for broadcast communication of
+  out-of-band messages without explicit connections or
+  declarations. This could for example be used to communicate the
+  passage of (real or virtual) time. Broadcast messages are associated
+  with (otherwise undeclared) signal-ids, and will be received by
+  processes selecting those ids. Broadcasts messages will arrive along
+  with messages from process inputs, so signal-ids must not conflict
+  with any process input-id. Thus namespaced keywords, UUIDs etc or
+  tuples thereof are recommended as signal-ids. See process
+  describe/transform and inject below for details.
+  
   It is expected that applications will rarely define instances of the
   process protocol but instead use the API function 'process' that
   implements the process protocol in terms of calls to ordinary
@@ -71,16 +82,22 @@
 
   :proc - a function that starts a process
   :args - a map of param->val which will be passed to the process ctor
-  :chan-opts - a map of in-or-out-id->{:keys [buf-or-n xform]}, where buf-or-n
+  :chan-opts - a map of io-id->{:keys [buf-or-n xform]},
+               where io-id is an input/output name, and buf-or-n
                and xform have their meanings per core.async/chan
-               the default is {:buf-or-n 10}
+               the default for inputs and outputs is {:buf-or-n 10}
   
   :conns - a collection of [[from-pid outid] [to-pid inid]] tuples.
 
   Inputs and outputs support multiple connections. When an output is
-  connected multiple times every connection will get every message,
-  as per a core.async/mult.
+  connected multiple times every connection will get every message, as
+  per a core.async/mult. Note that non-multed outputs do not have
+  corresponding channels and thus any chan-opts will be ignored.
 
+  Broadcast signals are conveyed to a process via a channel with an
+  async/sliding-buffer of size 100, thus signals not handled in a
+  timely manner will be dropped in favor of later arriving signals.
+  
   :mixed-exec/:io-exec/:compute-exec -> ExecutorService
   These can be used to specify the ExecutorService to use for the
   corresonding workload, in lieu of the lib defaults.
@@ -136,9 +153,11 @@
   (g/ping-proc g pid timeout-ms))
 
 (defn inject
-  "asynchronously puts the messages on the channel corresponding to the
-  input or output of the process, returning a future that will
-  complete when done."
+  "asynchronously puts the messages on the channel corresponding to
+  the input or output of the process, returning a future that will
+  complete when done. You can broadcast messages on a signal using the
+  special coord [::flow/cast a-signal-id]. Note that signals cannot be
+  sent to a particular process."
   [g [pid io-id :as coord] msgs] (g/inject g coord msgs))
 
 (defn process
@@ -160,20 +179,25 @@
   datafy.
 
   arity 0 - 'describe', () -> description
-  where description is a map with keys :params :ins and :outs, each of which
-  in turn is a map of keyword to doc string, and :workload with
-  possible values of :mixed :io :compute. All entries in the describe
-  return map are optional.
+  where description is a map with possible keys:
+  :params :ins and :outs, each of which in turn is a map of keyword to doc string
+  :signal-select - a predicate of a signal-id. Messages on approved
+                   signals will appear in the transform arity (see below)
+                   For the simple case of enumerated signal-ids, use a set,
+                   e.g. #{:this/signal :that/signal}
+                   If no :signal-select is provided, no signals will be received
+  :workload with possible values of :mixed :io :compute.
+  All entries in the describe return map are optional.
   
   :params describes the initial arguments to setup the state for the function.
-  :ins enumerates the input[s], for which the flow will create channels
-  :outs enumerates the output[s], for which the flow may create channels.
+  :ins enumerates the process input[s], for which the flow will create channels
+  :outs enumerates the process output[s], for which the flow _may_ create channels.
   :workload - describes the nature of the workload, one of :mixed :io or :compute
           an :io workload should not do extended computation
           a :compute workload should never block
   
-  No key may be present in both :ins and :outs, allowing for a uniform
-  channel coordinate system of [:process-id :channel-id]. The
+  No io-id key may be present in both :ins and :outs, allowing for a
+  uniform channel coordinate system of [:process-id :channel-id]. The
   ins/outs/params returned will be the ins/outs/params of the
   process. describe may be called by users to understand how to use
   the proc. It will also be called by the impl in order to discover
@@ -213,16 +237,23 @@
   process will no longer be used following that. See the SPI for
   details. state' will be the state supplied to subsequent calls.
 
-  arity 3 - 'transform', (state in-name msg) -> [state' output]
+  arity 3 - 'transform', (state in-or-signal-id msg) -> [state' output]
   where output is a map of outid->[msgs*]
 
-  The transform arity will be called every time a message arrives at any
-  of the inputs. Output can be sent to none, any or all of the :outs
-  enumerated, and/or an input named by a [pid inid] tuple (e.g. for
-  reply-to), and/or to the ::flow/report output. A step need not
-  output at all (output or msgs can be empyt/nil), however an output _message_
-  may never be nil (per core.async channels). state' will be the state
-  supplied to subsequent calls.
+  The transform arity will be called every time a message arrives at
+  any of the inputs or signals (selected via :signal-select in
+  describe), identified by the id. Output can be sent to none, any or
+  all of the :outs enumerated, and/or an input named by a [pid in-id]
+  coord tuple (e.g. for reply-to), and/or to the ::flow/report
+  output.
+
+  You can broadcast output to all processes selecting a signal via
+  the special coord [::flow/cast a-signal-id] Note that signals cannot
+  be sent to a particular process.
+
+  A step need not output at all (output or msgs can be empty/nil),
+  however an output _message_ may never be nil (per core.async
+  channels). state' will be the state supplied to subsequent calls.
 
   process also accepts an option map with keys:
   :workload - one of :mixed, :io or :compute

src/main/clojure/clojure/core/async/flow/impl.clj

@@ -34,7 +34,7 @@
       (.submit e ^Callable #(apply f args)))))
 
 (defn prep-proc [ret pid {:keys [proc, args, chan-opts] :or {chan-opts {}}}]
-  (let [{:keys [ins outs]} (spi/describe proc)
+  (let [{:keys [ins outs signal-select]} (spi/describe proc)
         copts (fn [cs]
                 (zipmap (keys cs) (map #(chan-opts %) (keys cs))))
         inopts (copts ins)
@@ -43,7 +43,7 @@
               (some (partial contains? outopts) (keys inopts)))
       (throw (ex-info ":ins and :outs cannot share ids within a process"
                       {:pid pid :ins (keys inopts) :outs (keys outopts)})))
-    (assoc ret pid {:pid pid :proc proc :ins inopts :outs outopts :args args})))
+    (assoc ret pid {:pid pid :proc proc :ins inopts :outs outopts :args args :signal-select signal-select})))
 
 (defn create-flow
   "see lib ns for docs"
@@ -119,7 +119,18 @@
                                                (needs-mult? coord conns) (make-chan co)
                                                ;;direct connect 1:1
                                                :else (in-chans (first conns))))) 
-                                         outopts))
+                                         outopts))                 
+                  ;;pid->{:select :chan}
+                  castees (reduce (fn [ret {:keys [pid signal-select]}]
+                                    (assoc ret pid
+                                           {:select signal-select
+                                            :chan (async/chan (async/sliding-buffer 100))}))
+                                     {} (vals pdescs))
+                  cast (fn [sigid msgs]
+                         (doseq [{:keys [select chan]} (vals castees)]
+                           (when (and select (select sigid))
+                             (doseq [m msgs]
+                               (async/>!! chan [sigid m])))))
                   ;;mults
                   _  (doseq [[out ins] conn-map]
                        (when (needs-mult? out ins)
@@ -128,7 +139,7 @@
                              (async/tap m (in-chans in))))))
                   write-chan #(if-let [[_ c] (or (find in-chans %) (find out-chans %))]
                                 c
-                                (throw (ex-info "can't resolve channel with coord" {:coord %})))
+                                (throw (ex-info "can't resolve channel with io-id" {:io-id %})))
                   resolver (reify spi/Resolver
                              (get-write-chan [_ coord]
                                (write-chan coord))
@@ -139,9 +150,11 @@
                       (let [chan-map (fn [ks coll] (zipmap (keys ks) (map #(coll [pid %]) (keys ks))))
                             control-tap (async/chan 10)]
                         (async/tap control-mult control-tap)
-                        (spi/start proc {:pid pid :args (assoc args ::flow/pid pid) :resolver resolver
+                        (spi/start proc {:pid pid :args (assoc args ::flow/pid pid)
+                                         :resolver resolver :cast cast
                                          :ins (assoc (chan-map ins in-chans)
-                                                     ::flow/control control-tap)
+                                                     ::flow/control control-tap
+                                                     ::flow/casts (-> pid castees :chan))
                                          :outs (assoc (chan-map outs out-chans)
                                                       ::flow/error error-chan
                                                       ::flow/report report-chan)}))
@@ -151,7 +164,7 @@
               (doseq [p (vals pdescs)]
                 (start-proc p))
               ;;the only connection to a running flow is via channels
-              (reset! chans {:control control-chan :resolver resolver
+              (reset! chans {:control control-chan :resolver resolver :cast cast
                              :report report-chan, :error error-chan
                              :ins in-chans, :outs out-chans})
               {:report-chan report-chan :error-chan error-chan}))
@@ -172,12 +185,14 @@
       (pause-proc [_ pid] (send-command ::flow/pause pid))
       (resume-proc [_ pid] (send-command ::flow/resume pid))
       (ping-proc [_ pid timeout-ms] (handle-ping pid timeout-ms))         
-      (inject [_ coord msgs]
-        (let [{:keys [resolver]} (running-chans)
-              chan (spi/get-write-chan resolver coord)]
-          ((futurize #(doseq [m msgs]
-                        (async/>!! chan m))
-                     {:exec :io})))))))
+      (inject [_ [target id :as coord] msgs]
+        (let [{:keys [resolver cast]} (running-chans)
+              do-io (if (= target ::flow/cast)
+                      #(cast id msgs)
+                      (let [chan (spi/get-write-chan resolver coord)]
+                        #(doseq [m msgs]
+                           (async/>!! chan m))))]
+          ((futurize do-io {:exec :io})))))))
 
 (defn handle-command
   [pid pong status cmd]
@@ -199,26 +214,29 @@
                             :paused ::flow/pause))
     state))
 
-(defn send-outputs [status state outputs outs resolver control handle-command transition]
+(defn send-outputs [status state outputs outs resolver control handle-command transition cast]
   (loop [nstatus status, nstate state, outputs (seq outputs)]
     (if (or (nil? outputs) (= nstatus :exit))
       [nstatus nstate]
       (let [[output msgs] (first outputs)]
-        (if-let [outc (or (outs output) (spi/get-write-chan resolver output))]
-          (let [[nstatus nstate]
-                (loop [nstatus nstatus, nstate nstate, msgs (seq msgs)]
-                  (if (or (nil? msgs) (= nstatus :exit))
-                    [nstatus nstate]
-                    (let [[v c] (async/alts!!
-                                 [control [outc (first msgs)]]
-                                 :priority true)]
-                      (if (= c control)
-                        (let [nnstatus (handle-command nstatus v)
-                              nnstate (handle-transition transition nstatus nnstatus nstate)]
-                          (recur nnstatus nnstate msgs))
-                        (recur nstatus nstate (next msgs))))))]
-            (recur nstatus nstate (next outputs)))
-          (recur nstatus nstate  (next outputs)))))))
+        (if (and (vector? output) (= (first output) ::flow/cast))
+          (do (cast (second output) msgs)
+              (recur nstatus nstate (next outputs)))
+          (if-let [outc (or (outs output) (spi/get-write-chan resolver output))]
+            (let [[nstatus nstate]
+                  (loop [nstatus nstatus, nstate nstate, msgs (seq msgs)]
+                    (if (or (nil? msgs) (= nstatus :exit))
+                      [nstatus nstate]
+                      (let [[v c] (async/alts!!
+                                   [control [outc (first msgs)]]
+                                   :priority true)]
+                        (if (= c control)
+                          (let [nnstatus (handle-command nstatus v)
+                                nnstate (handle-transition transition nstatus nnstatus nstate)]
+                            (recur nnstatus nnstate msgs))
+                          (recur nstatus nstate (next msgs))))))]
+              (recur nstatus nstate (next outputs)))
+            (recur nstatus nstate  (next outputs))))))))
 
 (defn proc
   "see lib ns for docs"
@@ -233,7 +251,7 @@
           (walk/postwalk datafy {:step step :desc desc})))
       spi/ProcLauncher
       (describe [_] desc)
-      (start [_ {:keys [pid args ins outs resolver]}]
+      (start [_ {:keys [pid args ins outs resolver cast]}]
         (assert (or (not params) args) "must provide :args if :params")
         (let [transform (if (= workload :compute)
                           #(.get ^Future ((futurize step {:exec (spi/get-exec resolver :compute)}) %1 %2 %3)
@@ -245,11 +263,12 @@
               outs (into (or outs {}) (::flow/out-ports state))
               io-id (zipmap (concat (vals ins) (vals outs)) (concat (keys ins) (keys outs)))
               control (::flow/control ins)
-              read-ins (dissoc ins ::flow/control)
+              casts (::flow/casts ins)
+              read-ins (dissoc ins ::flow/control ::flow/casts)
               run
               #(loop [status :paused, state state, count 0, read-ins read-ins]
                  (let [pong (fn [c]
-                              (let [pins (dissoc ins ::flow/control)
+                              (let [pins (dissoc ins ::flow/control ::flow/casts)
                                     pouts (dissoc outs ::flow/error ::flow/report)]
                                 (async/>!! c (walk/postwalk datafy
                                                 #::flow{:pid pid, :status status
@@ -269,18 +288,21 @@
                                                            (if (ipred cid)
                                                              (conj ret chan)
                                                              ret))
-                                                         [control] read-ins))
+                                                         [control casts] read-ins))
                                  [msg c] (async/alts!! read-chans :priority true)
                                  cid (io-id c)]
                              (if (= c control)
                                (let [nstatus (handle-command status msg)
                                      nstate (handle-transition step status nstatus state)]
                                  [nstatus nstate count read-ins])
                                (try
-                                 (let [[nstate outputs] (transform state cid msg)
+                                 (let [[nstate outputs]
+                                       (if (= c casts) ;;[sigid msg]
+                                         (transform state (first msg) (second msg))
+                                         (transform state cid msg))
                                        [nstatus nstate]
                                        (send-outputs status nstate outputs outs
-                                                     resolver control handle-command step)]
+                                                     resolver control handle-command step cast)]
                                    [nstatus nstate (inc count) (if (some? msg)
                                                                  read-ins
                                                                  (dissoc read-ins cid))])