Hotfix #120 (#136)

* cluster: fix deadlock in cluster synchronisation (#120)

For a impressively thorough breakdown of the problem, see:
	#120 (comment)

Huge thanks to @dvic and @KJTsanaktsidis for the report and fix.

* readme: credit @dvic and @KJTsanaktsidis

Author: domodwyer

README.md

@@ -36,6 +36,7 @@ Further PR's (with tests) are welcome, but please maintain backwards compatibili
 * Gracefully recover from a temporarily unreachable server ([details](https://github.com/globalsign/mgo/pull/69))
 * Use JSON tags when no explicit BSON are tags set ([details](https://github.com/globalsign/mgo/pull/91))
 * Support [$changeStream](https://docs.mongodb.com/manual/changeStreams/) tailing on 3.6+ ([details](https://github.com/globalsign/mgo/pull/97))
+* Fix deadlock in cluster synchronisation ([details](https://github.com/globalsign/mgo/issues/120))
 
 ---
 
@@ -46,11 +47,13 @@ Further PR's (with tests) are welcome, but please maintain backwards compatibili
 * @carter2000
 * @cezarsa
 * @drichelson
+* @dvic
 * @eaglerayp
 * @feliixx
 * @fmpwizard
 * @idy
 * @jameinel
+* @KJTsanaktsidis
 * @gazoon
 * @mapete94
 * @peterdeka

cluster.go

@@ -179,7 +179,7 @@ func (cluster *mongoCluster) isMaster(socket *mongoSocket, result *isMasterResul
 		})
 	})
 
-	err := session.Run(cmd, result)
+	err := session.runOnSocket(socket, cmd, result)
 	session.Close()
 	return err
 }

cluster_test.go

@@ -1964,6 +1964,41 @@ func (s *S) TestConnectCloseConcurrency(c *C) {
 	wg.Wait()
 }
 
+func (s *S) TestNoDeadlockOnClose(c *C) {
+	if *fast {
+		// Unfortunately I seem to need quite a high dial timeout to get this to work
+		// on my machine.
+		c.Skip("-fast")
+	}
+
+	var shouldStop int32
+	atomic.StoreInt32(&shouldStop, 0)
+
+	listener, err := net.Listen("tcp4", "127.0.0.1:")
+	c.Check(err, Equals, nil)
+
+	go func() {
+		for atomic.LoadInt32(&shouldStop) == 0 {
+			sock, err := listener.Accept()
+			if err != nil {
+				// Probs just closed
+				continue
+			}
+			sock.Close()
+		}
+	}()
+	defer func() {
+		atomic.StoreInt32(&shouldStop, 1)
+		listener.Close()
+	}()
+
+	session, err := mgo.DialWithTimeout(listener.Addr().String(), 10*time.Second)
+	// If execution reaches here, the deadlock did not happen and all is OK
+	if session != nil {
+		session.Close()
+	}
+}
+
 func (s *S) TestSelectServers(c *C) {
 	if !s.versionAtLeast(2, 2) {
 		c.Skip("read preferences introduced in 2.2")

session.go

@@ -806,6 +806,15 @@ func (db *Database) Run(cmd interface{}, result interface{}) error {
 	return db.run(socket, cmd, result)
 }
 
+// runOnSocket does the same as Run, but guarantees that your command will be run
+// on the provided socket instance; if it's unhealthy, you will receive the error
+// from it.
+func (db *Database) runOnSocket(socket *mongoSocket, cmd interface{}, result interface{}) error {
+	socket.Acquire()
+	defer socket.Release()
+	return db.run(socket, cmd, result)
+}
+
 // Credential holds details to authenticate with a MongoDB server.
 type Credential struct {
 	// Username and Password hold the basic details for authentication.
@@ -2270,6 +2279,13 @@ func (s *Session) Run(cmd interface{}, result interface{}) error {
 	return s.DB("admin").Run(cmd, result)
 }
 
+// runOnSocket does the same as Run, but guarantees that your command will be run
+// on the provided socket instance; if it's unhealthy, you will receive the error
+// from it.
+func (s *Session) runOnSocket(socket *mongoSocket, cmd interface{}, result interface{}) error {
+	return s.DB("admin").runOnSocket(socket, cmd, result)
+}
+
 // SelectServers restricts communication to servers configured with the
 // given tags. For example, the following statement restricts servers
 // used for reading operations to those with both tag "disk" set to