Handle functions that return EAGAIN.

Do a naive immediate retry for now. Smarter retry patterns can come
later.

Author: conormcd

LICENSE

@@ -1,4 +1,4 @@
-Copyright (c) 2015, Conor McDermottroe
+Copyright (c) 2015-2016, Conor McDermottroe
 All rights reserved.
 
 Redistribution and use in source and binary forms, with or without

README.md

@@ -51,7 +51,7 @@ TODO. :)
 
 ## License
 
-Copyright (c) 2015, Conor McDermottroe
+Copyright (c) 2015-2016, Conor McDermottroe
 All rights reserved.
 
 Redistribution and use in source and binary forms, with or without

src/clj_libssh2/agent.clj

@@ -1,5 +1,5 @@
 (ns clj-libssh2.agent
-  (:require [clj-libssh2.error :refer [handle-errors]]
+  (:require [clj-libssh2.error :refer [handle-errors with-timeout]]
             [clj-libssh2.libssh2 :as libssh2]
             [clj-libssh2.libssh2.agent :as libssh2-agent])
   (:import [com.sun.jna.ptr PointerByReference]))
@@ -9,10 +9,13 @@
    the first entry."
   [session ssh-agent previous]
   (when (nil? previous)
-    (handle-errors session (libssh2-agent/list-identities ssh-agent)))
+    (handle-errors session
+      (with-timeout :agent
+        (libssh2-agent/list-identities ssh-agent))))
   (let [id (PointerByReference.)
         ret (handle-errors session
-              (libssh2-agent/get-identity ssh-agent id previous))]
+              (with-timeout :agent
+                (libssh2-agent/get-identity ssh-agent id previous)))]
     (case ret
       0 (.getValue id)
       1 nil
@@ -25,18 +28,23 @@
     (when (nil? ssh-agent)
       (throw (Exception. "Failed to initialize agent.")))
     (try
-      (handle-errors session (libssh2-agent/connect ssh-agent))
+      (handle-errors session
+        (with-timeout :agent
+          (libssh2-agent/connect ssh-agent)))
       (when-not (loop [success false
                        previous nil]
                   (if success
                     success
                     (if-let [id (get-identity session ssh-agent previous)]
                       (recur
-                        (= 0 (libssh2-agent/userauth ssh-agent username id))
+                        (= 0 (with-timeout :agent
+                               (libssh2-agent/userauth ssh-agent username id)))
                         id)
                       false)))
         (throw (Exception. "Failed to authenticate with the agent.")))
       true
       (finally
-        (handle-errors session (libssh2-agent/disconnect ssh-agent))
+        (handle-errors session
+          (with-timeout :agent
+            (libssh2-agent/disconnect ssh-agent)))
         (libssh2-agent/free ssh-agent)))))

src/clj_libssh2/authentication.clj

@@ -2,7 +2,7 @@
   (:require [clojure.java.io :refer [file]]
             [clj-libssh2.libssh2.userauth :as libssh2-userauth]
             [clj-libssh2.agent :as ssh-agent]
-            [clj-libssh2.error :refer [handle-errors]])
+            [clj-libssh2.error :refer [handle-errors with-timeout]])
   (:import clojure.lang.PersistentArrayMap))
 
 (defprotocol Credentials
@@ -38,19 +38,21 @@
     (when-not (.exists (file keyfile))
       (throw (Exception. (format "%s does not exist." keyfile)))))
   (handle-errors session
-    (libssh2-userauth/publickey-fromfile (:session session)
-                                         (:username credentials)
-                                         (:public-key credentials)
-                                         (:private-key credentials)
-                                         (:passphrase credentials)))
+    (with-timeout :auth
+      (libssh2-userauth/publickey-fromfile (:session session)
+                                           (:username credentials)
+                                           (:public-key credentials)
+                                           (:private-key credentials)
+                                           (:passphrase credentials))))
   true)
 
 (defmethod authenticate PasswordCredentials
   [session credentials]
   (handle-errors session
-    (libssh2-userauth/password (:session session)
-                               (:username credentials)
-                               (:password credentials)))
+    (with-timeout :auth
+      (libssh2-userauth/password (:session session)
+                                 (:username credentials)
+                                 (:password credentials))))
   true)
 
 (defmethod authenticate PersistentArrayMap

src/clj_libssh2/error.clj

@@ -3,6 +3,10 @@
             [clj-libssh2.libssh2.session :as libssh2-session])
   (:import [com.sun.jna.ptr IntByReference PointerByReference]))
 
