Merge pull request #4 from chef/ss/concurrency-fixes

Fix two problems with :concurrent_connections option

2 files changed, 13 insertions, 3 deletions

diff --git a/lib/net/ssh/multi/server.rb b/lib/net/ssh/multi/server.rb
index bce228f..f8fa552 100644
--- a/lib/net/ssh/multi/server.rb
+++ b/lib/net/ssh/multi/server.rb
@@ -140,8 +140,9 @@ module Net; module SSH; module Multi
 
     # Returns +true+ if the session has been opened, and the session is currently
     # busy (as defined by Net::SSH::Connection::Session#busy?).
+    # Also returns false if the server has failed to connect.
     def busy?(include_invisible=false)
-      session && session.busy?(include_invisible)
+      !failed? && session && session.busy?(include_invisible)
     end
 
     # Closes this server's session. If the session has not yet been opened,
@@ -228,4 +229,4 @@ module Net; module SSH; module Multi
         session.postprocess(listeners & readers, listeners & writers)
       end
   end
-end; end; end
\ No newline at end of file
+end; end; end
diff --git a/lib/net/ssh/multi/session.rb b/lib/net/ssh/multi/session.rb
index 333fdf3..2e2468f 100644
--- a/lib/net/ssh/multi/session.rb
+++ b/lib/net/ssh/multi/session.rb
@@ -476,7 +476,12 @@ module Net; module SSH; module Multi
           return connection
         end
 
-        @open_connections += 1
+        # Only increment the open_connections count if the connection
+        # is not being forced. Incase of a force, it will already be
+        # incremented.
+        if !force
+          @open_connections += 1
+        end
       end
 
       begin
@@ -542,6 +547,10 @@ module Net; module SSH; module Multi
       count = concurrent_connections ? (concurrent_connections - open_connections) : @pending_sessions.length
       count.times do
         session = @pending_sessions.pop or break
+        # Increment the open_connections count here to prevent
+        # creation of connection thread again before that is
+        # incremented by the thread.
+        @session_mutex.synchronize { @open_connections += 1 }
         @connect_threads << Thread.new do
           session.replace_with(next_session(session.server, true))
         end