Repository: iotaledger/iota-core
Pull request #882 (2024-03-26 09:20:01)

Remove execution workerpool 
Execution workerpool with 1 worker was a bottleneck. It works much faster with VM execution by the same goroutine that is processing the block. 


Patch:
diff --git a/pkg/protocol/engine/ledger/ledger/ledger.go b/pkg/protocol/engine/ledger/ledger/ledger.go
index 66bde902..d73e88c5 100644
--- a/pkg/protocol/engine/ledger/ledger/ledger.go
+++ b/pkg/protocol/engine/ledger/ledger/ledger.go
@@ -69,7 +69,7 @@ func NewProvider() module.Provider[*engine.Engine, ledger.Ledger] {
 			})
 			e.Events.SpendDAG.LinkTo(l.spendDAG.Events())
 
-			l.memPool = mempoolv1.New(NewVM(l), l.resolveState, e.Storage.Mutations, e.Workers.CreateGroup("MemPool"), l.spendDAG, l.apiProvider, l.errorHandler)
+			l.memPool = mempoolv1.New(NewVM(l), l.resolveState, e.Storage.Mutations, l.spendDAG, l.apiProvider, l.errorHandler)
 
 			l.manaManager = mana.NewManager(l.apiProvider, l.resolveAccountOutput, l.accountsLedger.Account)
 			latestCommittedSlot := e.Storage.Settings().LatestCommitment().Slot()
