swim.clj

(def t'
  "T' time units"
  1000)

(def k
  "Size of failure detection subgroups"
  5)

(def period
  "Monotonically increasing sequence of the current protocol period"
  (atom 0))

(def members
  "Shared member list"
  (atom #{}))

(def states
  {:alive 1 :suspect 2 :faulty 3})

(def ex-member
  {:host "127.0.0.1" :port 5555 :name "joe" :state})

(defn remove! [member]
  (swap! @members disj member))

(defn add! [member]
  (swap! @members conj member))

(defn fail! [member]
  (swap! @members assoc :state (:faulty)))

; msg broadcast / dissemination
(defn send-msg [type member & opts])

(defn ping [member]
  (send-msg :ping member))

(defn ping-req [target member]
  (send-msg :ping-req member target))

(defn ack [member]
  (send-msg :ack member))

(defn acked? [response]
  (:ack response))

(defn dead? [{:state state} member]
  (= state (:dead states)))

(defn alive? [{:state state} member]
  (= state (:alive states)))

(defn indirectly-ping [target]
  (let [members (cond-> (rand-members k)
                        (disj member))]
    (map #(ping-req %1 target) members)))

; basic SWIM
(defn basic-swim []
  (loop [(swap! period inc)
         member (rand-members 1)
         response (ping member)]
    (when-not (acked? response)
      (if not-any? acked? (indirectly-ping member)
        (fail! member)))
    (Thread/sleep t')))

; robust SWIM
(defn robust-swim []
  (loop [(swap! period inc)
         member (rand-members 1)
         response (ping member)]
    (when-not (acked? response)
      (if not-any? acked? (indirectly-ping member)
        (suspect! member)))
    (Thread/sleep t')))