Skip to content


Subversion checkout URL

You can clone with HTTPS or Subversion.

Download ZIP
Fetching contributors…

Cannot retrieve contributors at this time

510 lines (408 sloc) 14.461 kb
;; Copyright (c) Jeffrey Straszheim. All rights reserved. The use and
;; distribution terms for this software are covered by the Eclipse Public
;; License 1.0 ( which can
;; be found in the file epl-v10.html at the root of this distribution. By
;; using this software in any fashion, you are agreeing to be bound by the
;; terms of this license. You must not remove this notice, or any other,
;; from this software.
;; dataflow.clj
;; A Library to Support a Dataflow Model of State
;; straszheimjeffrey (gmail)
;; Created 10 March 2009
^{:author "Jeffrey Straszheim",
:doc "A library to support a dataflow model of state"}
(:use [clojure.set :only (union intersection difference)])
(:use [clojure.contrib.graph :only (directed-graph
(:use [clojure.walk :only (postwalk)])
(:use [clojure.contrib.except :only (throwf)])
;;; Chief Data Structures
;; Source Cell
; The data of a source cell is directly set by a calling function. It
; never depends on other cells.
(defstruct source-cell
:name ; The name, a symbol
:value ; Its value, a Ref
:cell-type) ; Should be ::source-cell
;; Cell
; A standard cell that computes its value from other cells.
(defstruct standard-cell
:name ; The name, a symbol
:value ; Its value, a Ref
:dependents ; The names of cells on which this depends, a collection
:fun ; A closure that computes the value, given an environment
:display ; The original expression for display
:cell-type) ; Should be ::cell
(derive ::cell ::dependent-cell) ; A cell that has a dependents field
;; Validator
; A cell that has no value, but can throw an exception when run
(defstruct validator-cell
:name ; Always ::validator
:dependents ; The names of cells on which this depends, a collection
:fun ; A clojure that can throw an exception
:display ; The original exprssion for display
:cell-type) ; Should be ::validator-cell
(derive ::validator-cell ::dependent-cell)
;; A sentinal value
(def *empty-value* (java.lang.Object.))
;; Dataflow
; A collection of cells and dependency information
(defstruct dataflow
:cells ; A set of all cells
:cells-map ; A map of cell names (symbols) to collections of cells
:fore-graph ; The inverse of the dependency graph, nodes are cells
:topological) ; A vector of sets of independent nodes -- orders the computation
;;; Environment Access
(defn get-cells
"Get all the cells named by name"
[df name]
((:cells-map @df) name))
(defn get-cell
"Get the single cell named by name"
[df name]
(let [cells (get-cells df name)]
(= (count cells) 1) (first cells)
(> (count cells) 1) (throwf Exception "Cell %s has multiple instances" name)
:otherwise (throwf Exception "Cell %s is undefined" name))))
(defn source-cell?
"Is this cell a source cell?"
(isa? (:cell-type cell) ::source-cell))
(defn get-source-cells
"Returns a collection of source cells from the dataflow"
(for [cell (:cells @df)
:when (source-cell? cell)]
(defn get-value
"Gets a value from the df matching the passed symbol.
Signals an error if the name is not present, or if it not a single
[df name]
(let [cell (get-cell df name)
result @(:value cell)]
(do (when (= *empty-value* result)
(throwf Exception "Cell named %s empty" name))
(defn get-values
"Gets a collection of values from the df by name"
[df name]
(let [cells (get-cells df name)
results (map #(-> % :value deref) cells)]
(when (some #(= % *empty-value*) results)
(throwf Exception "At least one empty cell named %s found" name))
(defn get-old-value
"Looks up an old value"
[df env name]
(if (contains? env name)
(env name)
(get-value df name)))
(defn get-value-from-cell
"Given a cell, get its value"
(-> cell :value deref))
;;; Build Dataflow Structure
(defn- build-cells-map
"Given a collection of cells, build a name->cells-collection map
from it."
(let [step (fn [m c]
(let [n (:name c)
o (get m n #{})
s (conj o c)]
(assoc m n s)))]
(reduce step {} cs)))
(defn- build-back-graph
"Builds the backward dependency graph from the cells map. Each
node of the graph is a cell."
[cells cells-map]
(let [step (fn [n]
(apply union (for [dep-name (:dependents n)]
(cells-map dep-name))))
neighbors (zipmap cells (map step cells))]
(struct-map directed-graph
:nodes cells
:neighbors neighbors)))
(defn- build-dataflow*
"Builds the dataflow structure"
(let [cells (set cs)
cells-map (build-cells-map cs)
back-graph (build-back-graph cells cells-map)
fore-graph (reverse-graph back-graph)]
(struct-map dataflow
:cells cells
:cells-map cells-map
:fore-graph fore-graph
:topological (dependency-list back-graph))))
(def initialize)
(defn build-dataflow
"Given a collection of cells, build and return a dataflow object"
(let [df (ref (build-dataflow* cs))]
(initialize df)
;;; Displaying a dataflow
(defn print-dataflow
"Prints a dataflow, one cell per line"
(let [f (fn [cell] (-> cell :name str))]
(doseq [cell (sort-by f (:cells @df))]
(prn cell))))
;;; Modifying a Dataflow
(defn add-cells
"Given a collection of cells, add them to the dataflow."
[df cells]
(let [new-cells (union (set cells) (:cells @df))]
(ref-set df (build-dataflow* new-cells))
(initialize df))))
(defn remove-cells
"Given a collection of cells, remove them from the dataflow."
[df cells]
(let [new-cells (difference (:cells @df) (set cells))]
(ref-set df (build-dataflow* new-cells))
(initialize df))))
;;; Cell building
(def *meta* {:type ::dataflow-cell})
(defn build-source-cell
"Builds a source cell"
[name init]
(with-meta (struct source-cell name (ref init) ::source-cell)
(defn- is-col-var?
(let [name (name symb)]
(and (= \? (first name))
(= \* (second name)))))
(defn- is-old-var?
(let [name (name symb)]
(and (= \? (first name))
(= \- (second name)))))
(defn- is-var?
(let [name (name symb)]
(and (= \? (first name))
(-> symb is-col-var? not)
(-> symb is-old-var? not))))
(defn- cell-name
`(quote ~(cond (is-var? symb) (-> symb name (.substring 1) symbol)
(or (is-col-var? symb)
(is-old-var? symb)) (-> symb name (.substring 2) symbol))))
(defn- replace-symbol
"Walk the from replacing the ?X forms with the needed calls"
[dfs ov form]
(-> form symbol? not) form
(is-var? form) `(get-value ~dfs ~(cell-name form))
(is-col-var? form) `(get-values ~dfs ~(cell-name form))
(is-old-var? form) `(get-old-value ~dfs ~ov ~(cell-name form))
:otherwise form))
(defn- build-fun
"Build the closure needed to compute a cell"
(let [dfs (gensym "df_")
ov (gensym "old_")]
`(fn [~dfs ~ov] ~(postwalk (partial replace-symbol dfs ov) form))))
(defn- get-deps
"Get the names of the dependent cells"
(let [step (fn [f]
(coll? f) (apply union f)
(-> f symbol? not) nil
(is-var? f) #{(cell-name f)}
(is-col-var? f) #{(cell-name f)}
(is-old-var? f) #{(cell-name f)}
:otherwise nil))]
(postwalk step form)))
(defn build-standard-cell
"Builds a standard cell"
[name deps fun expr]
(with-meta (struct standard-cell name (ref *empty-value*) deps fun expr ::cell)
(defn build-validator-cell
"Builds a validator cell"
[deps fun expr]
(with-meta (struct validator-cell ::validator deps fun expr ::validator-cell)
(defmacro cell
"Build a standard cell, like this:
(cell fred
(* ?mary ?joe))
Which creates a cell named fred that is the product of a cell mary and cell joe
(cell joe
(apply * ?*sally))
Which creates a cell that applies * to the collection of all cells named sally
(cell :source fred 0)
Which builds a source cell fred with initial value 0
(cell :validator (when (< ?fred ?sally)
(throwf \"%s must be greater than %s\" ?fred ?sally))
Which will perform the validation"
[type & data]
(symbol? type) (let [name type ; No type for standard cell
expr (first data) ; we ignore extra data!
deps (get-deps expr)
fun (build-fun expr)]
`(build-standard-cell '~name ~deps ~fun '~expr))
(= type :source) (let [[name init] data]
`(build-source-cell '~name ~init))
(= type :validator) (let [[expr] data
deps (get-deps expr)
fun (build-fun expr)]
`(build-validator-cell ~deps ~fun '~expr))))
;;; Cell Display
(defmulti display-cell
"A 'readable' form of the cell"
(defmethod display-cell ::source-cell
(list 'cell :source (:name cell) (-> cell :value deref)))
(defmethod display-cell ::cell
(list 'cell (:name cell) (:display cell) (-> cell :value deref)))
(defmethod display-cell ::validator-cell
(list 'cell :validator (:display cell)))
(defmethod print-method ::dataflow-cell
[f ^Writer w]
(binding [*out* w]
(pr (display-cell f))))
;;; Evaluation
(defmulti eval-cell
"Evaluate a dataflow cell. Return [changed, old val]"
(fn [df data old cell] (:cell-type cell)))
(defmethod eval-cell ::source-cell
[df data old cell]
(let [name (:name cell)
val (:value cell)
ov @val]
(if (contains? data name)
(let [new-val (data name)]
(if (not= ov new-val)
(do (ref-set val new-val)
[true ov])
[false ov]))
[false ov])))
(defmethod eval-cell ::cell
[df data old cell]
(let [val (:value cell)
old-val @val
new-val ((:fun cell) df old)]
(if (not= old-val new-val)
(do (ref-set val new-val)
[true old-val])
[false old-val])))
(defmethod eval-cell ::validator-cell
[df data old cell]
(do ((:fun cell) df old)
[false nil]))
(defn- perform-flow
"Evaluate the needed cells (a set) from the given dataflow. Data is
a name-value mapping of new values for the source cells"
[df data needed]
(loop [needed needed
tops (:topological @df)
old {}]
(let [now (first tops) ; Now is a set of nodes
new-tops (next tops)]
(when (and (-> needed empty? not)
(-> now empty? not))
(let [step (fn [[needed old] cell]
(let [[changed ov] (try
(eval-cell df data old cell)
(catch Exception e
(throw (Exception. (str cell) e))))
nn (disj needed cell)]
(if changed
[(union nn (get-neighbors (:fore-graph @df) cell))
(assoc old (:name cell) ov)]
[nn old])))
[new-needed new-old] (reduce step
[needed old]
(intersection now needed))]
(recur new-needed new-tops new-old))))))
(defn- validate-update
"Ensure that all the updated cells are source cells"
[df names]
(let [scns (set (map :name (get-source-cells df)))]
(doseq [name names]
(when (-> name scns not)
(throwf Exception "Cell %n is not a source cell" name)))))
(defn update-values
"Given a dataflow, and a map of name-value pairs, update the
dataflow by binding the new values. Each name must be of a source
[df data]
(validate-update df (keys data))
(let [needed (apply union (for [name (keys data)]
(set ((:cells-map @df) name))))]
(perform-flow df data needed))))
(defn- initialize
"Apply all the current source cell values. Useful for a new
dataflow, or one that has been updated with new cells"
(let [needed (:cells @df)
fg (:fore-graph @df)]
(perform-flow df {} needed)))
;;; Watchers
(defn add-cell-watcher
"Adds a watcher to a cell to respond to changes of value. The is a
function of 4 values: a key, the cell, its old value, its new
value. This is implemented using Clojure's add-watch to the
underlying ref, and shared its sematics"
[cell key fun]
(let [val (:value cell)]
(add-watch val key (fn [key _ old-v new-v]
(fun key cell old-v new-v)))))
(def df
[(cell :source fred 1)
(cell :source mary 0)
(cell greg (+ ?fred ?mary))
(cell joan (+ ?fred ?mary))
(cell joan (* ?fred ?mary))
(cell sally (apply + ?*joan))
(cell :validator (when (number? ?-greg)
(when (<= ?greg ?-greg)
(throwf Exception "Non monotonic"))))]))
(do (println)
(print-dataflow df))
(add-cell-watcher (get-cell df 'sally)
(fn [key cell o n]
(printf "sally changed from %s to %s\n" o n)))
(update-values df {'fred 1 'mary 1})
(update-values df {'fred 5 'mary 1})
(update-values df {'fred 0 'mary 0})
(get-value df 'fred)
(get-values df 'joan)
(get-value df 'sally)
(get-value df 'greg)
(use :reload 'clojure.contrib.dataflow)
(use 'clojure.stacktrace) (e)
(use 'clojure.contrib.trace)
;; End of file
Jump to Line
Something went wrong with that request. Please try again.