+(def timeouts (atom {:agent 10000
+                     :auth 10000
+                     :session 10000}))
+
 (def error-messages
   "All of the error codes that are documented for libssh2 except for
    LIBSSH2_ERROR_SOCKET_NONE which despite its name is a generic error and
@@ -92,3 +96,33 @@
   `(let [res# (do ~@body)]
      (maybe-throw-error (:session ~session) res#)
      res#))
+
+(defn get-timeout
+  "Get a timeout value."
+  [name-or-value]
+  (or (get @timeouts name-or-value) name-or-value 1000))
+
+(defn set-timeout
+  "Update a named timeout value."
+  [timeout-name millis]
+  (swap! timeouts assoc timeout-name millis))
+
+(defmacro with-timeout
+  "Run some code that could return libssh2/ERROR_EAGAIN and if it does, retry
+   until the timeout is hit.
+
+   `timeout` can be either a number of milliseconds or a keyword referring to a
+   named timeout set using `set-timeout`.
+
+   This will *not* interrupt a blocking function as this is usually used with
+   native functions which probably should not be interrupted."
+  [timeout & body]
+  `(let [start# (System/currentTimeMillis)
+         timeout# (get-timeout ~timeout)]
+     (loop [timedout# false]
+       (if timedout#
+         (throw (Exception. "Timeout!"))
+         (let [r# (do ~@body)]
+           (if (= r# libssh2/ERROR_EAGAIN)
+             (recur (< timeout# (- (System/currentTimeMillis) start#)))
+             r#))))))

src/clj_libssh2/session.clj

@@ -2,7 +2,7 @@
   (:require [clj-libssh2.libssh2 :as libssh2]
             [clj-libssh2.libssh2.session :as libssh2-session]
             [clj-libssh2.authentication :refer [authenticate]]
-            [clj-libssh2.error :refer [handle-errors]]
+            [clj-libssh2.error :refer [handle-errors with-timeout]]
             [clj-libssh2.socket :as socket]))
 
 (def sessions (atom {}))
@@ -22,14 +22,20 @@
   ([^Session session]
    (destroy-session session "Shutting down normally." false))
   ([^Session {session :session} ^String reason ^Boolean raise]
-   (handle-errors nil (libssh2-session/disconnect session reason))
-   (handle-errors nil (libssh2-session/free session))
+   (handle-errors nil
+    (with-timeout :session
+      (libssh2-session/disconnect session reason)))
+   (handle-errors nil
+    (with-timeout :session
+      (libssh2-session/free session)))
    (when raise
      (throw (Exception. reason)))))
 
 (defn- handshake
   [^Session {session :session socket :socket :as s}]
-  (handle-errors s (libssh2-session/handshake session socket)))
+  (handle-errors s
+    (with-timeout :session
+      (libssh2-session/handshake session socket))))
 
 (defn- session-id
   "Generate the session ID that will be used to pool"

test/clj_libssh2/test_error.clj

@@ -39,3 +39,47 @@
     (is (thrown-with-msg? Exception
                           #".+"
                           (error/maybe-throw-error nil -10000)))))
+
+(deftest with-timeout-works
+  (let [test-func (fn [& args]
+                    (let [state (atom (vec args))]
+                      (fn []
+                        (when (empty? @state)
+                          (throw (Exception. "Called too many times!")))
+                        (let [result (first @state)
+                              new-state (rest @state)]
+                          (reset! state new-state)
+                          (if (instance? clojure.lang.Fn result)
+                            (result)
+                            result)))))]
+    (testing "test-func behaves as expected"
+      (let [f (test-func 1 (constantly 2) 3)]
+        (is (= 1 (f)))
+        (is (= 2 (f)))
+        (is (= 3 (f)))
+        (is (thrown? Exception (f)))))
+    (testing "with-timeout doesn't retry successful function calls"
+      (let [f (test-func 0)]
+        (is (= 0 (error/with-timeout 10000 (f))))))
+    (testing "with-timeout doesn't retry failed function calls"
+      (let [f (test-func libssh2/ERROR_ALLOC)]
+        (is (= libssh2/ERROR_ALLOC (error/with-timeout 10000 (f))))))
+    (testing "with-timeout retries when it sees EAGAIN"
+      (let [f (test-func libssh2/ERROR_EAGAIN 0)]
+        (is (= 0 (error/with-timeout 10000 (f))))))
+    (testing "with-timeout doesn't retry exceptions"
+      (let [f (test-func #(throw (Exception. "")) 0)]
+        (is (thrown-with-msg? Exception #"" (error/with-timeout 10000 (f))))
+        (is (= 0 (error/with-timeout 10000 (f))))))
+    (testing "with-timeout obeys the timeout"
+      (let [f (test-func #(do (Thread/sleep 20)
+                              libssh2/ERROR_EAGAIN)
+                         0)]
+        (is (thrown? Exception (error/with-timeout 10 (f))))))
+    (testing "with-timeout can deal with fast-returning functions."
+      (let [f (constantly libssh2/ERROR_EAGAIN)]
+        (is (thrown? Exception (error/with-timeout 100 (f))))))
+    (testing "with-timeout can use symbolic times"
+      (with-redefs [error/timeouts (atom {:sym 5000})]
+        (let [f (test-func libssh2/ERROR_EAGAIN 1)]
+          (is (= 1 (error/with-timeout :sym (f)))))))))