wip

.gitignore

@@ -0,0 +1,8 @@
+/pom.xml
+*jar
+/lib
+/classes
+/native
+/.lein-failures
+/checkouts
+/.lein-deps-sum

README

@@ -0,0 +1,13 @@
+# datomic-sim
+
+FIXME: write description
+
+## Usage
+
+FIXME: write
+
+## License
+
+Copyright (C) 2012 FIXME
+
+Distributed under the Eclipse Public License, the same as Clojure.

project.clj

@@ -0,0 +1,5 @@
+(defproject datomic-sim "1.0.0-SNAPSHOT"
+ :description "Simulation testing with Datomic"
+ :dependencies [[org.clojure/clojure "1.5.0-beta1"]
+ [com.datomic/datomic-free "0.8.3561"
+ :exclusions [org.clojure/clojure]]])

resources/datomic-sim/schema.dtm

@@ -0,0 +1,68 @@
+{:core
+ [[{:db.install/_partition :db.part/db,
+ :db/id #db/id[:db.part/db],
+ :db/ident :test}
+ {:db.install/_partition :db.part/db,
+ :db/id #db/id[:db.part/db],
+ :db/ident :sim}
+ {:db.install/_partition :db.part/db,
+ :db/id #db/id[:db.part/db],
+ :db/ident :process}]]
+
+ :action
+ [[{:db/id #db/id[:db.part/db]
+ :db/ident :action/atTime
+ :db/valueType :db.type/long
+ :db/doc "Time this action should happen, in msec relative to start of the sim."
+ :db/cardinality :db.cardinality/one
+ :db.install/_attribute :db.part/db}
+ {:db/id #db/id[:db.part/db]
+ :db/ident :action/type
+ :db/valueType :db.type/ref
+ :db/doc "Type of the action."
+ :db/index true
+ :db/cardinality :db.cardinality/one
+ :db.install/_attribute :db.part/db}]]
+
+ :process
+ [[{:db/id #db/id[:db.part/db]
+ :db/ident :process/ordinal
+ :db/valueType :db.type/long
+ :db/doc "Ordinal number of process within sim, determines which actions this process is responsible for."
+ :db/cardinality :db.cardinality/one
+ :db.install/_attribute :db.part/db}
+ {:db/id #db/id[:db.part/db]
+ :db/ident :process/uuid
+ :db/valueType :db.type/long
+ :db/doc "Unique id for process"
+ :db/unique :db.unique/value
+ :db/cardinality :db.cardinality/one
+ :db.install/_attribute :db.part/db}]]
+
+ :sim
+ [[{:db/id #db/id[:db.part/db]
+ :db/ident :sim/processCount
+ :db/valueType :db.type/long
+ :db/doc "Total number of processes desired for this sim."
+ :db/cardinality :db.cardinality/one
+ :db.install/_attribute :db.part/db}
+ {:db/id #db/id[:db.part/db]
+ :db/ident :sim/processes
+ :db/valueType :db.type/ref
+ :db/doc "Processes that have joined this sim."
+ :db/cardinality :db.cardinality/many
+ :db.install/_attribute :db.part/db}
+ {:db/id #db/id [:db.part/db]
+ :db/ident :sim/join
+ :db/doc "Add procid to the sim, if any slots are still available"
+ :db/fn #db/fn
+ {:lang "clojure"
+ :params [db simid procid]
+ :code (let [procs (q '[:find ?procid
+ :in $ ?simid
+ :where [?simid :sim/processes ?procid]]
+ db simid)]
+ (when (and (< (count procs) (:sim/processCount (d/entity db simid)))
+ (not (some (fn [[e]] (= e procid)) procs)))
+ [[:db/add simid :sim/processes procid]
+ [:db/add procid :process/ordinal (count procs)]]))}}]]}

src/datomic/sim.clj

