:exec option -> :workload, proc-fn :describe can return :workflow, process opt overridesflow definition -> config, no use of Clojure 1.12 features so 1.11 compatible

Author: richhickey

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

@@ -8,7 +8,10 @@
 
 (ns ^{:author "Rich Hickey"}
   clojure.core.async.flow
-  "A library for building concurrent, event driven data processing
+  "
+  Note - Alpha, work-in-progress, names and other details are in flux 
+
+  A library for building concurrent, event driven data processing
   flows out of communication-free functions, while centralizing
   control, reporting, execution and error handling. Built on core.async.
 
@@ -18,21 +21,21 @@
   a set of channels for centralized control, reporting, error-handling,
     and execution of the processes
 
-  A flow is constructed from flow definition data which defines a
+  A flow is constructed from flow configuration data which defines a
   directed graph of processes and the connections between
   them. Processes describe their I/O requirements and the
   flow (library) itself creates channels and passes them to the
   processes that requested them. See 'create-flow' for the
-  details. The flow definition provides a centralized place for policy
-  decisions regarding configuration, threading, buffering etc. 
+  details. The flow configuration provides a centralized place for
+  policy decisions regarding process settings, threading, buffering etc.
 
-  It is expected that applications will rarely define processes but
-  instead use the API functions here, 'process' and 'step-process',
-  that implement the process protocol in terms of calls to ordinary
-  functions that include no communication or core.async code. In this
-  way the library helps you achieve a strict separation of your
-  application logic from its execution, communication, lifecycle and
-  monitoring.
+  It is expected that applications will rarely define instances of the
+  process protocol but instead use the API functions here, 'process'
+  and 'step-process', that implement the process protocol in terms of
+  calls to ordinary functions that might include no communication or
+  core.async code. In this way the library helps you achieve a strict
+  separation of your application logic from its execution,
+  communication, lifecycle, error handling and monitoring.
 
   Note that at several points the library calls upon the user to
   supply ids for processes, inputs, outputs etc. These should be
@@ -52,7 +55,7 @@
 (set! *warn-on-reflection* true)
 
 (defn create-flow
-  "Creates a flow from the supplied definition: a map containing the
+  "Creates a flow from the supplied configuration: a map containing the
   keys :procs and :conns, and optionally :mixed-exec/:io-exec/:compute-exec
 
   :procs - a map of pid->proc-def
@@ -66,26 +69,26 @@
   
   :conns - a collection of [[from-pid outid] [to-pid inid]] tuples.
 
-  Inputs and outputs support mutliple connections. When an output is
+  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.
 
   :mixed-exec/:io-exec/:compute-exec -> ExecutorService
   These can be used to specify the ExecutorService to use for the
-  corresonding context, in lieu of the lib defaults
+  corresonding workflow, in lieu of the lib defaults.
 
   N.B. The flow is not started. See 'start'"
-  [def] (impl/create-flow def))
+  [config] (impl/create-flow config))
 
 (defn start
   "starts the entire flow from init values. The processes start paused.
   Call 'resume' or 'resume-proc' to start flow.  returns a map with keys:
   
-  ::flow/report-chan - a core.async chan for reading.'ping' reponses
+  :report-chan - a core.async chan for reading.'ping' reponses
   will show up here, as will any explicit ::flow/report outputs
   from :transform/:introduce
   
-  ::flow/error-chan - a core.async chan for reading. Any (and only)
+  :error-chan - a core.async chan for reading. Any (and only)
   exceptions thrown anywhere on any thread inside a flow will appear
   in maps sent here. There will at least be a ::flow/ex entry with the
   exception, and may be additional keys for pid, state, status etc
@@ -144,16 +147,23 @@
 
   :describe - required, () -> desc
   where desc is a map with keys :params :ins and :outs, each of which
-  in turn is a map of keyword to doc string
+  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.
   
   :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.
+  :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 The ins/outs/params of f
-  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 what channels are needed.
+  No 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
+  what channels are needed.
 
   :init - optional, (arg-map) -> initial-state
   
@@ -199,15 +209,19 @@
   for introduce to return with no output. Do not spin poll in the introduce
   fn.
 
-  proc accepts an option map with keys:
-  :exec - one of :mixed, :io or :compute, default :mixed
-  :compute-timeout-ms - if :exec is :compute, this timeout (default 5000 msec)
+  process accepts an option map with keys:
+  :workflow - one of :mixed, :io or :compute
+  :compute-timeout-ms - if :workflow is :compute, this timeout (default 5000 msec)
                 will be used when getting the return from the future - see below
 
-  The :compute context is not allowed for proc impls that
+  A :workflow supplied as an option to process will override
+  any :workflow returned by the :describe fn of the process. If neither
+  are provded the default is :mixed.
+
+  The :compute workload is not allowed for proc impls that
   provide :introduce (as I/O is presumed).
 
-  In the :exec context of :mixed or :io, this dictates the type of
+  In the :workflow context of :mixed or :io, this dictates the type of
   thread in which the process loop will run, _including its calls to
   transform/introduce_. 
 
@@ -223,8 +237,8 @@
 
   When :compute is specified transform must not block!"
   ([process-impl-map] (process process-impl-map nil))
-  ([process-impl-map {:keys [exec timeout-ms]
-                      :or {exec :mixed, timeout-ms 5000} :as opts}]
+  ([process-impl-map {:keys [workflow timeout-ms]
+                      :or {timeout-ms 5000} :as opts}]
    (impl/proc process-impl-map opts)))
 
 (defn step-process
@@ -248,13 +262,13 @@
    (process {:describe f, :init f, :transform f} opts)))
 
 (defn futurize
-  "Takes a fn f and returns a fn that takes the same arguments as f and
-  immediately returns a future, having starting a thread of the
-  indicated type, or via the supplied executor, that invokes f with
-  those args and completes that future with its return.
+  "Takes a fn f and returns a fn that takes the same arguments as f
+  and immediately returns a future, having started a thread for the
+  indicated workload, or via the supplied executor, that invokes f
+  with those args and completes that future with its return.
 
   futurize accepts kwarg options:
-  :exec - one of :mixed, :io, :compute
+  :exec - one of the workloads :mixed, :io, :compute
           or a j.u.c.ExecutorService object,
           default :mixed"
   [f & {:keys [exec]

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

@@ -34,13 +34,12 @@
 
 (defn futurize ^Future [f {:keys [exec]}]
   (fn [& args]
-    (^[Callable] ExecutorService/.submit
-     (case exec
-       :compute compute-exec
-       :io io-exec
-       :mixed mixed-exec
-       exec)
-     #(apply f args))))
+    (let [^ExecutorService e (case exec
+                                   :compute compute-exec
+                                   :io io-exec
+                                   :mixed mixed-exec
+                                   exec)]
+      (.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)
@@ -57,8 +56,7 @@
 (defn create-flow
   "see lib ns for docs"
   [{:keys [procs conns mixed-exec io-exec compute-exec]
-    :or {mixed-exec mixed-exec, io-exec io-exec, compute-exec compute-exec}
-    :as desc}]
+    :or {mixed-exec mixed-exec, io-exec io-exec, compute-exec compute-exec}}]
   (let [lock (ReentrantLock.)
         chans (atom nil)
         execs {:mixed mixed-exec :io io-exec :compute compute-exec}
@@ -216,75 +214,75 @@
 
 (defn proc
   "see lib ns for docs"
-  [{:keys [describe init transition transform introduce] :as impl} {:keys [exec compute-timeout-ms]}]
-  ;;validate the preconditions
-  (assert (= 1 (count (keep identity [transform introduce]))) "must provide exactly one of :transform or :introduce")
-  (assert (not (and introduce (= exec :compute))) "can't specify :introduce and :compute")
-  (reify
-    clojure.core.protocols/Datafiable
-    (datafy [_]
-      (let [{:keys [params ins outs]} (describe)]
-        {:impl impl :params (-> params keys vec) :ins (-> ins keys vec) :outs (-> outs keys vec)}))
-    spi/ProcLauncher
-    (describe [_]
-      (let [{:keys [params ins] :as desc} (describe)]
-        (assert (not (and ins introduce)) "can't specify :ins when :introduce")
-        (assert (or (not params) init) "must have :init if :params")
-        desc))
-    (start [_ {:keys [pid args ins outs resolver]}]
-      (let [comp? (= exec :compute)
-            transform (cond-> transform (= exec :compute)
-                              #(.get (futurize transform {:exec (spi/get-exec resolver :compute)})
-                                     compute-timeout-ms TimeUnit/MILLISECONDS))
-            exs (spi/get-exec resolver (if (= exec :mixed) :mixed :io))
-            io-id (zipmap (concat (vals ins) (vals outs)) (concat (keys ins) (keys outs)))
-            control (::flow/control ins)
-            ;;TODO rotate/randomize after control per normal alts?
-            read-chans (into [control] (-> ins (dissoc ::flow/control) vals))
-            run
-            #(loop [status :paused, state (when init (init args)), count 0]
-               (let [pong (fn []
-                            (let [pins (dissoc ins ::flow/control)
-                                  pouts (dissoc outs ::flow/error ::flow/report)]
-                              (async/>!! (outs ::flow/report)
-                                         #::flow{:report :ping, :pid pid, :status status
-                                                 :state state, :count count
-                                                 :ins (zipmap (keys pins) (map chan->data (vals pins)))
-                                                 :outs (zipmap (keys pouts) (map chan->data (vals pouts)))})))
-                     handle-command (partial handle-command pid pong)
-                     [nstatus nstate count]
-                     (try
-                       (if (= status :paused)
-                         (let [nstatus (handle-command status (async/<!! control))
-                               nstate (handle-transition transition status nstatus state)]
-                           [nstatus nstate count])
-                         ;;:running
-                         (let [[msg c] (if transform
-                                         (async/alts!! read-chans :priority true)
-                                         ;;introduce
-                                         (when-let [msg (async/poll! control)]
-                                           [msg control]))
-                               cid (io-id c)]
-                           (if (= c control)
-                             (let [nstatus (handle-command status msg)
-                                   nstate (handle-transition transition status nstatus state)]
-                               [nstatus nstate count])
-                             (try
-                               (let [[nstate outputs] (if transform
-                                                        (transform state cid msg)
-                                                        (introduce state))
-                                     [nstatus nstate]
-                                     (send-outputs status nstate outputs outs resolver control handle-command transition)]
-                                 [nstatus nstate (inc count)])
-                               (catch Throwable ex
-                                 (async/>!! (outs ::flow/error)
-                                            #::flow{:pid pid, :status status, :state state,
-                                                    :count (inc count), :cid cid, :msg msg :op :step, :ex ex})
-                                 [status state count])))))
-                       (catch Throwable ex
-                         (async/>!! (outs ::flow/error)
-                                    #::flow{:pid pid, :status status, :state state, :count (inc count), :ex ex})
-                         [status state count]))]
-                 (when-not (= nstatus :exit) ;;fall out
-                   (recur nstatus nstate (long count)))))]
-        ((futurize run {:exec exs}))))))
+  [{:keys [describe init transition transform introduce] :as impl} {:keys [workflow compute-timeout-ms]}]
+  (assert (= 1 (count (keep identity [transform introduce]))) "must provide exactly one of :transform or :introduce") 
+  (let [{:keys [params ins] :as desc} (describe)
+        workflow (or workflow (:workflow desc) :mixed)]
+    (assert (not (and ins introduce)) "can't specify :ins when :introduce")
+    (assert (or (not params) init) "must have :init if :params")
+    (assert (not (and introduce (= workflow :compute))) "can't specify :introduce and :compute")
+    (reify
+     clojure.core.protocols/Datafiable
+     (datafy [_]
+             (let [{:keys [params ins outs]} desc]
+               {:impl impl :params (-> params keys vec) :ins (-> ins keys vec) :outs (-> outs keys vec)}))
+     spi/ProcLauncher
+     (describe [_] desc)
+     (start [_ {:keys [pid args ins outs resolver]}]
+            (let [comp? (= workflow :compute)
+                  transform (cond-> transform (= workflow :compute)
+                                    #(.get (futurize transform {:exec (spi/get-exec resolver :compute)})
+                                           compute-timeout-ms TimeUnit/MILLISECONDS))
+                  exs (spi/get-exec resolver (if (= workflow :mixed) :mixed :io))
+                  io-id (zipmap (concat (vals ins) (vals outs)) (concat (keys ins) (keys outs)))
+                  control (::flow/control ins)
+                  ;;TODO rotate/randomize after control per normal alts?
+                  read-chans (into [control] (-> ins (dissoc ::flow/control) vals))
+                  run
+                  #(loop [status :paused, state (when init (init args)), count 0]
+                     (let [pong (fn []
+                                  (let [pins (dissoc ins ::flow/control)
+                                        pouts (dissoc outs ::flow/error ::flow/report)]
+                                    (async/>!! (outs ::flow/report)
+                                               #::flow{:report :ping, :pid pid, :status status
+                                                       :state state, :count count
+                                                       :ins (zipmap (keys pins) (map chan->data (vals pins)))
+                                                       :outs (zipmap (keys pouts) (map chan->data (vals pouts)))})))
+                           handle-command (partial handle-command pid pong)
+                           [nstatus nstate count]
+                           (try
+                             (if (= status :paused)
+                               (let [nstatus (handle-command status (async/<!! control))
+                                     nstate (handle-transition transition status nstatus state)]
+                                 [nstatus nstate count])
+                               ;;:running
+                               (let [[msg c] (if transform
+                                               (async/alts!! read-chans :priority true)
+                                               ;;introduce
+                                               (when-let [msg (async/poll! control)]
+                                                 [msg control]))
+                                     cid (io-id c)]
+                                 (if (= c control)
+                                   (let [nstatus (handle-command status msg)
+                                         nstate (handle-transition transition status nstatus state)]
+                                     [nstatus nstate count])
+                                   (try
+                                     (let [[nstate outputs] (if transform
+                                                              (transform state cid msg)
+                                                              (introduce state))
+                                           [nstatus nstate]
+                                           (send-outputs status nstate outputs outs
+                                                         resolver control handle-command transition)]
+                                       [nstatus nstate (inc count)])
+                                     (catch Throwable ex
+                                       (async/>!! (outs ::flow/error)
+                                                  #::flow{:pid pid, :status status, :state state,
+                                                          :count (inc count), :cid cid, :msg msg :op :step, :ex ex})
+                                       [status state count])))))
+                             (catch Throwable ex
+                               (async/>!! (outs ::flow/error)
+                                          #::flow{:pid pid, :status status, :state state, :count (inc count), :ex ex})
+                               [status state count]))]
+                       (when-not (= nstatus :exit) ;;fall out
+                         (recur nstatus nstate (long count)))))]
+              ((futurize run {:exec exs})))))))

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

@@ -9,7 +9,7 @@
 (ns clojure.core.async.flow.spi)
 
 (defprotocol ProcLauncher
-  "Note - definine a ProcLauncher is an advanced feature and should not
+  "Note - defining a ProcLauncher is an advanced feature and should not
   be needed for ordinary use of the library. This protocol is for
   creating new types of Processes that are not possible to create
   with ::flow/process.