control: Add concurrency tests, semaphores to clj-ssh and sshj

Both of these remotes now have concurrency safety tests designed to
stress their behavior under rapid-fire multithreaded calls. Both were
susceptible to issues related, I think, to a server-side constraint on
concurrent channels. I've added semaphores to both remotes, which seems
to fix the issue!

Author: aphyr

jepsen/src/jepsen/control.clj

@@ -5,15 +5,16 @@
 
   Note that a whole bunch of this namespace refers to things as 'ssh',
   although they really can apply to any remote, not just SSH."
-  (:import java.io.File)
   (:require [clj-ssh.ssh    :as ssh]
             [jepsen.util    :as util :refer [real-pmap with-thread-name]]
             [dom-top.core :refer [with-retry]]
             [jepsen.reconnect :as rc]
             [clojure.string :as str]
             [clojure.java.io :as io]
             [clojure.tools.logging :refer [warn info debug error]]
-            [slingshot.slingshot :refer [try+ throw+]]))
+            [slingshot.slingshot :refer [try+ throw+]])
+  (:import java.io.File
+           (java.util.concurrent Semaphore)))
 
 (defprotocol Remote
   (connect [this host]
@@ -349,7 +350,9 @@
       (ssh/connect))))
 
 
-(defrecord SSHRemote [session]
+(defrecord SSHRemote [concurrency-limit
+                      session
+                      semaphore]
   Remote
   (connect [this host]
     (assoc this :session (if *dummy*
@@ -360,13 +363,19 @@
                               (throw+ (merge (debug-data)
                                              {:type ::session-error
                                               :message "Error opening SSH session. Verify username, password, and node hostnames are correct."
-                                              :host host})))))))
+                                              :host host})))))
+           :semaphore (Semaphore. concurrency-limit true)))
 
   (disconnect! [_]
     (when-not (:dummy session) (ssh/disconnect session)))
 
   (execute! [_ action]
-    (when-not (:dummy session) (ssh/ssh session action)))
+    (when-not (:dummy session)
+      (.acquire semaphore)
+      (try
+        (ssh/ssh session action)
+        (finally
+          (.release semaphore)))))
 
   (upload! [_ local-paths remote-path rest]
     (when-not (:dummy session)
@@ -376,7 +385,17 @@
     (when-not (:dummy session)
       (apply ssh/scp-from session remote-paths local-path rest))))
 
