[FAB-10469] Gossip: Don't lock when sending

The gossip connection locks the connection to preserve FIFO order
when waiting for room in the output buffer to be available.

A gossip connection has a corresponding goroutine for each reading
or writing from the input/output buffer for the connection.

When a connection is ordered to send a message, it tries to enqueue
it into the output buffer, and if there is no room - it discards
the message to not block the application layer.

However, private data message shouldn't be discarded and the goroutine
sending them should wait for room to be available in the buffer.
The problem is, that this is done under a lock and thus the following
might occur:

peer p sends to peer q a private data message and q sends one to p.
If p's and q's output buffers are full, the goroutine that send()s
holds the lock, and waits for the output buffer to have room.

However, to drain the receive buffer on the other side, both
peers also need to obtain a lock on the connection.
This results in a distributed deadlock.

The most sensible fix in my opinion is to just no lock the connection
when waiting on the output buffer.

Change-Id: I63a64e9cf08364d2023d99f2bedb1e382765e6a8
Signed-off-by: yacovm <yacovm@il.ibm.com>

Author: yacovm

gossip/comm/comm_test.go

@@ -202,6 +202,50 @@ func TestViperConfig(t *testing.T) {
 	assert.Equal(t, 200, util.GetIntOrDefault("peer.gossip.sendBuffSize", 0))
 }
 
+func TestMutualParallelSendWithAck(t *testing.T) {
+	t.Parallel()
+
+	// This test tests concurrent and parallel sending of many (1000) messages
+	// from 2 instances to one another at the same time.
+
+	msgNum := 1000
+
+	comm1, _ := newCommInstance(15201, naiveSec)
+	comm2, _ := newCommInstance(15202, naiveSec)
+	defer comm1.Stop()
+	defer comm2.Stop()
+
+	acceptData := func(o interface{}) bool {
+		return o.(proto.ReceivedMessage).GetGossipMessage().IsDataMsg()
+	}
+
+	inc1 := comm1.Accept(acceptData)
+	inc2 := comm2.Accept(acceptData)
+
+	// Send a message from comm1 to comm2, to make the instances establish a preliminary connection
+	comm1.Send(createGossipMsg(), remotePeer(15202))
+	// Wait for the message to be received in comm2
+	<-inc2
+
+	for i := 0; i < msgNum; i++ {
+		go comm1.SendWithAck(createGossipMsg(), time.Second*5, 1, remotePeer(15202))
+	}
+
+	for i := 0; i < msgNum; i++ {
+		go comm2.SendWithAck(createGossipMsg(), time.Second*5, 1, remotePeer(15201))
+	}
+
+	go func() {
+		for i := 0; i < msgNum; i++ {
+			<-inc1
+		}
+	}()
+
+	for i := 0; i < msgNum; i++ {
+		<-inc2
+	}
+}
+
 func TestHandshake(t *testing.T) {
 	t.Parallel()
 	signer := func(msg []byte) ([]byte, error) {

gossip/comm/conn.go

@@ -256,22 +256,21 @@ func (conn *connection) send(msg *proto.SignedGossipMessage, onErr func(error),
 		conn.logger.Debug("Aborting send() to ", conn.info.Endpoint, "because connection is closing")
 		return
 	}
-	conn.Lock()
-	defer conn.Unlock()
 
 	m := &msgSending{
 		envelope: msg.Envelope,
 		onErr:    onErr,
 	}
 
-	if len(conn.outBuff) == util.GetIntOrDefault("peer.gossip.sendBuffSize", defSendBuffSize) {
+	if len(conn.outBuff) == cap(conn.outBuff) {
 		if conn.logger.IsEnabledFor(logging.DEBUG) {
 			conn.logger.Debug("Buffer to", conn.info.Endpoint, "overflowed, dropping message", msg.String())
 		}
 		if !shouldBlock {
 			return
 		}
 	}
+
 	conn.outBuff <- m
 }