optimistic sig share creation for managed keys

consensus/interface.go

@@ -189,7 +189,7 @@ type PeerBlacklistHandler interface {
 // SigningHandler defines the behaviour of a component that handles multi and single signatures used in consensus operations
 type SigningHandler interface {
 	Reset(pubKeys []string) error
-	CreateSignatureShareForPublicKey(message []byte, index uint16, epoch uint32, publicKeyBytes []byte) ([]byte, error)
+	CreateSignatureShareForPublicKey(ctx context.Context, message []byte, index uint16, epoch uint32, publicKeyBytes []byte) ([]byte, error)
 	CreateSignatureForPublicKey(message []byte, publicKeyBytes []byte) ([]byte, error)
 	VerifySingleSignature(publicKeyBytes []byte, message []byte, signature []byte) error
 	StoreSignatureShare(index uint16, sig []byte) error

consensus/spos/bls/v1/subroundSignature.go

@@ -73,7 +73,7 @@ func checkNewSubroundSignatureParams(
 }
 
 // doSignatureJob method does the job of the subround Signature
-func (sr *subroundSignature) doSignatureJob(_ context.Context) bool {
+func (sr *subroundSignature) doSignatureJob(ctx context.Context) bool {
 	if !sr.CanDoSubroundJob(sr.Current()) {
 		return false
 	}
@@ -93,6 +93,7 @@ func (sr *subroundSignature) doSignatureJob(_ context.Context) bool {
 		}
 
 		signatureShare, err := sr.SigningHandler().CreateSignatureShareForPublicKey(
+			ctx,
 			sr.GetData(),
 			uint16(selfIndex),
 			sr.GetHeader().GetEpoch(),
@@ -116,7 +117,7 @@ func (sr *subroundSignature) doSignatureJob(_ context.Context) bool {
 		}
 	}
 
-	return sr.doSignatureJobForManagedKeys()
+	return sr.doSignatureJobForManagedKeys(ctx)
 }
 
 func (sr *subroundSignature) createAndSendSignatureMessage(signatureShare []byte, pkBytes []byte) bool {
@@ -351,7 +352,7 @@ func (sr *subroundSignature) remainingTime() time.Duration {
 	return remainigTime
 }
 
-func (sr *subroundSignature) doSignatureJobForManagedKeys() bool {
+func (sr *subroundSignature) doSignatureJobForManagedKeys(ctx context.Context) bool {
 	isMultiKeyLeader := sr.IsMultiKeyLeaderInCurrentRound()
 
 	numMultiKeysSignaturesSent := 0
@@ -371,6 +372,7 @@ func (sr *subroundSignature) doSignatureJobForManagedKeys() bool {
 		}
 
 		signatureShare, err := sr.SigningHandler().CreateSignatureShareForPublicKey(
+			ctx,
 			sr.GetData(),
 			uint16(selfIndex),
 			sr.GetHeader().GetEpoch(),

consensus/spos/bls/v1/subroundSignature_test.go

@@ -1,6 +1,7 @@
 package v1_test
 
 import (
+	"context"
 	"testing"
 
 	"github.com/multiversx/mx-chain-core-go/core/check"
@@ -360,7 +361,7 @@ func TestSubroundSignature_DoSignatureJob(t *testing.T) {
 
 	err := errors.New("create signature share error")
 	signingHandler := &consensusMocks.SigningHandlerStub{
-		CreateSignatureShareForPublicKeyCalled: func(msg []byte, index uint16, epoch uint32, publicKeyBytes []byte) ([]byte, error) {
+		CreateSignatureShareForPublicKeyCalled: func(_ context.Context, msg []byte, index uint16, epoch uint32, publicKeyBytes []byte) ([]byte, error) {
 			return nil, err
 		},
 	}
@@ -370,7 +371,7 @@ func TestSubroundSignature_DoSignatureJob(t *testing.T) {
 	assert.False(t, r)
 
 	signingHandler = &consensusMocks.SigningHandlerStub{
-		CreateSignatureShareForPublicKeyCalled: func(msg []byte, index uint16, epoch uint32, publicKeyBytes []byte) ([]byte, error) {
+		CreateSignatureShareForPublicKeyCalled: func(_ context.Context, msg []byte, index uint16, epoch uint32, publicKeyBytes []byte) ([]byte, error) {
 			return []byte("SIG"), nil
 		},
 	}
@@ -441,7 +442,7 @@ func TestSubroundSignature_DoSignatureJobWithMultikey(t *testing.T) {
 
 	err := errors.New("create signature share error")
 	signingHandler := &consensusMocks.SigningHandlerStub{
-		CreateSignatureShareForPublicKeyCalled: func(msg []byte, index uint16, epoch uint32, publicKeyBytes []byte) ([]byte, error) {
+		CreateSignatureShareForPublicKeyCalled: func(_ context.Context, msg []byte, index uint16, epoch uint32, publicKeyBytes []byte) ([]byte, error) {
 			return nil, err
 		},
 	}
@@ -451,7 +452,7 @@ func TestSubroundSignature_DoSignatureJobWithMultikey(t *testing.T) {
 	assert.False(t, r)
 
 	signingHandler = &consensusMocks.SigningHandlerStub{
-		CreateSignatureShareForPublicKeyCalled: func(msg []byte, index uint16, epoch uint32, publicKeyBytes []byte) ([]byte, error) {
+		CreateSignatureShareForPublicKeyCalled: func(_ context.Context, msg []byte, index uint16, epoch uint32, publicKeyBytes []byte) ([]byte, error) {
 			return []byte("SIG"), nil
 		},
 	}

consensus/spos/bls/v2/benchmark_send_proof_test.go

@@ -1,6 +1,7 @@
 package v2_test
 
 import (
+	"context"
 	"sort"
 	"testing"
 
@@ -171,7 +172,7 @@ func benchmarkSendProof(b *testing.B, numberOfKeys int) {
 
 	// Create signature shares for all validators (simulating that all have signed)
 	for i := 0; i < len(keys); i++ {
-		_, err := signingHandler.CreateSignatureShareForPublicKey(dataToBeSigned, uint16(i), 0, []byte(keys[i]))
+		_, err := signingHandler.CreateSignatureShareForPublicKey(context.TODO(), dataToBeSigned, uint16(i), 0, []byte(keys[i]))
 		require.Nil(b, err)
 		err = srEndRound.SetJobDone(keys[i], bls.SrSignature, true)
 		require.Nil(b, err)
@@ -183,7 +184,7 @@ func benchmarkSendProof(b *testing.B, numberOfKeys int) {
 	for i := 0; i < b.N; i++ {
 		// Reset signature state for each iteration
 		for j := 0; j < len(keys); j++ {
-			_, _ = signingHandler.CreateSignatureShareForPublicKey(dataToBeSigned, uint16(j), 0, []byte(keys[j]))
+			_, _ = signingHandler.CreateSignatureShareForPublicKey(context.TODO(), dataToBeSigned, uint16(j), 0, []byte(keys[j]))
 		}
 
 		b.StartTimer()

consensus/spos/bls/v2/benchmark_verify_signatures_test.go

@@ -114,7 +114,7 @@ func BenchmarkSubroundEndRound_VerifyNodesOnAggSigFailTime(b *testing.B) {
 
 	sr := initSubroundEndRoundWithContainerAndConsensusState(container, &statusHandler.AppStatusHandlerStub{}, consensusState)
 	for i := 0; i < len(sr.ConsensusGroup()); i++ {
-		_, err := sr.SigningHandler().CreateSignatureShareForPublicKey(dataToBeSigned, uint16(i), sr.EnableEpochsHandler().GetCurrentEpoch(), []byte(keys[i]))
+		_, err := sr.SigningHandler().CreateSignatureShareForPublicKey(context.TODO(), dataToBeSigned, uint16(i), sr.EnableEpochsHandler().GetCurrentEpoch(), []byte(keys[i]))
 		require.Nil(b, err)
 		_ = sr.SetJobDone(keys[i], bls.SrSignature, true)
 	}

consensus/spos/bls/v2/blsSubroundsFactory.go

@@ -218,6 +218,7 @@ func (fct *factory) generateBlockSubround() error {
 		processingThresholdPercent,
 		fct.worker,
 		syncController,
+		fct.signatureThrottler,
 	)
 	if err != nil {
 		return err

consensus/spos/bls/v2/common.go

@@ -0,0 +1,30 @@
+package v2
+
+import (
+	"context"
+	"fmt"
+	"time"
+
+	"github.com/multiversx/mx-chain-core-go/core"
+	"github.com/multiversx/mx-chain-go/consensus/spos"
+)
+
+func checkGoRoutinesThrottler(
+	ctx context.Context,
+	signatureThrottler core.Throttler,
+) error {
+	for {
+		if signatureThrottler.CanProcess() {
+			break
+		}
+
+		select {
+		case <-time.After(timeSpentBetweenChecks):
+			continue
+		case <-ctx.Done():
+			return fmt.Errorf("%w while checking the throttler", spos.ErrTimeIsOut)
+		}
+	}
+
+	return nil
+}

consensus/spos/bls/v2/export_test.go

@@ -203,7 +203,7 @@ func (sr *subroundBlock) SendBlockBody(body data.BodyHandler, marshalizedBody []
 
 // SendBlockHeader method sends the proposed block header in the subround Block
 func (sr *subroundBlock) SendBlockHeader(header data.HeaderHandler, marshalizedHeader []byte) bool {
-	return sr.sendBlockHeader(header, marshalizedHeader)
+	return sr.sendBlockHeader(context.TODO(), header, marshalizedHeader)
 }
 
 // ComputeSubroundProcessingMetric computes processing metric related to the subround Block
@@ -395,3 +395,8 @@ func (sr *subroundEndRound) UpdateNonceDeltaMetrics() {
 func (sr *subroundBlock) PrepareBlockForExecution(header data.HeaderHandler, body data.BodyHandler) error {
 	return sr.prepareBlockForExecution(header, body)
 }
+
+// TriggerCreateSignaturesForManagedKeys -
+func (sr *subroundBlock) TriggerCreateSignaturesForManagedKeys(ctx context.Context, headerHash []byte, headerHandler data.HeaderHandler) {
+	sr.triggerCreateSignaturesForManagedKeys(ctx, headerHash, headerHandler)
+}

consensus/spos/bls/v2/subroundBlock.go

@@ -18,6 +18,11 @@ import (
 	"github.com/multiversx/mx-chain-go/process/asyncExecution/cache"
 )
 
+type consensusPubKey struct {
+	idx     int
+	pkBytes []byte
+}
+
 // maxAllowedSizeInBytes defines how many bytes are allowed as payload in a message
 const maxAllowedSizeInBytes = uint32(core.MegabyteSize * 95 / 100)
 
@@ -29,6 +34,7 @@ type subroundBlock struct {
 	worker                        spos.WorkerHandler
 	mutBlockProcessing            sync.Mutex
 	syncController                spos.NtpSyncControllerHandler
+	signatureThrottler            core.Throttler
 }
 
 // NewSubroundBlock creates a subroundBlock object
@@ -37,6 +43,7 @@ func NewSubroundBlock(
 	processingThresholdPercentage int,
 	worker spos.WorkerHandler,
 	syncController spos.NtpSyncControllerHandler,
+	signatureThrottler core.Throttler,
 ) (*subroundBlock, error) {
 	err := checkNewSubroundBlockParams(baseSubround)
 	if err != nil {
@@ -49,12 +56,16 @@ func NewSubroundBlock(
 	if check.IfNil(syncController) {
 		return nil, ErrNilRoundSyncController
 	}
+	if check.IfNil(signatureThrottler) {
+		return nil, spos.ErrNilThrottler
+	}
 
 	srBlock := subroundBlock{
 		Subround:                      baseSubround,
 		processingThresholdPercentage: processingThresholdPercentage,
 		worker:                        worker,
 		syncController:                syncController,
+		signatureThrottler:            signatureThrottler,
 	}
 
 	srBlock.Job = srBlock.doBlockJob
@@ -142,7 +153,7 @@ func (sr *subroundBlock) doBlockJob(ctx context.Context) bool {
 		return false
 	}
 
-	sentWithSuccess := sr.sendBlock(header, body, leader)
+	sentWithSuccess := sr.sendBlock(ctx, header, body, leader)
 	if !sentWithSuccess {
 		return false
 	}
@@ -220,7 +231,12 @@ func printLogMessage(ctx context.Context, baseMessage string, err error) {
 	log.Debug(baseMessage, "error", err.Error())
 }
 
-func (sr *subroundBlock) sendBlock(header data.HeaderHandler, body data.BodyHandler, leader string) bool {
+func (sr *subroundBlock) sendBlock(
+	ctx context.Context,
+	header data.HeaderHandler,
+	body data.BodyHandler,
+	leader string,
+) bool {
 	marshalledBody, err := sr.Marshalizer().Marshal(body)
 	if err != nil {
 		log.Debug("sendBlock.Marshal: body", "error", err.Error())
@@ -236,7 +252,7 @@ func (sr *subroundBlock) sendBlock(header data.HeaderHandler, body data.BodyHand
 	sr.logBlockSize(marshalledBody, marshalledHeader)
 	headerHash := sr.Hasher().Compute(string(marshalledHeader))
 
-	if !sr.sendBlockBody(body, marshalledBody) || !sr.sendBlockHeader(header, headerHash) {
+	if !sr.sendBlockBody(body, marshalledBody) || !sr.sendBlockHeader(ctx, header, headerHash) {
 		return false
 	}
 
@@ -313,6 +329,7 @@ func (sr *subroundBlock) sendBlockBody(
 
 // sendBlockHeader method sends the proposed block header in the subround Block
 func (sr *subroundBlock) sendBlockHeader(
+	ctx context.Context,
 	headerHandler data.HeaderHandler,
 	headerHash []byte,
 ) bool {
@@ -336,6 +353,8 @@ func (sr *subroundBlock) sendBlockHeader(
 	sr.SetData(headerHash)
 	sr.SetHeader(headerHandler)
 
+	sr.triggerCreateSignaturesForManagedKeys(ctx, headerHash, headerHandler)
+
 	// log the header output for debugging purposes
 	headerOutput, err := common.PrettifyStruct(headerHandler)
 	if err == nil {
@@ -345,6 +364,100 @@ func (sr *subroundBlock) sendBlockHeader(
 	return true
 }
 
+func (sr *subroundBlock) triggerCreateSignaturesForManagedKeys(
+	ctx context.Context,
+	headerHash []byte,
+	headerHandler data.HeaderHandler,
+) {
+	if check.IfNil(headerHandler) {
+		log.Debug("triggerCreateSignaturesForManagedKeys: triggered with nil header")
+		return
+	}
+
+	if sr.RoundHandler().Index() != int64(headerHandler.GetRound()) {
+		log.Debug("triggerCreateSignaturesForManagedKeys: not for current round",
+			"currentRound", sr.RoundHandler().Index(),
+			"headerRound", headerHandler.GetRound(),
+		)
+		return
+	}
+
+	currentEpoch := headerHandler.GetEpoch()
+
+	sigSubroundEndTime := time.Duration(float64(sr.RoundHandler().TimeDuration()) * srSignatureEndTime)
+	timeLeft := sr.RoundHandler().RemainingTime(sr.RoundHandler().TimeStamp(), sigSubroundEndTime)
+	sigCtx, cancel := context.WithTimeout(ctx, timeLeft)
+	sr.SetSignaturesCtxCancelFunc(cancel)
+
+	keys := sr.getManagedKeysByIndex()
+	if len(keys) == 0 {
+		return
+	}
+
+	wg := sr.SignaturesWaitGroup()
+	wg.Add(len(keys))
+
+	go func() {
+		triggered := 0
+		for _, pk := range keys {
+			err := checkGoRoutinesThrottler(sigCtx, sr.signatureThrottler)
+			if err != nil {
+				log.Debug("triggerCreateSignaturesForManagedKeys.checkGoRoutinesThrottler", "err", err)
+				cancel()
+
+				for i := triggered; i < len(keys); i++ {
+					wg.Done()
+				}
+
+				return
+			}
+			sr.signatureThrottler.StartProcessing()
+
+			triggered++
+
+			go func(sigCtx context.Context, idx int, pkBytes []byte) {
+				defer sr.signatureThrottler.EndProcessing()
+				defer wg.Done()
+
+				select {
+				case <-sigCtx.Done():
+					log.Info("triggerCreateSignaturesForManagedKeys: context done", "timeLeft", timeLeft)
+					return
+				default:
+				}
+
+				_, err := sr.SigningHandler().CreateSignatureShareForPublicKey(
+					sigCtx,
+					headerHash,
+					uint16(idx),
+					currentEpoch,
+					pkBytes,
+				)
+				if err != nil {
+					log.Info("triggerCreateSignaturesForManagedKeys.CreateSignatureShareForPublicKey", "error", err.Error())
+					return
+				}
+			}(sigCtx, pk.idx, pk.pkBytes)
+		}
+	}()
+
+	log.Debug("step 1: multi keys signatures creation has been triggered", "num", len(keys))
+}
+
+func (sr *subroundBlock) getManagedKeysByIndex() []consensusPubKey {
+	keys := make([]consensusPubKey, 0)
+	for idx, pk := range sr.ConsensusGroup() {
+		pkBytes := []byte(pk)
+		if !sr.IsKeyManagedBySelf(pkBytes) {
+			continue
+		}
+
+		keys = append(keys, consensusPubKey{idx: idx, pkBytes: pkBytes})
+	}
+
+	return keys
+}
+
 func (sr *subroundBlock) sendDirectSentTransactions(
 	header data.HeaderHandler,
 	body data.BodyHandler,
@@ -643,6 +756,8 @@ func (sr *subroundBlock) receivedBlockHeader(headerHandler data.HeaderHandler) {
 
 	sr.AddReceivedHeader(headerHandler)
 
+	sr.triggerCreateSignaturesForManagedKeys(context.Background(), headerHash, headerHandler)
+
 	ctx, cancel := context.WithTimeout(context.Background(), sr.RoundHandler().TimeDuration())
 	defer cancel()