Merge pull request globalsign#16 from joomcode/feature/busy-socket-reuse

Fix busy socket reuse and kill iterator from another connection

Author: bozaro

session.go

@@ -4279,10 +4279,20 @@ func (iter *Iter) Close() error {
 		}
 		return err
 	}
+
 	socket, err := iter.acquireSocket()
 	if err == nil {
+		var assassin *mongoSocket
 		// TODO Batch kills.
-		err = socket.Query(&killCursorsOp{[]int64{cursorId}})
+		if socket.isBusy() {
+			assassin, _, _ = iter.server.acquireSocketInternal(socket.dialInfo, false)
+		}
+		if assassin != nil {
+			err = assassin.Query(&killCursorsOp{[]int64{cursorId}})
+			assassin.Release()
+		} else {
+			err = socket.Query(&killCursorsOp{[]int64{cursorId}})
+		}
 		socket.Release()
 	}
 

socket.go

@@ -204,7 +204,12 @@ func newSocket(server *mongoServer, conn net.Conn, info *DialInfo) *mongoSocket
 	stats.socketsAlive(+1)
 	debugf("Socket %p to %s: initialized", socket, socket.addr)
 	socket.resetNonce()
-	go socket.readLoop()
+	go func() {
+		socket.readLoop()
+		for _, _ = range socket.replyFuncs {
+			socket.Release()
+		}
+	}()
 	return socket
 }
 
@@ -643,6 +648,7 @@ func (socket *mongoSocket) Query(ops ...interface{}) (err error) {
 		request := &requests[i]
 		setInt32(buf, request.bufferPos+4, int32(requestId))
 		socket.replyFuncs[requestId] = request.replyFunc
+		socket.references++
 		requestId++
 	}
 	socket.Unlock()
@@ -767,10 +773,20 @@ func (socket *mongoSocket) readLoop() {
 		}
 		socket.Unlock()
 
+		if replyFunc != nil {
+			socket.Release()
+		}
 		// XXX Do bound checking against totalLen.
 	}
 }
 
+func (socket *mongoSocket) isBusy() bool {
+	socket.Lock()
+	busy := len(socket.replyFuncs) != 0
+	socket.Unlock()
+	return busy
+}
+
 var emptyHeader = []byte{0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0}
 
 func addHeader(b []byte, opcode int) []byte {