Update exception handling for workloads (#158)

Author: brfrn169

scalardb/src/scalardb/core.clj

@@ -171,7 +171,8 @@
 (defn rollback-txs
   "Given transactions as a vector are rollbacked."
   [txs]
-  (doseq [tx txs] (.rollback tx)))
+  (doseq [tx txs] (try (.rollback tx)
+                       (catch Exception e (warn (.getMessage e))))))
 
 (defmacro with-retry
   "If the result of the body is nil, it retries it"

scalardb/src/scalardb/elle_append.clj

@@ -97,23 +97,26 @@
         (scalar/prepare-transaction-service! test))))
 
   (invoke! [_ test op]
-    (let [tx (try (scalar/start-transaction test)
-                  (catch Exception e
-                    (warn (.getMessage e))))
-          [seq-id txn] (:value op)]
-      (try
-        (when (<= @(:table-id test) seq-id)
-          ;; add tables for the next sequence
-          (add-tables test (inc seq-id)))
-        (let [txn' (mapv (partial tx-execute seq-id tx) txn)]
-          (.commit tx)
-          (assoc op :type :ok :value (independent/tuple seq-id txn')))
-        (catch UnknownTransactionStatusException _
-          (swap! (:unknown-tx test) conj (.getId tx))
-          (assoc op :type :info :error {:unknown-tx-status (.getId tx)}))
-        (catch Exception e
-          (scalar/try-reconnection! test scalar/prepare-transaction-service!)
-          (assoc op :type :fail :error {:crud-error (.getMessage e)})))))
+    (if-let [tx (try (scalar/start-transaction test)
+                     (catch Exception e (warn (.getMessage e))))]
+      (let [[seq-id txn] (:value op)]
+        (try
+          (when (<= @(:table-id test) seq-id)
+            ;; add tables for the next sequence
+            (add-tables test (inc seq-id)))
+          (let [txn' (mapv (partial tx-execute seq-id tx) txn)]
+            (.commit tx)
+            (assoc op :type :ok :value (independent/tuple seq-id txn')))
+          (catch UnknownTransactionStatusException _
+            (swap! (:unknown-tx test) conj (.getId tx))
+            (assoc op :type :info :error {:unknown-tx-status (.getId tx)}))
+          (catch Exception e
+            (scalar/rollback-txs [tx])
+            (scalar/try-reconnection! test scalar/prepare-transaction-service!)
+            (assoc op :type :fail :error {:crud-error (.getMessage e)}))))
+      (do
+        (scalar/try-reconnection! test scalar/prepare-transaction-service!)
+        (assoc op :type :fail :error {:tx-error "starting tx failed"}))))
 
   (close! [_ _])
 

scalardb/src/scalardb/elle_append_2pc.clj

@@ -41,24 +41,30 @@
     (let [tx1 (try (scalar/start-2pc test)
                    (catch Exception e
                      (warn (.getMessage e))))
-          tx2 (try (scalar/join-2pc test (.getId tx1))
-                   (catch Exception e
-                     (warn (.getMessage e))))
-          [seq-id txn] (:value op)]
-      (try
-        (when (<= @(:table-id test) seq-id)
-          ;; add tables for the next sequence
-          (append/add-tables test (inc seq-id)))
-        (let [txn' (mapv (partial tx-execute seq-id tx1 tx2) txn)]
-          (scalar/prepare-validate-commit-txs [tx1 tx2])
-          (assoc op :type :ok :value (independent/tuple seq-id txn')))
-        (catch UnknownTransactionStatusException _
-          (swap! (:unknown-tx test) conj (.getId tx1))
-          (assoc op :type :info :error {:unknown-tx-status (.getId tx1)}))
-        (catch Exception e
-          (scalar/rollback-txs [tx1 tx2])
+          tx2 (when tx1
+                (try (scalar/join-2pc test (.getId tx1))
+                     (catch Exception e
+                       (warn (.getMessage e)))))]
+      (if (and tx1 tx2)
+        (let [[seq-id txn] (:value op)]
+          (try
+            (when (<= @(:table-id test) seq-id)
+              ;; add tables for the next sequence
+              (append/add-tables test (inc seq-id)))
+            (let [txn' (mapv (partial tx-execute seq-id tx1 tx2) txn)]
+              (scalar/prepare-validate-commit-txs [tx1 tx2])
+              (assoc op :type :ok :value (independent/tuple seq-id txn')))
+            (catch UnknownTransactionStatusException _
+              (swap! (:unknown-tx test) conj (.getId tx1))
+              (assoc op :type :info :error {:unknown-tx-status (.getId tx1)}))
+            (catch Exception e
+              (scalar/rollback-txs [tx1 tx2])
+              (scalar/try-reconnection! test scalar/prepare-2pc-service!)
+              (assoc op :type :fail :error {:crud-error (.getMessage e)}))))
+        (do
+          (when tx1 (scalar/rollback-txs [tx1]))
           (scalar/try-reconnection! test scalar/prepare-2pc-service!)
-          (assoc op :type :fail :error {:crud-error (.getMessage e)})))))
+          (assoc op :type :fail :error {:tx-error "starting tx failed"})))))
 
   (close! [_ _])
 

scalardb/src/scalardb/elle_write_read.clj

@@ -1,5 +1,5 @@
 (ns scalardb.elle-write-read
-  (:require [clojure.tools.logging :refer [info]]
+  (:require [clojure.tools.logging :refer [info warn]]
             [jepsen.client :as client]
             [jepsen.generator :as gen]
             [jepsen.independent :as independent]
@@ -92,21 +92,26 @@
         (scalar/prepare-transaction-service! test))))
 
   (invoke! [_ test op]
-    (let [tx (scalar/start-transaction test)
-          [seq-id txn] (:value op)]
-      (try
-        (when (<= @(:table-id test) seq-id)
-          ;; add tables for the next sequence
-          (add-tables test (inc seq-id)))
-        (let [txn' (mapv (partial tx-execute seq-id tx) txn)]
-          (.commit tx)
-          (assoc op :type :ok :value (independent/tuple seq-id txn')))
-        (catch UnknownTransactionStatusException _
-          (swap! (:unknown-tx test) conj (.getId tx))
-          (assoc op :type :info :error {:unknown-tx-status (.getId tx)}))
-        (catch Exception e
-          (scalar/try-reconnection! test scalar/prepare-transaction-service!)
-          (assoc op :type :fail :error {:crud-error (.getMessage e)})))))
+    (if-let [tx (try (scalar/start-transaction test)
+                     (catch Exception e (warn (.getMessage e))))]
+      (let [[seq-id txn] (:value op)]
+        (try
+          (when (<= @(:table-id test) seq-id)
+            ;; add tables for the next sequence
+            (add-tables test (inc seq-id)))
+          (let [txn' (mapv (partial tx-execute seq-id tx) txn)]
+            (.commit tx)
+            (assoc op :type :ok :value (independent/tuple seq-id txn')))
+          (catch UnknownTransactionStatusException _
+            (swap! (:unknown-tx test) conj (.getId tx))
+            (assoc op :type :info :error {:unknown-tx-status (.getId tx)}))
+          (catch Exception e
+            (scalar/rollback-txs [tx])
+            (scalar/try-reconnection! test scalar/prepare-transaction-service!)
+            (assoc op :type :fail :error {:crud-error (.getMessage e)}))))
+      (do
+        (scalar/try-reconnection! test scalar/prepare-transaction-service!)
+        (assoc op :type :fail :error {:tx-error "starting tx failed"}))))
 
   (close! [_ _])
 

scalardb/src/scalardb/elle_write_read_2pc.clj

@@ -1,5 +1,6 @@
 (ns scalardb.elle-write-read-2pc
-  (:require [jepsen.client :as client]
+  (:require [clojure.tools.logging :refer [warn]]
+            [jepsen.client :as client]
             [jepsen.generator :as gen]
             [jepsen.independent :as independent]
             [scalardb.core :as scalar :refer [DEFAULT_TABLE_COUNT]]
@@ -29,23 +30,33 @@
         (scalar/prepare-2pc-service! test))))
 
   (invoke! [_ test op]
-    (let [tx1 (scalar/start-2pc test)
-          tx2 (scalar/join-2pc test (.getId tx1))
-          [seq-id txn] (:value op)]
-      (try
-        (when (<= @(:table-id test) seq-id)
-          ;; add tables for the next sequence
-          (wr/add-tables test (inc seq-id)))
-        (let [txn' (mapv (partial tx-execute seq-id tx1 tx2) txn)]
-          (scalar/prepare-validate-commit-txs [tx1 tx2])
-          (assoc op :type :ok :value (independent/tuple seq-id txn')))
-        (catch UnknownTransactionStatusException _
-          (swap! (:unknown-tx test) conj (.getId tx1))
-          (assoc op :type :info :error {:unknown-tx-status (.getId tx1)}))
-        (catch Exception e
-          (scalar/rollback-txs [tx1 tx2])
+    (let [tx1 (try (scalar/start-2pc test)
+                   (catch Exception e
+                     (warn (.getMessage e))))
+          tx2 (when tx1
+                (try (scalar/join-2pc test (.getId tx1))
+                     (catch Exception e
+                       (warn (.getMessage e)))))]
+      (if (and tx1 tx2)
+        (let [[seq-id txn] (:value op)]
+          (try
+            (when (<= @(:table-id test) seq-id)
+              ;; add tables for the next sequence
+              (wr/add-tables test (inc seq-id)))
+            (let [txn' (mapv (partial tx-execute seq-id tx1 tx2) txn)]
+              (scalar/prepare-validate-commit-txs [tx1 tx2])
+              (assoc op :type :ok :value (independent/tuple seq-id txn')))
+            (catch UnknownTransactionStatusException _
+              (swap! (:unknown-tx test) conj (.getId tx1))
+              (assoc op :type :info :error {:unknown-tx-status (.getId tx1)}))
+            (catch Exception e
+              (scalar/rollback-txs [tx1 tx2])
+              (scalar/try-reconnection! test scalar/prepare-2pc-service!)
+              (assoc op :type :fail :error {:crud-error (.getMessage e)}))))
+        (do
+          (when tx1 (scalar/rollback-txs [tx1]))
           (scalar/try-reconnection! test scalar/prepare-2pc-service!)
-          (assoc op :type :fail :error {:crud-error (.getMessage e)})))))
+          (assoc op :type :fail :error {:tx-error "starting tx failed"})))))
 
   (close! [_ _])
 

scalardb/src/scalardb/transfer.clj

@@ -113,6 +113,7 @@
         :unknown-tx-status)
       (catch Exception e
         (warn (.getMessage e))
+        (scalar/rollback-txs [tx])
         :fail))
     :start-fail))
 

scalardb/src/scalardb/transfer_2pc.clj

@@ -10,43 +10,43 @@
 
 (defn- tx-transfer
   [tx1 tx2 from to amount]
-  (try
-    (infof "Transferring %d from %d to %d (tx: %s)" amount from to (.getId tx1))
-    (let [fromResult (.get tx1 (transfer/prepare-get from))
-          toResult (.get tx2 (transfer/prepare-get to))]
-      (infof "fromID: %d, the balance: %d, the version: %d (tx: %s)" from (transfer/get-balance fromResult) (transfer/get-version fromResult) (.getId tx1))
-      (->> (transfer/calc-new-balance fromResult (- amount))
-           (transfer/prepare-put from)
-           (.put tx1))
-      (infof "toID: %d, the balance: %d, the version: %d (tx: %s)" to (transfer/get-balance toResult) (transfer/get-version toResult) (.getId tx1))
-      (->> (transfer/calc-new-balance toResult amount)
-           (transfer/prepare-put to)
-           (.put tx2)))
-    (scalar/prepare-validate-commit-txs [tx1 tx2])
-    (catch UnknownTransactionStatusException e
-      (throw e))
-    (catch Exception e
-      (scalar/rollback-txs [tx1 tx2])
-      (throw e))))
+  (infof "Transferring %d from %d to %d (tx: %s)" amount from to (.getId tx1))
+  (let [fromResult (.get tx1 (transfer/prepare-get from))
+        toResult (.get tx2 (transfer/prepare-get to))]
+    (infof "fromID: %d, the balance: %d, the version: %d (tx: %s)" from (transfer/get-balance fromResult) (transfer/get-version fromResult) (.getId tx1))
+    (->> (transfer/calc-new-balance fromResult (- amount))
+         (transfer/prepare-put from)
+         (.put tx1))
+    (infof "toID: %d, the balance: %d, the version: %d (tx: %s)" to (transfer/get-balance toResult) (transfer/get-version toResult) (.getId tx1))
+    (->> (transfer/calc-new-balance toResult amount)
+         (transfer/prepare-put to)
+         (.put tx2)))
+  (scalar/prepare-validate-commit-txs [tx1 tx2]))
 
 (defn- try-tx-transfer
   [test {:keys [from to amount]}]
   (let [tx1 (try (scalar/start-2pc test)
                  (catch Exception e
                    (warn (.getMessage e))))
-        tx2 (try (scalar/join-2pc test (.getId tx1))
-                 (catch Exception e
-                   (warn (.getMessage e))))]
-    (try
-      (tx-transfer tx1 tx2 from to amount)
-      :commit
-      (catch UnknownTransactionStatusException _
-        (swap! (:unknown-tx test) conj (.getId tx1))
-        (warn "Unknown transaction: " (.getId tx1))
-        :unknown-tx-status)
-      (catch Exception e
-        (warn (.getMessage e))
-        :fail))))
+        tx2 (when tx1
+              (try (scalar/join-2pc test (.getId tx1))
+                   (catch Exception e
+                     (warn (.getMessage e)))))]
+    (if (and tx1 tx2)
+      (try
+        (tx-transfer tx1 tx2 from to amount)
+        :commit
+        (catch UnknownTransactionStatusException _
+          (swap! (:unknown-tx test) conj (.getId tx1))
+          (warn "Unknown transaction: " (.getId tx1))
+          :unknown-tx-status)
+        (catch Exception e
+          (warn (.getMessage e))
+          (scalar/rollback-txs [tx1 tx2])
+          :fail))
+      (do
+        (when tx1 (scalar/rollback-txs [tx1]))
+        :start-fail))))
 
 (defrecord TransferClient [initialized? n initial-balance max-txs]
   client/Client

scalardb/src/scalardb/transfer_append.clj

@@ -127,6 +127,7 @@
         :unknown-tx-status)
       (catch Exception e
         (warn (.getMessage e))
+        (scalar/rollback-txs [tx])
         :fail))
     :start-fail))
 

scalardb/src/scalardb/transfer_append_2pc.clj

@@ -12,48 +12,49 @@
 
 (defn- tx-transfer
   [tx1 tx2 from to amount]
-  (try
-    (infof "Transferring %d from %d to %d (tx: %s)" amount from to (.getId tx1))
-    (let [^Result from-result
-          (t-append/scan-for-latest tx1 (t-append/prepare-scan-for-latest from))
-          ^Result to-result
-          (t-append/scan-for-latest tx2 (t-append/prepare-scan-for-latest to))]
-      (infof "fromID: %d, the latest balance: %d, the latest age: %d (tx: %s)" from (t-append/get-balance from-result) (t-append/get-age from-result) (.getId tx1))
-      (->> (t-append/prepare-put from
-                                 (t-append/calc-new-age from-result)
-                                 (t-append/calc-new-balance from-result
-                                                            (- amount)))
-           (.put tx1))
-      (infof "toID: %d, the latest balance: %d, the latest age: %d (tx: %s)" to (t-append/get-balance to-result) (t-append/get-age to-result) (.getId tx1))
-      (->> (t-append/prepare-put to
-                                 (t-append/calc-new-age to-result)
-                                 (t-append/calc-new-balance to-result amount))
-           (.put tx2)))
-    (scalar/prepare-validate-commit-txs [tx1 tx2])
-    (catch UnknownTransactionStatusException e
-      (throw e))
-    (catch Exception e
-      (scalar/rollback-txs [tx1 tx2])
-      (throw e))))
+  (infof "Transferring %d from %d to %d (tx: %s)" amount from to (.getId tx1))
+  (let [^Result from-result
+        (t-append/scan-for-latest tx1 (t-append/prepare-scan-for-latest from))
+        ^Result to-result
+        (t-append/scan-for-latest tx2 (t-append/prepare-scan-for-latest to))]
+    (infof "fromID: %d, the latest balance: %d, the latest age: %d (tx: %s)" from (t-append/get-balance from-result) (t-append/get-age from-result) (.getId tx1))
+    (->> (t-append/prepare-put from
+                               (t-append/calc-new-age from-result)
+                               (t-append/calc-new-balance from-result
+                                                          (- amount)))
+         (.put tx1))
+    (infof "toID: %d, the latest balance: %d, the latest age: %d (tx: %s)" to (t-append/get-balance to-result) (t-append/get-age to-result) (.getId tx1))
+    (->> (t-append/prepare-put to
+                               (t-append/calc-new-age to-result)
+                               (t-append/calc-new-balance to-result amount))
+         (.put tx2)))
+  (scalar/prepare-validate-commit-txs [tx1 tx2]))
 
 (defn- try-tx-transfer
   [test {:keys [from to amount]}]
   (let [tx1 (try (scalar/start-2pc test)
                  (catch Exception e
                    (warn (.getMessage e))))
-        tx2 (try (scalar/join-2pc test (.getId tx1))
-                 (catch Exception e
-                   (warn (.getMessage e))))]
-    (try
-      (tx-transfer tx1 tx2 from to amount)
-      :commit
-      (catch UnknownTransactionStatusException _
-        (swap! (:unknown-tx test) conj (.getId tx1))
-        (warn "Unknown transaction: " (.getId tx1))
-        :unknown-tx-status)
-      (catch Exception e
-        (warn (.getMessage e))
-        :fail))))
+        tx2 (if tx1
+              (try (scalar/join-2pc test (.getId tx1))
+                   (catch Exception e
+                     (warn (.getMessage e))))
+              nil)]
+    (if (and tx1 tx2)
+      (try
+        (tx-transfer tx1 tx2 from to amount)
+        :commit
+        (catch UnknownTransactionStatusException _
+          (swap! (:unknown-tx test) conj (.getId tx1))
+          (warn "Unknown transaction: " (.getId tx1))
+          :unknown-tx-status)
+        (catch Exception e
+          (warn (.getMessage e))
+          (scalar/rollback-txs [tx1 tx2])
+          :fail))
+      (do
+        (when tx1 (scalar/rollback-txs [tx1]))
+        :start-fail))))
 
 (defrecord TransferClient [initialized? n initial-balance max-txs]
   client/Client