Implement pruning

project.clj

@@ -1,6 +1,7 @@
 (defproject clojurewerkz/vclock "1.0.0-SNAPSHOT"
   :description "A vector clock implementation roughly ported from Riak Core"
-  :dependencies [[org.clojure/clojure "1.3.0"]]
+  :dependencies [[org.clojure/clojure  "1.3.0"]
+                 [clojurewerkz/support "0.7.0-beta1"]]
   :profiles {:1.4 {:dependencies [[org.clojure/clojure "1.4.0"]]}
              :1.5 {:dependencies [[org.clojure/clojure "1.5.0-master-SNAPSHOT"]]}
              :dev {:resource-paths ["test/resources"]

src/clojure/clojurewerkz/vclock/core.clj

@@ -1,13 +1,14 @@
 (ns ^{:doc "An implementation of vector clocks, heavily inspired by Basho's Riak Core."}
   clojurewerkz.vclock.core
-  (:refer-clojure :exclude [find merge]))
+  (:refer-clojure :exclude [find merge])
+  (:use [clojurewerkz.support.core :only [map->pairs pairs->map]]))
 
 
 ;;
 ;; Implementation
 ;;
 
-(defn- timestamp-in-seconds
+(defn timestamp-in-seconds
   []
   (Math/round (double (/ (.getTime (java.util.Date.)) 1000))))
 
@@ -23,6 +24,11 @@
   [& kv]
   (apply sorted-map kv))
 
+(defn entry
+  "Instantiates a new vclock entry"
+  [^long counter ^long timestamp]
+  (VClockEntry. counter timestamp))
+
 (defn increment
   "Increments vlock by adding a new entry"
   ([m node]
@@ -54,3 +60,33 @@
   "Merge multiple VClocks into one"
   [& vclocks]
   (apply merge-with (cons merge-helper vclocks)))
+
+
+(defn prune
+  "Prunes (possibly minimizes) vclocks and returns a new version. Used to limit vector clock growth.
+
+   See http://wiki.basho.com/Vector-Clocks.html for more information about vector clocks.
+
+   Accepted options:
+
+   :small-vclock (default: 10): vclock with the number of entries less than or equal to this value are not pruned
+   :big-vclock (default: 20): vclocks with more entries than this value will be pruned down to this many latest entries
+   :young-vclock (default: 20 seconds)
+   :old-vclock (default: 24 hours): vlock entries older than this value (in seconds) will be pruned"
+  [vclock & {:keys [small-vclock big-vclock young-vclock old-vclock] :or {small-vclock 10
+                                                                          big-vclock 20
+                                                                          ;; 20 seconds
+                                                                          young-vclock 20
+                                                                          ;; 24 hours
+                                                                          old_vclock (* 24 60 60)
+                                                                          } :as options}]
+  (let [ps (map->pairs vclock)
+        xs (sort-by (fn [[k ^VClockEntry v]] (.timestamp v)) ps)
+        ys (cond
+             (> (count xs) big-vclock)    (take-last big-vclock xs)
+             (<= (count xs) small-vclock) xs             
+             :else                        (let [now (timestamp-in-seconds)]
+                                            (filter (fn [[k ^VClockEntry v]]
+                                                      (< old-vclock (- now (.timestamp v))))
+                                                    xs)))]
+    (pairs->map ys)))

test/clojurewerkz/vclock/core_test.clj

@@ -56,3 +56,31 @@
           c (vclock/merge a b)]
       (is (= c (vclock/fresh :a (VClockEntry. 6 6)
                              :b (VClockEntry. 7 7)))))))
+
+
+(deftest test-pruning
+  (testing "vclocks greater than big-vclock are pruned"
+    (let [ts       (- (vclock/timestamp-in-seconds) 500)
+          a        (VClockEntry. 8 (+ 5 ts))
+          b        (VClockEntry. 6 (+ 10 ts))
+          c        (VClockEntry. 2 ts)
+          d        (VClockEntry. 3 (+ ts 100))
+          vc       (sorted-map :a a :b b :c c :d d)
+          expected (sorted-map :d d :b b)]
+      (is (= expected (vclock/prune vc :small-vclock 1 :big-vclock 2)))))
+  (testing "vclocks lesser than small-vclock are not pruned"
+    (let [ts       (- (vclock/timestamp-in-seconds) 500)
+          a        (VClockEntry. 8 (+ 5 ts))
+          b        (VClockEntry. 6 (+ 10 ts))
+          c        (VClockEntry. 2 ts)
+          d        (VClockEntry. 3 (+ ts 100))
+          vc       (sorted-map :a a :b b :c c :d d)]
+      (is (= vc (vclock/prune vc :small-vclock 10 :big-vclock 20)))))
+  (testing "vclocks of size between small-vclock and big-vclock are pruned based on their age"
+    (let [ts       (- (vclock/timestamp-in-seconds) 200)
+          a        (vclock/entry 8 (+ 5 ts))
+          b        (vclock/entry 6 (+ 10 ts))
+          c        (vclock/entry 2 ts)
+          d        (vclock/entry 3 (+ ts 100))
+          vc       (sorted-map :a a :b b :c c :d d)]
+      (is (= vc (vclock/prune vc :small-vclock 1 :big-vclock 10 :old-vclock 99))))))