FAB-11094 Fix deadlock in block iterator

The deliver service used to not close ledger iterators until after a
block had been committed.  With FAB-10799, the Deliver service more
proactively cleans up ledger resources.  This leads to a more likely
contention between the block iterator's Close() function and the Next()
function.

These two code paths acquire the same two mutexes, but in different
orders.  The Next() path always acquires the itr.mgr.cpInfoCond.L first,
then the itr.closeMarkerLock, while the Close() path inverts this order.
If both Next() and Close() are invoked at the same time by goroutines,
this can result in a deadlock where both mutexes lock and never unlock.

This further prevents all blocks from committing and begins to leak
memory resources.

Change-Id: I99180fec2639a62cdf1cd9a6ce8b33f91ce498b9
Signed-off-by: Jason Yellick <jyellick@us.ibm.com>

common/ledger/blkstorage/fsblkstorage/blocks_itr.go

@@ -92,11 +92,11 @@ func (itr *blocksItr) Next() (ledger.QueryResult, error) {
 
 // Close releases any resources held by the iterator
 func (itr *blocksItr) Close() {
+	itr.mgr.cpInfoCond.L.Lock()
+	defer itr.mgr.cpInfoCond.L.Unlock()
 	itr.closeMarkerLock.Lock()
 	defer itr.closeMarkerLock.Unlock()
 	itr.closeMarker = true
-	itr.mgr.cpInfoCond.L.Lock()
-	defer itr.mgr.cpInfoCond.L.Unlock()
 	itr.mgr.cpInfoCond.Broadcast()
 	if itr.stream != nil {
 		itr.stream.close()

common/ledger/blkstorage/fsblkstorage/blocks_itr_test.go

@@ -75,6 +75,39 @@ func TestBlockItrClose(t *testing.T) {
 	testutil.AssertNil(t, bh)
 }
 
+func TestRaceToDeadlock(t *testing.T) {
+	env := newTestEnv(t, NewConf(testPath(), 0))
+	defer env.Cleanup()
+	blkfileMgrWrapper := newTestBlockfileWrapper(env, "testLedger")
+	defer blkfileMgrWrapper.close()
+	blkfileMgr := blkfileMgrWrapper.blockfileMgr
+
+	blocks := testutil.ConstructTestBlocks(t, 5)
+	blkfileMgrWrapper.addBlocks(blocks)
+
+	for i := 0; i < 1000; i++ {
+		itr, err := blkfileMgr.retrieveBlocks(5)
+		if err != nil {
+			panic(err)
+		}
+		go func() {
+			itr.Next()
+		}()
+		itr.Close()
+	}
+
+	for i := 0; i < 1000; i++ {
+		itr, err := blkfileMgr.retrieveBlocks(5)
+		if err != nil {
+			panic(err)
+		}
+		go func() {
+			itr.Close()
+		}()
+		itr.Next()
+	}
+}
+
 func TestBlockItrCloseWithoutRetrieve(t *testing.T) {
 	env := newTestEnv(t, NewConf(testPath(), 0))
 	defer env.Cleanup()