diff --git a/pkg/protocol/engine/mempool/v1/mempool.go b/pkg/protocol/engine/mempool/v1/mempool.go
index 930d2366..dc0a4458 100644
--- a/pkg/protocol/engine/mempool/v1/mempool.go
+++ b/pkg/protocol/engine/mempool/v1/mempool.go
@@ -12,7 +12,6 @@ import (
 	"github.com/iotaledger/hive.go/runtime/event"
 	"github.com/iotaledger/hive.go/runtime/options"
 	"github.com/iotaledger/hive.go/runtime/syncutils"
-	"github.com/iotaledger/hive.go/runtime/workerpool"
 	"github.com/iotaledger/iota-core/pkg/core/promise"
 	"github.com/iotaledger/iota-core/pkg/core/vote"
 	"github.com/iotaledger/iota-core/pkg/protocol/engine/mempool"
@@ -60,9 +59,6 @@ type MemPool[VoteRank spenddag.VoteRankType[VoteRank]] struct {
 
 	errorHandler func(error)
 
-	// executionWorkers is the worker pool that is used to execute the state transitions of transactions.
-	executionWorkers *workerpool.WorkerPool
-
 	// lastCommittedSlot is the last slot that was committed in the MemPool.
 	lastCommittedSlot iotago.SlotIndex
 
@@ -81,7 +77,6 @@ func New[VoteRank spenddag.VoteRankType[VoteRank]](
 	vm mempool.VM,
 	stateResolver mempool.StateResolver,
 	mutationsFunc func(iotago.SlotIndex) (kvstore.KVStore, error),
-	workers *workerpool.Group,
 	spendDAG spenddag.SpendDAG[iotago.TransactionID, mempool.StateID, VoteRank],
 	apiProvider iotago.APIProvider,
 	errorHandler func(error),
@@ -96,7 +91,6 @@ func New[VoteRank spenddag.VoteRankType[VoteRank]](
 		cachedSignedTransactions:   shrinkingmap.New[iotago.SignedTransactionID, *SignedTransactionMetadata](),
 		cachedStateRequests:        shrinkingmap.New[mempool.StateID, *promise.Promise[*StateMetadata]](),
 		stateDiffs:                 shrinkingmap.New[iotago.SlotIndex, *StateDiff](),
-		executionWorkers:           workers.CreatePool("executionWorkers", workerpool.WithWorkerCount(1)),
 		delayedTransactionEviction: shrinkingmap.New[iotago.SlotIndex, ds.Set[iotago.TransactionID]](),
 		delayedOutputStateEviction: shrinkingmap.New[iotago.SlotIndex, *shrinkingmap.ShrinkingMap[iotago.Identifier, *StateMetadata]](),
 		spendDAG:                   spendDAG,
@@ -408,15 +402,13 @@ func (m *MemPool[VoteRank]) solidifyInputs(transaction *TransactionMetadata) {
 }
 
 func (m *MemPool[VoteRank]) executeTransaction(executionContext context.Context, transaction *TransactionMetadata) {
-	m.executionWorkers.Submit(func() {
-		if outputStates, err := m.vm.Execute(executionContext, transaction.Transaction()); err != nil {
-			transaction.setInvalid(err)
-		} else {
-			transaction.setExecuted(outputStates)
+	if outputStates, err := m.vm.Execute(executionContext, transaction.Transaction()); err != nil {
+		transaction.setInvalid(err)
+	} else {
+		transaction.setExecuted(outputStates)
 
-			m.bookTransaction(transaction)
-		}
-	})
+		m.bookTransaction(transaction)
+	}
 }
 
 func (m *MemPool[VoteRank]) bookTransaction(transaction *TransactionMetadata) {
diff --git a/pkg/requesthandler/blockissuance.go b/pkg/requesthandler/blockissuance.go
index a703b0e7..80b261c5 100644
--- a/pkg/requesthandler/blockissuance.go
+++ b/pkg/requesthandler/blockissuance.go
@@ -14,11 +14,6 @@ import (
 	iotago "github.com/iotaledger/iota.go/v4"
 )
 
-var (
-	// errBlockRetained is not really an error but it is used to signal that the block was retained.
-	errBlockRetained = ierrors.New("block retained")
-)
-
 // TODO: make sure an honest validator does not issue blocks within the same slot ratification period in two conflicting chains.
 //  - this can be achieved by remembering the last issued block together with the engine name/chain.
 //  - if the engine name/chain is the same we can always issue a block.
@@ -31,27 +26,29 @@ func (r *RequestHandler) SubmitBlockWithoutAwaitingBooking(block *model.Block) e
 
 // submitBlockAndAwaitRetainer submits a block to be processed and waits for the block gets retained.
 func (r *RequestHandler) submitBlockAndAwaitRetainer(ctx context.Context, block *model.Block) error {
-	// Calculate the blockID so that we don't capture the block pointer in the event handlers.
-	blockID := block.ID()
+	exit := make(chan struct{})
+	defer close(exit)
+
+	blockFiltered := make(chan error, 1)
+	defer close(blockFiltered)
 
 	// Make sure we don't wait forever here. If the block is not dispatched to the main engine,
 	// it will never trigger one of the below events.
 	processingCtx, processingCtxCancel := context.WithTimeout(ctx, 10*time.Second)
 	defer processingCtxCancel()
 
-	blockCtx, blockCtxCancel := context.WithCancelCause(context.Background())
-	defer blockCtxCancel(context.Canceled) // make sure the context is canceled when we return to prevent memory leaks
+	// Calculate the blockID so that we don't capture the block pointer in the event handlers.
+	blockID := block.ID()
 
 	// Hook to TransactionRetained event if the block contains a transaction.
-	var txCtx context.Context
+	var txRetained chan struct{}
 	var txRetainedUnhook func()
 
 	if signedTx, isTx := block.SignedTransaction(); isTx {
 		txID := signedTx.Transaction.MustID()
 
-		var txCtxCancel context.CancelFunc
-		txCtx, txCtxCancel = context.WithCancel(context.Background())
-		defer txCtxCancel() // make sure the context is canceled when we return to prevent memory leaks
+		txRetained = make(chan struct{}, 2) // buffered to 2 to avoid blocking in case of a race condition
+		defer close(txRetained)
 
 		// we hook to the TransactionRetained event first, because there could be a race condition when the transaction gets retained
 		// the moment we check if the transaction is already retained.
@@ -61,14 +58,14 @@ func (r *RequestHandler) submitBlockAndAwaitRetainer(ctx context.Context, block
 			}
 
 			// signal that the transaction is retained
-			txCtxCancel()
+			txRetained <- struct{}{}
 		}, event.WithWorkerPool(r.workerPool)).Unhook
 
 		// if the transaction is already retained, we don't need to wait for the event because
 		// the onTransactionAttached event is only triggered if it's a new transaction.
 		if _, err := r.protocol.Engines.Main.Get().TxRetainer.TransactionMetadata(txID); err == nil {
 			// signal that the transaction is retained
-			txCtxCancel()
+			txRetained <- struct{}{}
 		}
 	}
 
@@ -76,27 +73,30 @@ func (r *RequestHandler) submitBlockAndAwaitRetainer(ctx context.Context, block
 		if blockID != eventBlock.ID() {
 			return
 		}
-
-		// signal that block was retained
-		blockCtxCancel(errBlockRetained)
+		select {
+		case blockFiltered <- nil:
+		case <-exit:
+		}
 	}, event.WithWorkerPool(r.workerPool)).Unhook
 
 	blockPreFilteredUnhook := r.protocol.Events.Engine.PreSolidFilter.BlockPreFiltered.Hook(func(event *presolidfilter.BlockPreFilteredEvent) {
 		if blockID != event.Block.ID() {
 			return
 		}
-
-		// signal that block was filtered
-		blockCtxCancel(event.Reason)
+		select {
+		case blockFiltered <- event.Reason:
+		case <-exit:
+		}
 	}, event.WithWorkerPool(r.workerPool)).Unhook
 
 	blockPostFilteredUnhook := r.protocol.Events.Engine.PostSolidFilter.BlockFiltered.Hook(func(event *postsolidfilter.BlockFilteredEvent) {
 		if blockID != event.Block.ID() {
 			return
 		}
-
-		// signal that block was filtered
-		blockCtxCancel(event.Reason)
+		select {
+		case blockFiltered <- event.Reason:
+		case <-exit:
+		}
 	}, event.WithWorkerPool(r.workerPool)).Unhook
 
 	defer lo.BatchReverse(txRetainedUnhook, blockRetainedUnhook, blockPreFilteredUnhook, blockPostFilteredUnhook)()
@@ -113,12 +113,12 @@ func (r *RequestHandler) submitBlockAndAwaitRetainer(ctx context.Context, block
 
 		return ierrors.Errorf("context canceled whilst waiting for event on block %s", blockID)
 
-	case <-blockCtx.Done():
-		if err := context.Cause(blockCtx); !ierrors.Is(err, errBlockRetained) {
+	case err := <-blockFiltered:
+		if err != nil {
 			return ierrors.Wrapf(err, "block filtered %s", blockID)
 		}
 
-		if txCtx != nil {
+		if txRetained != nil {
 			select {
 			case <-processingCtx.Done():
 				if ierrors.Is(processingCtx.Err(), context.DeadlineExceeded) {
@@ -127,9 +127,8 @@ func (r *RequestHandler) submitBlockAndAwaitRetainer(ctx context.Context, block
 
 				return ierrors.Errorf("context canceled whilst waiting for transaction retained event on block %s", blockID)
 
-			case <-txCtx.Done():
-				// we need to wait for the transaction to be retained before we can return.
-				// the txCtx can only be canceled manually, so we can be sure that the transaction was retained.
+			case <-txRetained:
+				// we need to wait for the transaction to be retained before we can return
 			}
 		}
 


Test patch:
diff --git a/.github/workflows/docker-network-tests-nightly.yml b/.github/workflows/docker-network-tests-nightly.yml
index 5b70e30e..2a2176d4 100644
--- a/.github/workflows/docker-network-tests-nightly.yml
+++ b/.github/workflows/docker-network-tests-nightly.yml
@@ -31,10 +31,3 @@ jobs:
       run: |
         cd tools/docker-network/tests &&
         ./run_tests.sh ${{ github.event.inputs.testCases }}
-
-    - name: Upload logs as artifacts
-      uses: actions/upload-artifact@v4
-      if: success() || failure()
-      with:
-        name: logs
-        path: tools/docker-network/tests/logs
diff --git a/pkg/protocol/engine/mempool/tests/testframework.go b/pkg/protocol/engine/mempool/tests/testframework.go
index 4422506f..d89e05ce 100644
--- a/pkg/protocol/engine/mempool/tests/testframework.go
+++ b/pkg/protocol/engine/mempool/tests/testframework.go
@@ -9,7 +9,6 @@ import (
 
 	"github.com/iotaledger/hive.go/lo"
 	"github.com/iotaledger/hive.go/runtime/debug"
-	"github.com/iotaledger/hive.go/runtime/workerpool"
 	"github.com/iotaledger/iota-core/pkg/core/vote"
 	ledgertests "github.com/iotaledger/iota-core/pkg/protocol/engine/ledger/tests"
 	"github.com/iotaledger/iota-core/pkg/protocol/engine/mempool"
@@ -28,12 +27,11 @@ type TestFramework struct {
 	blockIDsByAlias          map[string]iotago.BlockID
 
 	ledgerState *ledgertests.MockStateResolver
-	workers     *workerpool.Group
 
 	test *testing.T
 }
 
-func NewTestFramework(t *testing.T, instance mempool.MemPool[vote.MockedRank], spendDAG spenddag.SpendDAG[iotago.TransactionID, mempool.StateID, vote.MockedRank], ledgerState *ledgertests.MockStateResolver, workers *workerpool.Group) *TestFramework {
+func NewTestFramework(t *testing.T, instance mempool.MemPool[vote.MockedRank], spendDAG spenddag.SpendDAG[iotago.TransactionID, mempool.StateID, vote.MockedRank], ledgerState *ledgertests.MockStateResolver) *TestFramework {
 	t.Helper()
 
 	tf := &TestFramework{
@@ -46,7 +44,6 @@ func NewTestFramework(t *testing.T, instance mempool.MemPool[vote.MockedRank], s
 		blockIDsByAlias:          make(map[string]iotago.BlockID),
 
 		ledgerState: ledgerState,
-		workers:     workers,
 		test:        t,
 	}
 
@@ -379,12 +376,7 @@ func (t *TestFramework) AssertStateDiff(slot iotago.SlotIndex, spentOutputAliase
 	}
 }
 
-func (t *TestFramework) WaitChildren() {
-	t.workers.WaitChildren()
-}
-
 func (t *TestFramework) Cleanup() {
-	t.workers.WaitChildren()
 	t.ledgerState.Cleanup()
 
 	iotago.UnregisterIdentifierAliases()
diff --git a/pkg/protocol/engine/mempool/v1/mempool_test.go b/pkg/protocol/engine/mempool/v1/mempool_test.go
index 8f15b792..127b823c 100644
--- a/pkg/protocol/engine/mempool/v1/mempool_test.go
+++ b/pkg/protocol/engine/mempool/v1/mempool_test.go
@@ -12,7 +12,6 @@ import (
 	"github.com/iotaledger/hive.go/kvstore"
 	"github.com/iotaledger/hive.go/kvstore/mapdb"
 	"github.com/iotaledger/hive.go/runtime/memanalyzer"
-	"github.com/iotaledger/hive.go/runtime/workerpool"
 	"github.com/iotaledger/iota-core/pkg/core/account"
 	"github.com/iotaledger/iota-core/pkg/core/promise"
 	"github.com/iotaledger/iota-core/pkg/core/vote"
@@ -29,8 +28,6 @@ func TestMemPoolV1_InterfaceWithForkingEverything(t *testing.T) {
 }
 
 func TestMempoolV1_ResourceCleanup(t *testing.T) {
-	workers := workerpool.NewGroup(t.Name())
-
 	mutationsFunc := func(index iotago.SlotIndex) (kvstore.KVStore, error) {
 		return mapdb.NewMapDB(), nil
 	}
@@ -39,9 +36,9 @@ func TestMempoolV1_ResourceCleanup(t *testing.T) {
 	spendDAG := spenddagv1.New[iotago.TransactionID, mempool.StateID, vote.MockedRank](func() int { return 0 })
 	memPoolInstance := New[vote.MockedRank](new(mempooltests.VM), func(reference mempool.StateReference) *promise.Promise[mempool.State] {
 		return ledgerState.ResolveOutputState(reference)
-	}, mutationsFunc, workers, spendDAG, iotago.SingleVersionProvider(tpkg.ZeroCostTestAPI), func(error) {})
+	}, mutationsFunc, spendDAG, iotago.SingleVersionProvider(tpkg.ZeroCostTestAPI), func(error) {})
 
-	tf := mempooltests.NewTestFramework(t, memPoolInstance, spendDAG, ledgerState, workers)
+	tf := mempooltests.NewTestFramework(t, memPoolInstance, spendDAG, ledgerState)
 
 	issueTransactions := func(startIndex, transactionCount int, prevStateAlias string) (int, string) {
 		index := startIndex
@@ -71,7 +68,6 @@ func TestMempoolV1_ResourceCleanup(t *testing.T) {
 	fmt.Println(memanalyzer.MemoryReport(tf))
 
 	txIndex, prevStateAlias := issueTransactions(1, 10, "genesis")
-	tf.WaitChildren()
 
 	require.Equal(t, 10, memPoolInstance.cachedTransactions.Size())
 	require.Equal(t, 10, memPoolInstance.delayedTransactionEviction.Size())
@@ -81,7 +77,6 @@ func TestMempoolV1_ResourceCleanup(t *testing.T) {
 	require.Equal(t, 11, memPoolInstance.cachedStateRequests.Size())
 
 	txIndex, prevStateAlias = issueTransactions(txIndex, 10, prevStateAlias)
-	tf.WaitChildren()
 
 	require.Equal(t, 20, memPoolInstance.cachedTransactions.Size())
 	require.Equal(t, 20, memPoolInstance.delayedTransactionEviction.Size())
@@ -90,7 +85,6 @@ func TestMempoolV1_ResourceCleanup(t *testing.T) {
 	require.Equal(t, 21, memPoolInstance.cachedStateRequests.Size())
 
 	txIndex, prevStateAlias = issueTransactions(txIndex, 10, prevStateAlias)
-	tf.WaitChildren()
 
 	// Cached transactions stop increasing after 20, because eviction is delayed by MCA.
 	require.Equal(t, 20, memPoolInstance.cachedTransactions.Size())
@@ -100,7 +94,6 @@ func TestMempoolV1_ResourceCleanup(t *testing.T) {
 	require.Equal(t, 21, memPoolInstance.cachedStateRequests.Size())
 
 	txIndex, _ = issueTransactions(txIndex, 10, prevStateAlias)
-	tf.WaitChildren()
 
 	require.Equal(t, 20, memPoolInstance.cachedTransactions.Size())
 	require.Equal(t, 20, memPoolInstance.delayedTransactionEviction.Size())
@@ -136,8 +129,6 @@ func TestMempoolV1_ResourceCleanup(t *testing.T) {
 }
 
 func newTestFramework(t *testing.T) *mempooltests.TestFramework {
-	workers := workerpool.NewGroup(t.Name())
-
 	ledgerState := ledgertests.New(ledgertests.NewMockedState(iotago.EmptyTransactionID, 0))
 	spendDAG := spenddagv1.New[iotago.TransactionID, mempool.StateID, vote.MockedRank](account.NewAccounts().SeatedAccounts().SeatCount)
 
@@ -147,5 +138,5 @@ func newTestFramework(t *testing.T) *mempooltests.TestFramework {
 
 	return mempooltests.NewTestFramework(t, New[vote.MockedRank](new(mempooltests.VM), func(reference mempool.StateReference) *promise.Promise[mempool.State] {
 		return ledgerState.ResolveOutputState(reference)
-	}, mutationsFunc, workers, spendDAG, iotago.SingleVersionProvider(tpkg.ZeroCostTestAPI), func(error) {}), spendDAG, ledgerState, workers)
+	}, mutationsFunc, spendDAG, iotago.SingleVersionProvider(tpkg.ZeroCostTestAPI), func(error) {}), spendDAG, ledgerState)
 }