gridgain · ahitrin · Nov 7, 2023 · Oct 30, 2023 · Nov 1, 2023 · Nov 1, 2023
diff --git a/ignite-3/src/jepsen/ignite3.clj b/ignite-3/src/jepsen/ignite3.clj
@@ -96,29 +96,26 @@
         (info node files)
         (into [] (.split files "\n"))))))
 
-(defn generator
-  [operations time-limit]
-  (->> (gen/mix operations)
+(defn wrap-generator
+  "Add default wrapper for generator (frequency, nemesis, time limit)."
+  [generator time-limit]
+  (->> generator
        (gen/stagger 1/10)
        (gen/nemesis
-         ; without "take 100", we fail into infinity here (most probably, during print)
-         (take 100 (cycle [(gen/sleep 5)
-                           {:type :info, :f :start}
-                           (gen/sleep 1)
-                           {:type :info, :f :stop}])))
+         (cycle [(gen/sleep 5)
+                 {:type :info, :f :start}
+                 (gen/sleep 1)
+                 {:type :info, :f :stop}]))
        (gen/time-limit time-limit)))
 
 (defn basic-test
   "Sets up the test parameters common to all tests."
   [options]
-  (info :opts options)
   (merge tests/noop-test
          (dissoc options :test-fns)
-         {:name    "basic-test"
-          :os      (case (:os options)
+         {:os      (case (:os options)
                      :centos centos/os
                      :debian debian/os
                      :noop jepsen.os/noop)
           :db      (db (:version options))
-          :pds     (:pds options)
           :nemesis (:nemesis options)}))
diff --git a/ignite-3/src/jepsen/ignite3/append.clj b/ignite-3/src/jepsen/ignite3/append.clj
@@ -0,0 +1,116 @@
+(ns jepsen.ignite3.append
+  "Append test.
+  Values are lists of integers. Each operation performs a transaction,
+  comprised of micro-operations which are either reads of some value (returning
+  the entire list) or appends (adding a single number to whatever the present
+  value of the given list is). We detect cycles in these transactions using
+  Jepsen's cycle-detection system."
+  (:require [clojure.tools.logging :as log]
+            [jepsen [client :as client]
+                    [ignite3 :as ignite3]
+                    [nemesis :as nemesis]]
+            [jepsen.tests.cycle.append :as app])
+  (:import (org.apache.ignite Ignite)
+           (org.apache.ignite.client IgniteClient)))
+
+(def table-name "APPEND")
+
+(def sql-create (str "create table if not exists " table-name "(key int primary key, vals varchar(1000))"))
+
+(def sql-insert (str "insert into " table-name " (key, vals) values (?, ?)"))
+
+(def sql-update (str "update " table-name " set vals = ? where key = ?"))
+
+(def sql-select (str "select * from " table-name " where key = ?"))
+
+(defn run-sql
+  "Run a SQL query. Return ResultSet instance that should be closed afterwards."
+  ([session query params] (run-sql session nil query params))
+  ([session txn query params]
+    (log/info query params)
+    (.execute session txn query (object-array params))))
+
+(defn invoke-op [^Ignite ignite [opcode k v]]
+  "Perform a single operation in separate transaction."
+  (let [txn (.begin (.transactions ignite))
+        sql (.sql ignite)]
+    (if
+      (= :r opcode)
+      (let [select-result
+              (with-open [session  (.createSession sql)
+                          rs       (run-sql session txn sql-select [k])]
+                (if (.hasNext rs)
+                  (let [raw-result (.stringValue (.next rs) 1)
+                        strings    (clojure.string/split raw-result #",")]
+                    (->> strings
+                         (map #(Integer/parseInt %))
+                         (into [])))
+                  []))]
+        (.commit txn)
+        [:r k select-result])
+      (with-open [session   (.createSession sql)
+                  read-rs   (run-sql session txn sql-select [k])]
+        (if (.hasNext read-rs)
+          ; update existing list
+          (let [old-list    (.stringValue (.next read-rs) 1)
+                new-list    (str old-list "," v)]
+            (with-open [write-rs (run-sql session txn sql-update [new-list k])]))
+          ; create a new list
+          (do
+            (with-open [write-rs (run-sql session txn sql-insert [k (str v)])])))
+        (.commit txn)
+        [opcode k v]))))
+
+(defrecord Client [^Ignite ignite]
+  client/Client
+  ;
+  (open! [this test node]
+    ; (log/info "Node: " node)
+    (let [ignite (.build (.addresses (IgniteClient/builder) (into-array [(str node ":10800")])))]
+      (assoc this :ignite ignite)))
+  ;
+  (setup! [this test]
+    (with-open [create-stmt (.createStatement (.sql ignite) sql-create)
+                session (.createSession (.sql ignite))
+                rs (run-sql session create-stmt [])]
+      (log/info "Table" table-name "created")))
+  ;
+  (invoke! [this test op]
+    ; (log/info "Received: " op)
+    (let [ops   (:value op)
+          result (map #(invoke-op ignite %) ops)
+          overall-result (assoc op
+                                :type :info
+                                :value (into [] result))]
+      ; (log/info "Returned: " overall-result)
+      overall-result))
+  ;
+  (teardown! [this test])
+  ;
+  (close! [this test]
+    (.close ignite)))
+
+(comment "for repl"
+
+(def c (client/open! (Client. nil) {} "127.0.0.1"))
+(client/setup! c {})
+
+(client/invoke! c {} {:type :invoke, :process 0, :f :txn, :value [[:r 9 nil]]})
+
+(client/invoke! c {} {:type :invoke, :process 1, :f :txn, :value [[:append 9 2]]})
+
+(client/close! c {})
+
+"/for repl"
+)
+
+(defn append-test
+  [opts]
+  (ignite3/basic-test
+    (merge
+      (let [test-ops {:consistency-models [:serializable]}]
+        {:name      "append-test"
+         :client    (Client. nil)
+         :checker   (app/checker test-ops)
+         :generator (ignite3/wrap-generator (app/gen test-ops) (:time-limit opts))})
+      opts)))
diff --git a/ignite-3/src/jepsen/ignite3/register.clj b/ignite-3/src/jepsen/ignite3/register.clj
@@ -3,6 +3,7 @@
   (:require [clojure.tools.logging :as log]
             [jepsen [checker :as checker]
                     [client :as client]
+                    [generator :as gen]
                     [ignite3 :as ignite3]
                     [independent :as independent]]
             [jepsen.checker.timeline :as timeline]
@@ -11,7 +12,7 @@
 
 (def table-name "REGISTER")
 
-(def key "k")
+(def common-key "k")
 
 (def sql-create (str"create table if not exists " table-name "(key varchar primary key, val int)"))
 
@@ -45,21 +46,21 @@
     (try
       (case (:f op)
         :read (let [kv-view (.keyValueView (.table (.tables ignite) table-name) String Integer)
-                    value (.get kv-view nil key)]
+                    value (.get kv-view nil common-key)]
                 (assoc op :type :ok :value value))
         :write (let [kv-view (.keyValueView (.table (.tables ignite) table-name) String Integer)]
-                 (.put kv-view nil key (:value op))
+                 (.put kv-view nil common-key (:value op))
                  (assoc op :type :ok))
         :cas (let [kv-view (.keyValueView (.table (.tables ignite) table-name) String Integer)
                    [value value'] (:value op)]
-               (assoc op :type (if (.replace kv-view nil key value value') :ok :fail))))))
+               (assoc op :type (if (.replace kv-view nil common-key value value') :ok :fail))))))
 
   (teardown! [this test])
 
   (close! [this test]
     (.close ignite)))
 
-(defn test
+(defn register-test
   [opts]
   (ignite3/basic-test
     (merge
@@ -69,5 +70,7 @@
                     (checker/compose
                       {:linearizable (checker/linearizable {:model (model/cas-register)})
                        :timeline  (timeline/html)}))
-       :generator (ignite3/generator [r w cas] (:time-limit opts))}
-      opts)))
+       :generator (ignite3/wrap-generator
+                    (gen/mix [r w cas])
+                    (:time-limit opts))}
+      opts)))
diff --git a/ignite-3/src/jepsen/ignite3/runner.clj b/ignite-3/src/jepsen/ignite3/runner.clj
@@ -1,15 +1,17 @@
 (ns jepsen.ignite3.runner
   "Runs Apache Ignite 3 tests."
   (:gen-class)
-  (:require [clojure.pprint :refer [pprint]]
-            [clojure.tools.logging :refer :all]
-            [jepsen.cli :as jc]
-            [jepsen.core :as jepsen]
-            [jepsen.ignite3.register :as register]))
+  (:require [clojure.pprint             :refer [pprint]]
+            [clojure.tools.logging      :refer :all]
+            [jepsen.cli                 :as jc]
+            [jepsen.core                :as jepsen]
+            [jepsen.ignite3.append      :as append]
+            [jepsen.ignite3.register    :as register]))
 
 (def tests
   "A map of test names to test constructors."
-  {"register" register/test})
+  {"append"     append/append-test
+   "register"   register/register-test})
 
 (def nemesis-types
   {"noop"                   jepsen.nemesis/noop})
@@ -35,7 +37,7 @@
 
 (defn log-test
   [t]
-  (info "Testing\n" (with-out-str (pprint t)))
+  (binding [*print-length* 100] (info "Testing\n" (with-out-str (pprint t))))
   t)
 
 (defn test-cmd
@@ -48,7 +50,6 @@
                                              :test :test-fns})))
            :usage (jc/test-usage)
            :run (fn [{:keys [options]}]
-                  (pprint options)
                   (doseq [i        (range (:test-count options))
                           test-fn  (:test-fns options)]
                     ; Rehydrate test and run