A Jepsen client takes invocation operations and applies them to the system
being tested, returning corresponding completion operations. For our
Zookeeper test, we might model the system as a single register, stored in a
znode, storing an integer. The operations against that register might be
read, write, and compare-and-set, which we could model like so:
(defn r [_ _] {:type :invoke, :f :read, :value nil})
(defn w [_ _] {:type :invoke, :f :write, :value (rand-int 5)})
(defn cas [_ _] {:type :invoke, :f :cas, :value [(rand-int 5) (rand-int 5)]})These functions can be used by jepsen.generator to construct new invocations
for reads, writes, and compare-and-set ops, respectively. Note that the read
value is nil--we don't know what value is being read until we actually
perform the read. When the client reads a particular value it'll fill it in for
the completion op.
Now we need to take these operations and apply them to zookeeper. We'll use
the avout library, which provides a zk-atom backed by
zookeeper. We'll require avout.core and jepsen.client, and write a trivial
implementation of Jepsen's Client protocol...
(ns jepsen.zookeeper
(:require [avout.core :as avout]
[clojure.tools.logging :refer :all]
[clojure.java.io :as io]
[clojure.string :as str]
[jepsen [db :as db]
[client :as client]
[control :as c]
[tests :as tests]]
[jepsen.os.debian :as debian]))
(defn client
"A client for a single compare-and-set register"
[]
(reify client/Client
(setup! [_ test node]
(client))
(invoke! [this test op])
(teardown! [_ test])))
(defn zk-test
[version]
(assoc tests/noop-test
:name "zookeeper"
:os debian/os
:db (db version)
:client (client)))Clients have a three-part lifecycle. We begin with a single dormant client
(client), whose setup! function initializes a fresh client, and performs
any necessary setup work. We create fresh clients because each process needs an
independent client, talking to independent nodes. Once initialized, invoke!
lets our client handle operations. Finally, teardown! releases any resources
the client may be holding on to.
Our client doesn't hold any state yet, so we simply call (client) in setup!
to construct a fresh client for each process.
What state do we need for each client? In Avout, one makes a connection like so:
(def conn (avout/connect "some-host"))And using that connetion, Avout models a linearizable register like this:
; Construct a register at the path "/jepsen" with initial value 0.
(def a (avout/zk-atom conn "/jepsen" 0))
@a ; Read atom
(avout/reset!! a 1) ; Reset the atom's value to 1
(avout/swap!! a inc) ; Use the inc function to increment the atom's valueSo we'll need two pieces of state in our client: a connection, and an atom.
(defn client
"A client for a single compare-and-set register"
[conn a]
(reify client/Client
(setup! [_ test node]
(let [conn (avout/connect (name node))
a (avout/zk-atom conn "/jepsen" 0)]
(client conn a)))
(invoke! [this test op])
(teardown! [_ test]
(.close conn))))
(defn zk-test
[version]
(assoc tests/noop-test
:name "zookeeper"
:os debian/os
:db (db version)
:client (client nil nil)))Remember, the initial client has no connections--like a stem cell, it has the
potential to become an active client but doesn't do any work directly. We
call (client nil nil) to construct that initial client--its conn and atom
will be filled in when Jepsen calls setup!.
Running lein test, we can see the ZK connections opening and closing:
$ lein test
INFO org.apache.zookeeper.ZooKeeper - Client environment:zookeeper.version=3.4.0-1202560, built on 11/16/2011 07:18 GMT
...
INFO org.apache.zookeeper.ZooKeeper - Initiating client connection, connectString=n5 sessionTimeout=5000 watcher=zookeeper.internal$make_watcher$reify__7183@719567fd
INFO org.apache.zookeeper.ZooKeeper - Initiating client connection, connectString=n1 sessionTimeout=5000 watcher=zookeeper.internal$make_watcher$reify__7183@63a650df
INFO org.apache.zookeeper.ZooKeeper - Initiating client connection, connectString=n3 sessionTimeout=5000 watcher=zookeeper.internal$make_watcher$reify__7183@78007b36
INFO org.apache.zookeeper.ZooKeeper - Initiating client connection, connectString=n2 sessionTimeout=5000 watcher=zookeeper.internal$make_watcher$reify__7183@1e5a45c3
INFO org.apache.zookeeper.ZooKeeper - Initiating client connection, connectString=n4 sessionTimeout=5000 watcher=zookeeper.internal$make_watcher$reify__7183@21701c04
INFO org.apache.zookeeper.ClientCnxn - Opening socket connection to server /192.168.122.13:2181
INFO org.apache.zookeeper.ClientCnxn - Opening socket connection to server /192.168.122.11:2181
INFO org.apache.zookeeper.ClientCnxn - Opening socket connection to server /192.168.122.14:2181
INFO org.apache.zookeeper.ClientCnxn - Opening socket connection to server /192.168.122.12:2181
INFO org.apache.zookeeper.ClientCnxn - Opening socket connection to server /192.168.122.15:2181Now we have to actually do something with the client. Let's start with fifteen
seconds of reads, randomly staggered about a second apart. We'll pull in jepsen.generator to schedule operations, and jepsen.util's timeout capability:
(ns jepsen.zookeeper
(:require [avout.core :as avout]
[clojure.tools.logging :refer :all]
[clojure.java.io :as io]
[clojure.string :as str]
[jepsen [db :as db]
[client :as client]
[control :as c]
[generator :as gen]
[tests :as tests]
[util :refer [timeout]]]
[jepsen.os.debian :as debian]))... and write a simple generator: take reads, stagger them by about a second, give those operations to clients only (not the nemesis, which has other duties), and stop after 15 seconds.
(defn zk-test
[version]
(assoc tests/noop-test
:name "zookeeper"
:os debian/os
:db (db version)
:client (client nil nil)
:generator (->> r
(gen/stagger 1)
(gen/clients)
(gen/time-limit 15))))This fails, because we haven't told the client how to intepret these operations yet.
$ lein test
...
WARN jepsen.core - Process 24 indeterminate
java.lang.AssertionError: Assert failed: (= (:process op) (:process completion))The client's invoke! function needs to return a corresponding completion op,
with type :ok if the operation succeeded, :fail if it didn't take place, or
:info if we're not sure. invoke! can also throw an exception, which is
automatically converted to an :info.
(invoke! [this test op]
(timeout 5000 (assoc op :type :info, :error :timeout)
(case (:f op)
:read (assoc op :type :ok, :value @a))))timeout allows 5000 milliseconds for its body to run, and if that time runs
out, it calls (assoc op :type :info, :error :timeout) to construct a default
value. We dispatch based on the :f field of the operation, and when it's a
:read, we take the invoke op and return a copy of it, with :type :ok and
a value obtained by reading the register a.
$ lein test
...
INFO jepsen.util - 2 :invoke :read nil
INFO jepsen.util - 2 :ok :read 0
INFO jepsen.util - 2 :invoke :read nil
INFO jepsen.util - 2 :ok :read 0
INFO jepsen.util - 4 :invoke :read nil
INFO jepsen.util - 4 :ok :read 0Much better! We're successfully reading the initial value of 0. Let's add some
writes to the generator by replacing r with (gen/mix [r w]):
(defn zk-test
[version]
(assoc tests/noop-test
:name "zookeeper"
:os debian/os
:db (db version)
:client (client nil nil)
:generator (->> (gen/mix [r w])
(gen/stagger 1)
(gen/clients)
(gen/time-limit 15))))To handle those writes, we'll use avout/reset!!, and return the op with
:type :ok. If reset!! fails it'll throw, and Jepsen's machinery will
automatically convert it to an :info crash.
(invoke! [this test op]
(timeout 5000 (assoc op :type :info, :error :timeout)
(case (:f op)
:read (assoc op :type :ok, :value @a)
:write (do (avout/reset!! a (:value op))
(assoc op :type :ok)))))We'll confirm writes work by watching the test:
$ lein test
...
INFO jepsen.util - 1 :ok :read 1
INFO jepsen.util - 1 :invoke :read nil
INFO jepsen.util - 1 :ok :read 1
INFO jepsen.util - 2 :invoke :write 0
INFO jepsen.util - 2 :ok :write 0
INFO jepsen.util - 4 :invoke :write 2
INFO jepsen.util - 4 :ok :write 2
INFO jepsen.util - 3 :invoke :read nil
INFO jepsen.util - 3 :ok :read 2Seems reasonable! The final analysis is going to crash because we haven't told it how to check the system yet, but we'll get to that later. First, we'll finish the client by adding compare-and-set to the mix:
(gen/mix [r w cas])Handling CaS is a little trickier. Avout gives us a swap!! function, which
takes a function f and updates the atom's value atomically to (f current-value), returning the new value.
(invoke! [this test op]
(timeout 5000 (assoc op :type :info, :error :timeout)
(case (:f op)
:read (assoc op :type :ok, :value @a)
:write (do (avout/reset!! a (:value op))
(assoc op :type :ok))
:cas (let [[value value'] (:value op)]
(avout/swap!! a (fn [current]
(if (= current value)
value'
current)))
; So... did it change or not?
))))The let binding here uses destructuring: it breaks apart the [old-value new-value] pair from the operation's :value field into value and value'.
Our swap function compares the atom's current value to value, and if it's
equal, sets it to value'. Otherwise, it stays the same.
The tricky problem is that we have no way to know whether the CaS actually took
effect. We know the new value of the register, because swap!! returns it--but
we don't know if that's the unchanged value and the predicate comparison
failed, or if it's the new value and the predicate comparison succeeded. We
need another piece of mutable state to determine what happened. We'll use a
local Clojure atom (not an Avout atom, which is backed by Zookeeper!).
(invoke! [this test op]
(timeout 5000 (assoc op :type :info, :error :timeout)
(case (:f op)
:read (assoc op :type :ok, :value @a)
:write (do (avout/reset!! a (:value op))
(assoc op :type :ok))
:cas (let [[value value'] (:value op)
type (atom :fail)]
(avout/swap!! a (fn [current]
(if (= current value)
(do (reset! type :ok)
value')
(do (reset! type :fail)
current))))
(assoc op :type @type)))))We bind type to a fresh atom, whose value is initially :fail. In each
branch of the swap function, we use reset! to set the atom to either :ok or
:fail. When it comes time to return a completion operation, we just read off
type's current value.
$ lein test
INFO jepsen.util - 2 :invoke :write 2
INFO jepsen.util - 2 :ok :write 2
INFO jepsen.util - 0 :invoke :cas [2 3]
INFO jepsen.util - 0 :ok :cas [2 3]
INFO jepsen.util - 1 :invoke :cas [1 2]
INFO jepsen.util - 1 :fail :cas [1 2]We expect some CaS ops to fail because their predicate value doesn't match the current value, but a few (~1/5, since there are 5 possible values) should succeed.
With our client performing operations, it's time to analyze results using a checker.