knossos-queue-model-test.clj

(ns locksmithing.queue-model-test
  (:require [clojure.pprint :refer [pprint]]
            [clojure.test :refer :all]
            [knossos.core :refer [linearizations
                                  linearizable-prefix
                                  op
                                  ->Register]]
            [locksmithing.queue :refer [queue]])
  (:import [locksmithing.queue Node]
           [java.util.concurrent.atomic AtomicReference]))

;; Utils
(defmacro dothreads  ;; Stolen from knossos.core-test
  "Calls body in each of t threads, blocking until completion. Binds i to the
  thread number."
  [[i t] & body]
  `(->> (range ~t)
        (map (fn [~i] (future ~@body)))
        doall
        (map deref)
        dorun))

(defn update
  "Appends an operation to the history of a system."
  [sys op]
  (update-in sys [:history] conj op))

;; Model
(defn system
  "Our system models the logic of the lock-free queue implementation in
  locksmithing.queue.

  Queues are composed of Nodes and expose two operations: pop and push.

  Additionally we allot some number of competing threads, to model contention.
  "
  []
  {:queue   (queue)
   :history []})

;; Queue method "push" state transitions; points of linearization
(declare queue-push-start)
(declare queue-push-2)
(declare queue-push-3)
(declare queue-push-4)
(declare queue-push-complete)
(declare maybe-push-complete)

(def push-state      (atom 0))

(defn queue-push-start
  [sys & [v]]
  (let [v    (or v (swap! push-state inc))
        head (-> sys :queue .head-ref .get)]
    (-> sys
        (assoc :push-state
               {:v     v
                :head  head
                :head' (Node. v (AtomicReference. nil) head)})
        (assoc :push-transition queue-push-2)
        (update (op :push :invoke :write v)))))

(defn queue-push-2
  [sys]
  (let [v           (-> sys :push-state :v)
        head        (-> sys :push-state :head)
        head'       (-> sys :push-state :head')
        transition  (atom (constantly (queue-push-start sys v)))]

    (if head
      ;; 1a. in locksmithing.queue
      (do (when (.compareAndSet (.prev-ref head) nil head')
            (reset! transition queue-push-3))

          (-> sys
              (assoc :push-transition @transition)
              (update (op :push :info :queue-push-2 v))))

      ;; 1b. in locksmithing.queue
      (do (when (.compareAndSet (-> sys :queue .head-ref) head head')
            (reset! transition queue-push-complete))
          (.set (-> sys :queue .tail-ref) head')

          (-> sys
              (assoc :push-transition @transition)
              (update (op :push :info :queue-push-2 v))
              maybe-push-complete)))))

(defn queue-push-3
  [sys]
  (let [v           (-> sys :push-state :v)
        head        (-> sys :push-state :head)
        head'       (-> sys :push-state :head')
        transition  (atom (constantly (queue-push-start sys v)))]

    ;; 2. in locksmithing.queue
    (do (when (.compareAndSet (-> sys :queue .head-ref) head head')
          (reset! transition queue-push-complete))

        (-> sys
            (assoc :push-transition @transition)
            (update (op :push :info :queue-push-3 v))
            maybe-push-complete))))

(defn maybe-push-complete
  [sys]
  (let [transition (:push-transition sys)]
    (if (identical? transition queue-push-complete)
      (queue-push-complete sys)
      sys)))

(defn queue-push-complete
  [sys]
  (let [v (-> sys :push-state :v)]
    (-> sys
      (dissoc :push-transition)  ;; Reset the push-transition fn
      (update (op :push :ok :write v)))))

(defn queue-push
  [sys]
  (let [transition (or (:push-transition sys) queue-push-start)]
    (transition sys)))


;; Queue method "pop" state tranisitons; points of linearization
(declare queue-pop-start)
(declare queue-pop-2)
(declare queue-pop-3)
(declare queue-pop-complete)
(declare maybe-pop-complete)


(defn queue-pop-start
  [sys]
  (if-let [tail (-> sys :queue .tail-ref .get)]  ;; Cannot pop without tail
    (let [tail' (-> tail .prev-ref .get)
          head  (-> sys :queue .head-ref .get)
          v     (.data tail)]
      (-> sys
          (assoc :pop-state {:tail  tail
                             :tail' tail'
                             :head  head
                             :v     v})
          (assoc :pop-transition queue-pop-2)
          (update (op :pop :invoke :read v))))
    sys))

(defn queue-pop-2
  [sys]
  (let [v          (-> sys :pop-state :v)
        head       (-> sys :pop-state :head)
        tail       (-> sys :pop-state :tail)
        transition (atom (constantly (queue-pop-start sys)))]

    ;; 1. in locksmithing.queue
    (when (identical? tail head)
      (when (.compareAndSet (-> sys :queue .head-ref) head nil)
        (reset! transition queue-pop-3)))

    (-> sys
        (assoc :pop-transition @transition)
        (update (op :pop :info :queue-pop-2 v)))))

(defn queue-pop-3
  [sys]
  (let [v          (-> sys :pop-state :v)
        tail       (-> sys :pop-state :tail)
        tail'      (-> sys :pop-state :tail')
        transition (atom (constantly (queue-pop-start sys)))]

    ;; 2. in locksmithing.queue
    (when (.compareAndSet (-> sys :queue .tail-ref) tail tail')
      (reset! transition queue-pop-complete))

    (-> sys
        (assoc :pop-transition @transition)
        (update (op :pop :info :queue-pop-3 v)))))

(defn maybe-pop-complete
  [sys]
  (let [transition (:pop-transition sys)]
    (if (identical? transition queue-pop-complete)
      (queue-pop-complete sys)
      sys)))

(defn queue-pop-complete
  [sys]
  (let [v (-> sys :pop-state :v)]
    (-> sys
      (dissoc :pop-transition)  ;; Reset the pop-transition fn
      (update (op :pop :ok :read v)))))

(defn queue-pop
  [sys]
  (let [transition (or (:pop-transition sys) queue-pop-start)]
    (transition sys)))


(defn step
  "All method results from a given system state."
  [sys]
  (list (queue-push sys)
        (queue-pop  sys)))

(defn trajectory
  "Returns a system from a randomized trajectory, `depth` steps away from the
  given system."
  [sys depth]
  (if (zero? depth)
    sys
    (let [possibilities (step sys)]
      (if (empty? possibilities)
        ;; Dead end
        sys
        ;; Descend
        (recur (rand-nth possibilities)
               (dec depth))))))

(defn print-system
  [sys history]
  (let [linearizable (linearizable-prefix (->Register nil) history)]
    (locking *out*
      (println "\n\n### No linearizable history for system ###\n")
      (pprint (dissoc sys :history))
      (println "\nHistory:\n")
      (pprint linearizable)
      (println "\nUnable to linearize past this point!\n")
      (pprint (drop (count linearizable) history)))))

(deftest queue-model-test
  (dothreads [_ 4]
    (dotimes [_ 1e4]
      (let [sys (trajectory (system) 15)]
        ;; Is this system linearizable?
        (let [history (:history sys)
              linears (linearizations (->Register nil) history)]
          (when (empty? linears)
            (print-system system history))
          (is (not (empty? linears))))))))