@@ -0,0 +1,146 @@
+(ns datomic.sim
+ (:require [datomic.api :refer (q db transact entity)]))
+
+(set! *warn-on-reflection* true)
+
+
+(defmulti perform-action
+ "Perform the action"
+ (fn [action] (:action/type action)))
+
+(defmulti action-seq
+ "Given a test and a process participating in the test, return a time-ordered
+ sequence of actions that process should perform. Default is to round robin
+ agents across all processes, so that each agent's actions are localized
+ to a process."
+ (fn [test process] (:test/type test)))
+
+;; general utilities ;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;
+(defn keep-partition
+ "Keep 1/group-size items from coll, round robin,
+ offset from zero by ordinal."
+ [ordinal group-size coll]
+ (assert (< -1 ordinal group-size))
+ (keep-indexed
+ (fn [idx item]
+ (when (zero? (mod (- idx ordinal) group-size))
+ item))
+ coll))
+
+(defn solo
+ "Like first, but throws if more than one item"
+ [coll]
+ (assert (not (next coll)))
+ (first coll))
+
+(def ssolo (comp solo solo))
+
+(defprotocol Eid
+ (e [_]))
+
+(extend-protocol Eid
+ java.lang.Long
+ (e [n] n)
+
+ clojure.lang.Keyword
+ (e [k] k)
+
+ datomic.Entity
+ (e [ent] (:db/id ent)))
+
+;; sim time ;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;
+(def sim-start (atom nil))
+
+(defn reset
+ "Zero the sim clock. You must call this before running
+ a simulation!"
+ []
+ (reset! sim-start (System/currentTimeMillis)))
+
+(defn now
+ "Returns the current simulation time. This is the naive version using the
+ system time in ms."
+ []
+ (- (System/currentTimeMillis) @sim-start))
+
+(defn sleep-until
+ "Checks if the target time is less-than the actual time and sleeps the remaining ms
+ if it is."
+ [twhen]
+ (let [tnow (now)]
+ (when (< tnow twhen)
+ (Thread/sleep (- twhen tnow)))))
+
+
+;; sim runner ;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;
+(defn handle-action-error
+ [actor agent ^Throwable error]
+ (.printStackTrace error))
+
+(def ^:private via-agent-for
+ (memoize
+ (fn [x]
+ (assert x)
+ (let [a (agent nil)]
+ (set-error-handler! a (partial handle-action-error x))
+ (set-error-mode! a :continue)
+ a))))
+
+
+(defn feed-action
+ "Feed a single action to the actor's agent."
+ [{actor :sim/_actor :as action}]
+ (send-off
+ (via-agent-for (:db/id actor))
+ (fn [agent-state]
+ (perform-action action)
+ agent-state)))
+
+(defn feed-all
+ "Feed all actions, which should be sorted by ascending
+ :action/atTime"
+ [actions]
+ (doseq [{t :action/atTime :as action} actions]
+ (sleep-until t)
+ (feed-action action)))
+
+(defn await-all
+ "Given a collection of objects, calls await on the agent for each one"
+ [coll]
+ (apply await (map via-agent-for coll)))
+
+(defn join-sim
+ "Returns process or nil."
+ [conn run process]
+ (let [{:keys [db-after]} @(transact conn [[:run/join (e run) (e process)]])]
+ (-> (q '[:find ?process
+ :in $ ?run ?process
+ :where [?run :run/processes ?process]]
+ db-after (e run) (e process))
+ ssolo boolean)))
+
+(defmethod action-seq
+ [test process]
+ (let [nprocs (:sim/processCount test)
+ ord (:process/ordinal process)
+ agentids (->> (:test/agents test) (sort-by :db/id) (keep-partition ord nprocs) (map :db/id) (into #{}))]
+ (->> (datoms db :avet :action/atTime)
+ (map (fn [datom] (entity db (:e datom))))
+ (filter (fn [datom] (contains? agentids (-> action :agent/_actions first)))))))
+
+(def puuid (squuid))
+
+(defn process-uuid
+ []
+ (return puuid))
+
+(defn create-process
+ [conn]
+ (let []))
+
+(defn run-sim-process
+ [uri simuuid]
+ (let [procid (squuid)
+ conn (connect uri)
+ id (tempid :process)]
+ (transact conn )))

test/datomic/sim/repl.clj

@@ -0,0 +1,21 @@
+(ns datomic.sim.repl
+ (:require [datomic.api :as d]))
+
+(defn scratch-conn
+ "Create a connection to an anonymous, in-memory database."
+ []
+ (let [uri (str "datomic:mem:https://" (d/squuid))]
+ (d/delete-database uri)
+ (d/create-database uri)
+ (d/connect uri)))
+
+(defn convenient
+ []
+ (in-ns 'user)
+ (set! *warn-on-reflection* true)
+ (set! *print-length* 20)
+ (use '[datomic.api :as d])
+ (require
+ '[clojure.string :as str]
+ '[clojure.java.io :as io]
+ '[clojure.pprint :as pprint]))