-(def ssh "A remote that does things via clj-ssh." (SSHRemote. nil))
+(def concurrency-limit
+  "OpenSSH has a standard limit of 10 concurrent channels per connection.
+  However, commands run in quick succession with 10 concurrent *also* seem to
+  blow out the channel limit--perhaps there's an asynchronous channel teardown
+  process. We set the limit a bit lower here. This is experimentally determined
+  for clj-ssh by running jepsen.control-test's integration test... <sigh>"
+  8)
+
+(def ssh
+  "A remote that does things via clj-ssh."
+  (SSHRemote. concurrency-limit nil nil))
 
 (defn session
   "Wraps control session in a wrapper for reconnection."

jepsen/src/jepsen/control/sshj.clj

@@ -3,7 +3,7 @@
   jepsen.control's use of clj-ssh with this instead."
   (:require [byte-streams :as bs]
             [clojure.tools.logging :refer [info warn]]
-            [jepsen [control :refer :all]]
+            [jepsen [control :as c]]
             [slingshot.slingshot :refer [try+ throw+]])
   (:import (com.jcraft.jsch.agentproxy AgentProxy
                                        ConnectorFactory)
@@ -15,7 +15,8 @@
            (net.schmizz.sshj.userauth.method AuthMethod)
            (net.schmizz.sshj.xfer FileSystemFile)
            (java.io IOException)
-           (java.util.concurrent TimeUnit)))
+           (java.util.concurrent Semaphore
+                                 TimeUnit)))
 
 (defn auth-methods
   "Returns a list of AuthMethods we can use for logging in via an AgentProxy."
@@ -36,38 +37,42 @@
   to username/password."
   [^SSHClient c]
   (or ; Try given key
-      (when-let [k *private-key-path*]
-        (.authPublickey c *username* (into-array [k]))
+      (when-let [k c/*private-key-path*]
+        (.authPublickey c c/*username* (into-array [k]))
         true)
 
       ; Try agent
       (try
         (let [agent-proxy (agent-proxy)
               methods (auth-methods agent-proxy)]
-          (.auth c *username* methods)
+          (.auth c c/*username* methods)
           true)
         (catch UserAuthException e
           false))
 
       ; Fall back to standard id_rsa/id_dsa keys
-      (try (.authPublickey c ^String *username*)
+      (try (.authPublickey c ^String c/*username*)
            true
            (catch UserAuthException e
              false))
 
       ; OK, standard keys didn't work, try username+password
-      (.authPassword c *username* *password*)))
+      (.authPassword c c/*username* c/*password*)))
 
-(defrecord SSHJRemote [^SSHClient client]
+(defrecord SSHJRemote [concurrency-limit
+                       ^SSHClient client
+                       ^Semaphore semaphore]
   jepsen.control/Remote
   (connect [this host]
     (try+ (let [c (doto (SSHClient.)
                     (.loadKnownHosts)
-                    (.connect host *port*)
+                    (.connect host c/*port*)
                     auth!)]
-            (assoc this :client c))
+            (assoc this
+                   :client c
+                   :semaphore (Semaphore. concurrency-limit true)))
           (catch Exception e
-            (throw+ (assoc (debug-data)
+            (throw+ (assoc (c/debug-data)
                            :type    :jepsen.control/session-error
                            :message "Error opening SSH session. Verify username, password, and node hostnames are correct."
                            :host    host)))))
@@ -77,21 +82,26 @@
       (.close c)))
 
   (execute! [this action]
-    (with-open [session (.startSession client)]
-      (let [cmd (.exec session (:cmd action))]
-        ; Feed it input
-        (when-let [input (:in action)]
-          (let [stream (.getOutputStream cmd)]
-            (bs/transfer input stream)))
-        ; Wait on command
-        (.join cmd)
-        ; Return completion
-        (assoc action
-               :out   (.toString (IOUtils/readFully (.getInputStream cmd)))
-               :err   (.toString (IOUtils/readFully (.getErrorStream cmd)))
-               ; There's also a .getExitErrorMessage that might be interesting
-               ; here?
-               :exit  (.getExitStatus cmd)))))
+  ;  (info :permits (.availablePermits semaphore))
+    (.acquire semaphore)
+    (try
+      (with-open [session (.startSession client)]
+        (let [cmd (.exec session (:cmd action))]
+          ; Feed it input
+          (when-let [input (:in action)]
+            (let [stream (.getOutputStream cmd)]
+              (bs/transfer input stream)))
+          ; Wait on command
+          (.join cmd)
+          ; Return completion
+          (assoc action
+                 :out   (.toString (IOUtils/readFully (.getInputStream cmd)))
+                 :err   (.toString (IOUtils/readFully (.getErrorStream cmd)))
+                 ; There's also a .getExitErrorMessage that might be interesting
+                 ; here?
+                 :exit  (.getExitStatus cmd))))
+      (finally
+        (.release semaphore))))
 
   (upload! [this local-paths remote-path more]
     (with-open [sftp (.newSFTPClient client)]
@@ -101,7 +111,15 @@
     (with-open [sftp (.newSFTPClient client)]
       (.get sftp remote-paths (FileSystemFile. local-path)))))
 
+(def concurrency-limit
+  "OpenSSH has a standard limit of 10 concurrent channels per connection.
+  However, commands run in quick succession with 10 concurrent *also* seem to
+  blow out the channel limit--perhaps there's an asynchronous channel teardown
+  process. We set the limit a bit lower here. This is experimentally determined
+  by running jepsen.control-test's integration test... <sigh>"
+  6)
+
 (defn remote
   "Constructs an SSHJ remote."
   []
-  (SSHJRemote. nil))
+  (SSHJRemote. concurrency-limit nil nil))

jepsen/test/jepsen/control_test.clj

@@ -1,11 +1,13 @@
 (ns jepsen.control-test
-  (:require [jepsen [control :as c]
+  (:require [clojure [string :as str]
+                     [test :refer :all]]
+            [clojure.java.io :as io]
+            [clojure.tools.logging :refer [info warn]]
+            [jepsen [control :as c]
                     [common-test :refer [quiet-logging]]
-                    [util :refer [contains-many?]]]
+                    [util :refer [contains-many? real-pmap]]]
             [jepsen.control.sshj :as sshj]
-            [slingshot.slingshot :refer [try+ throw+]]
-            [clojure.java.io :as io]
-            [clojure.test :refer :all])
+            [slingshot.slingshot :refer [try+ throw+]])
   (:import (java.io File)))
 
 (use-fixtures :once quiet-logging)
@@ -21,7 +23,8 @@
       (testing "on failure, session throws debug data"
         (try+
           (c/on "thishostshouldnotresolve"
-                (c/exec :echo "hello"))
+                (c/exec :echo "hello")
+                (is false))
           (catch [:type :jepsen.control/session-error] e
             (is (= "Error opening SSH session. Verify username, password, and node hostnames are correct." (:message e)))
             (is (= "thishostshouldnotresolve" (:host e)))
@@ -61,10 +64,39 @@
                 (finally
                   (.delete tmp)))))
 
-          (c/exec :rm :-f remote-path))))))
+          (c/exec :rm :-f remote-path))
+
+        (testing "thread safety"
+          (let [; Which nodes are we going to act over?
+                nodes ["n1" "n2" "n3" "n4" "n5"]
+                ; Concurrency per node
+                concurrency 10
+                ; Makes the string we echo bigger or smaller
+                str-count 10
+                ; How many times do we run each command?
+                rep-count 5
+                ; What strings are we going to echo on each channel?
+                xs  (->> (range concurrency)
+                         (map (fn [i]
+                                (str/join "," (repeat str-count i)))))
+                t0 (System/nanoTime)]
+            ; On each each node, and on concurrency channels, that string
+            ; several times, making sure it comes back properly.
+            (->> (for [node nodes, x xs]
+                   (future
+                     (dotimes [i rep-count]
+                       ;(info :call node :i i :x x)
+                       (is (= x (c/exec :echo x))))))
+                 vec
+                 (mapv deref))
+            ;(info :time (-> (System/nanoTime) (- t0) (/ 1e6)) "ms")
+            ))))))
+
 
 (deftest ^:integration ssh-remote-test
+  (info :clj-ssh)
   (test-remote c/ssh))
 
 (deftest ^:integration sshj-remote-test
+  (info :sshj)
   (test-remote (sshj/remote)))