Remove redundant lock (#469)

* Remove redundant lock

Signed-off-by: Liran Funaro <[email protected]>

Author: liran-funaro

internal/bcdb/transaction_processor.go

@@ -1,11 +1,10 @@
 // Copyright IBM Corp. All Rights Reserved.
 // SPDX-License-Identifier: Apache-2.0
+
 package bcdb
 
 import (
 	"encoding/json"
-	"fmt"
-	"sync"
 	"time"
 
 	"github.com/google/uuid"
@@ -45,7 +44,6 @@ type transactionProcessor struct {
 	blockStore           *blockstore.Store
 	pendingTxs           *queue.PendingTxs
 	logger               *logger.SugarLogger
-	sync.Mutex
 }
 
 type txProcessorConfig struct {
@@ -266,37 +264,47 @@ func (t *transactionProcessor) SubmitTransaction(tx interface{}, timeout time.Du
 		return nil, err
 	}
 
-	t.Lock()
-	duplicate, err := t.isTxIDDuplicate(txID)
-	if err != nil {
-		t.Unlock()
-		return nil, err
-	}
-	if duplicate {
-		t.Unlock()
+	// We attempt to insert the txID atomically.
+	// If we succeed, then future TX fill fail at this point.
+	// However, if the TX already exists in the block store, then we will fail the subsequent check.
+	// Since a TX will be removed from the pending queue only after it is inserted to the block store,
+	// then it is guaranteed that we won't use the same txID twice.
+	// TODO: add limit on the number of pending sync tx
+	promise := queue.NewCompletionPromise(timeout)
+	if existed := t.pendingTxs.Add(txID, promise); existed {
 		return nil, &internalerror.DuplicateTxIDError{TxID: txID}
 	}
 
-	if t.txQueue.IsFull() {
-		t.Unlock()
-		return nil, fmt.Errorf("transaction queue is full. It means the server load is high. Try after sometime")
+	duplicate, err := t.blockStore.DoesTxIDExist(txID)
+	if err != nil || duplicate {
+		t.pendingTxs.DeleteWithNoAction(txID)
+		if err == nil {
+			err = &internalerror.DuplicateTxIDError{TxID: txID}
+		}
+		return nil, err
 	}
 
-	jsonBytes, err := json.MarshalIndent(tx, "", "\t")
-	if err != nil {
-		t.Unlock()
-		return nil, fmt.Errorf("failed to marshal transaction: %v", err)
+	// Avoids marshaling the TX in production mode
+	if t.logger.IsDebug() {
+		if jsonBytes, err := json.MarshalIndent(tx, "", "\t"); err != nil {
+			t.logger.Debugf("failed to marshal transaction: %v", err)
+		} else {
+			t.logger.Debugf("enqueuing transaction %s\n", jsonBytes)
+		}
 	}
-	t.logger.Debugf("enqueuing transaction %s\n", string(jsonBytes))
 
-	t.txQueue.Enqueue(tx)
+	if timeout <= 0 {
+		// Enqueue will block until the queue is not full
+		t.txQueue.Enqueue(tx)
+	} else {
+		// EnqueueWithTimeout will block until the queue is not full or timeout occurs
+		if success := t.txQueue.EnqueueWithTimeout(tx, timeout); !success {
+			t.pendingTxs.DeleteWithNoAction(txID)
+			return nil, &internalerror.TimeoutErr{ErrMsg: "timeout has occurred while inserting the transaction to the queue"}
+		}
+	}
 	t.logger.Debug("transaction is enqueued for re-ordering")
 
-	promise := queue.NewCompletionPromise(timeout)
-	// TODO: add limit on the number of pending sync tx
-	t.pendingTxs.Add(txID, promise)
-	t.Unlock()
-
 	receipt, err := promise.Wait()
 
 	if err != nil {
@@ -336,49 +344,31 @@ func (t *transactionProcessor) PostBlockCommitProcessing(block *types.Block) err
 		return errors.Errorf("unexpected transaction envelope in the block")
 	}
 
-	t.pendingTxs.DoneWithReceipt(txIDs, block.Header)
+	go t.pendingTxs.DoneWithReceipt(txIDs, block.Header)
 
 	return nil
 }
 
-func (t *transactionProcessor) isTxIDDuplicate(txID string) (bool, error) {
-	if t.pendingTxs.Has(txID) {
-		return true, nil
-	}
-
-	isTxIDAlreadyCommitted, err := t.blockStore.DoesTxIDExist(txID)
-	if err != nil {
-		return false, err
-	}
-	return isTxIDAlreadyCommitted, nil
-}
-
 func (t *transactionProcessor) Close() error {
-	t.Lock()
-	defer t.Unlock()
-
+	// It is safe to use without locks because all the following calls are protected internally.
 	t.txReorderer.Stop()
 	t.blockCreator.Stop()
-	t.blockReplicator.Close()
+	_ = t.blockReplicator.Close()
 	t.peerTransport.Close()
 	t.blockProcessor.Stop()
 
 	return nil
 }
 
 func (t *transactionProcessor) IsLeader() *internalerror.NotLeaderError {
-	t.Lock()
-	defer t.Unlock()
-
+	// It is safe to use without locks because the following call is protected internally.
 	return t.blockReplicator.IsLeader()
 }
 
 // ClusterStatus returns the leader NodeID, and the active nodes NodeIDs.
 // Note: leader is always in active.
 func (t *transactionProcessor) ClusterStatus() (leader string, active []string) {
-	t.Lock()
-	defer t.Unlock()
-
+	// It is safe to use without locks because the following call is protected internally.
 	leaderID, activePeers := t.blockReplicator.GetClusterStatus()
 	for _, peer := range activePeers {
 		active = append(active, peer.NodeId)

internal/blockcreator/blockcreator_test.go

@@ -332,7 +332,7 @@ func TestBlockCreator_ReleaseAsync(t *testing.T) {
 	})
 
 	for i := 1; i < 6; i++ {
-		testEnv.pendingTxs.Add(fmt.Sprintf("txid:%d", i), nil)
+		require.False(t, testEnv.pendingTxs.Add(fmt.Sprintf("txid:%d", i), nil))
 	}
 
 	for _, txBatch := range txBatches {
@@ -358,7 +358,7 @@ func TestBlockCreator_ReleaseSync(t *testing.T) {
 	wg.Add(5)
 	for i := 1; i < 6; i++ {
 		promise := queue.NewCompletionPromise(5 * time.Second)
-		testEnv.pendingTxs.Add(fmt.Sprintf("txid:%d", i), promise)
+		require.False(t, testEnv.pendingTxs.Add(fmt.Sprintf("txid:%d", i), promise))
 		go func() {
 			receipt, err := promise.Wait()
 			require.Nil(t, receipt)

internal/blockprocessor/processor.go

@@ -4,9 +4,9 @@
 package blockprocessor
 
 import (
-	"github.com/hyperledger-labs/orion-server/config"
 	"sync"
 
+	"github.com/hyperledger-labs/orion-server/config"
 	"github.com/hyperledger-labs/orion-server/internal/blockstore"
 	"github.com/hyperledger-labs/orion-server/internal/mptrie"
 	"github.com/hyperledger-labs/orion-server/internal/mtree"

internal/blockstore/commit_and_query.go

@@ -477,9 +477,7 @@ func (s *Store) GetHeaderByHash(blockHash []byte) (*types.BlockHeader, error) {
 // DoesTxIDExist returns true if any of the committed block has a transaction with
 // the given txID. Otherwise, it returns false
 func (s *Store) DoesTxIDExist(txID string) (bool, error) {
-	s.mu.RLock()
-	defer s.mu.RUnlock()
-
+	// It is safe to use without locks because the following call is protected internally.
 	return s.txValidationInfoDB.Has([]byte(txID), &opt.ReadOptions{})
 }
 

internal/queue/pending_txs.go

@@ -11,9 +11,8 @@ import (
 )
 
 type PendingTxs struct {
-	sync.RWMutex
-	txs map[string]*CompletionPromise
-
+	txs    map[string]*CompletionPromise
+	lock   sync.Mutex // We use a simple mutex because we only access the pending TXs to add or delete TXs
 	logger *logger.SugarLogger
 }
 
@@ -24,30 +23,45 @@ func NewPendingTxs(logger *logger.SugarLogger) *PendingTxs {
 	}
 }
 
-func (p *PendingTxs) Add(txID string, promise *CompletionPromise) {
-	p.Lock()
-	defer p.Unlock()
+// Add returns true if the txId was already taken
+func (p *PendingTxs) Add(txID string, promise *CompletionPromise) bool {
+	p.lock.Lock()
+	defer p.lock.Unlock()
+	_, loaded := p.txs[txID]
+	if !loaded {
+		p.txs[txID] = promise
+	}
+	return loaded
+}
+
+func (p *PendingTxs) DeleteWithNoAction(txID string) {
+	p.lock.Lock()
+	defer p.lock.Unlock()
+	delete(p.txs, txID)
+}
 
-	p.txs[txID] = promise
+func (p *PendingTxs) loadAndDelete(txID string) (*CompletionPromise, bool) {
+	p.lock.Lock()
+	defer p.lock.Unlock()
+	promise, loaded := p.txs[txID]
+	delete(p.txs, txID)
+	return promise, loaded
 }
 
 // DoneWithReceipt is called after the commit of a block.
 // The `txIDs` slice must be in the same order that transactions appear in the block.
 func (p *PendingTxs) DoneWithReceipt(txIDs []string, blockHeader *types.BlockHeader) {
 	p.logger.Debugf("Done with receipt, block number: %d; txIDs: %v", blockHeader.GetBaseHeader().GetNumber(), txIDs)
 
-	p.Lock()
-	defer p.Unlock()
-
 	for txIndex, txID := range txIDs {
-		p.txs[txID].done(
-			&types.TxReceipt{
-				Header:  blockHeader,
-				TxIndex: uint64(txIndex),
-			},
-		)
-
-		delete(p.txs, txID)
+		if promise, loaded := p.loadAndDelete(txID); loaded {
+			promise.done(
+				&types.TxReceipt{
+					Header:  blockHeader,
+					TxIndex: uint64(txIndex),
+				},
+			)
+		}
 	}
 }
 
@@ -57,27 +71,24 @@ func (p *PendingTxs) DoneWithReceipt(txIDs []string, blockHeader *types.BlockHea
 func (p *PendingTxs) ReleaseWithError(txIDs []string, err error) {
 	p.logger.Debugf("Release with error: %s; txIDs: %v", err, txIDs)
 
-	p.Lock()
-	defer p.Unlock()
-
 	for _, txID := range txIDs {
-		p.txs[txID].error(err)
-
-		delete(p.txs, txID)
+		if promise, loaded := p.loadAndDelete(txID); loaded {
+			promise.error(err)
+		}
 	}
 }
 
+// Has is used only for testing.
 func (p *PendingTxs) Has(txID string) bool {
-	p.RLock()
-	defer p.RUnlock()
-
+	p.lock.Lock()
+	defer p.lock.Unlock()
 	_, ok := p.txs[txID]
 	return ok
 }
 
+// Empty is used only for testing.
 func (p *PendingTxs) Empty() bool {
-	p.RLock()
-	defer p.RUnlock()
-
+	p.lock.Lock()
+	defer p.lock.Unlock()
 	return len(p.txs) == 0
 }

internal/queue/pending_txs_test.go

@@ -4,14 +4,14 @@
 package queue_test
 
 import (
-	ierrors "github.com/hyperledger-labs/orion-server/internal/errors"
 	"sync"
 	"testing"
 	"time"
 
+	"github.com/golang/protobuf/proto"
+	ierrors "github.com/hyperledger-labs/orion-server/internal/errors"
 	"github.com/hyperledger-labs/orion-server/internal/queue"
 	"github.com/hyperledger-labs/orion-server/pkg/types"
-	"github.com/golang/protobuf/proto"
 	"github.com/stretchr/testify/require"
 )
 
@@ -20,10 +20,10 @@ func TestPendingTxs_Async(t *testing.T) {
 
 	var p *queue.CompletionPromise
 	require.True(t, pendingTxs.Empty())
-	pendingTxs.Add("tx1", p)
+	require.False(t, pendingTxs.Add("tx1", p))
 	require.True(t, pendingTxs.Has("tx1"))
 	require.False(t, pendingTxs.Has("tx2"))
-	pendingTxs.Add("tx2", p)
+	require.False(t, pendingTxs.Add("tx2", p))
 	require.True(t, pendingTxs.Has("tx2"))
 	pendingTxs.DoneWithReceipt([]string{"tx1", "tx2"}, nil)
 	require.True(t, pendingTxs.Empty())
@@ -45,7 +45,7 @@ func TestPendingTxs_Sync(t *testing.T) {
 
 	t.Run("Wait before Done", func(t *testing.T) {
 		p := queue.NewCompletionPromise(time.Hour)
-		pendingTxs.Add("tx3", p)
+		require.False(t, pendingTxs.Add("tx3", p))
 
 		go func() {
 			time.Sleep(10 * time.Millisecond)
@@ -59,7 +59,7 @@ func TestPendingTxs_Sync(t *testing.T) {
 
 	t.Run("Done before Wait", func(t *testing.T) {
 		p := queue.NewCompletionPromise(time.Hour)
-		pendingTxs.Add("tx3", p)
+		require.False(t, pendingTxs.Add("tx3", p))
 		pendingTxs.DoneWithReceipt([]string{"tx3"}, blockHeader)
 		actualReceipt, err := p.Wait()
 		require.NoError(t, err)
@@ -68,7 +68,7 @@ func TestPendingTxs_Sync(t *testing.T) {
 
 	t.Run("Wait before Release with Error", func(t *testing.T) {
 		p := queue.NewCompletionPromise(time.Hour)
-		pendingTxs.Add("tx3", p)
+		require.False(t, pendingTxs.Add("tx3", p))
 
 		go func() {
 			time.Sleep(10 * time.Millisecond)
@@ -82,7 +82,7 @@ func TestPendingTxs_Sync(t *testing.T) {
 
 	t.Run("Release with Error before Wait", func(t *testing.T) {
 		p := queue.NewCompletionPromise(time.Hour)
-		pendingTxs.Add("tx3", p)
+		require.False(t, pendingTxs.Add("tx3", p))
 		pendingTxs.ReleaseWithError([]string{"tx3"}, &ierrors.NotLeaderError{LeaderID: 1, LeaderHostPort: "10.10.10.10:666"})
 		actualReceipt, err := p.Wait()
 		require.EqualError(t, err, "not a leader, leader is RaftID: 1, with HostPort: 10.10.10.10:666")
@@ -94,7 +94,7 @@ func TestPendingTxs_Timeout(t *testing.T) {
 	pendingTxs := queue.NewPendingTxs(testLogger(t, "debug"))
 
 	p := queue.NewCompletionPromise(1 * time.Millisecond)
-	pendingTxs.Add("tx3", p)
+	require.False(t, pendingTxs.Add("tx3", p))
 
 	var wg sync.WaitGroup
 	wg.Add(1)