cache signatures, and dont check them in a lock

build/params_shared.go

@@ -109,6 +109,10 @@ const BadBlockCacheSize = 1 << 15
 // 10 block reorg.
 const BlsSignatureCacheSize = 40000
 
+// Size of signature verification cache
+// 32k keeps the cache around 10MB in size, max
+const VerifSigCacheSize = 32000
+
 // ///////
 // Limits
 

chain/messagepool/messagepool.go

@@ -4,6 +4,7 @@ import (
 	"bytes"
 	"context"
 	"errors"
+	"fmt"
 	"math"
 	"sort"
 	"sync"
@@ -87,6 +88,8 @@ type MessagePool struct {
 	localMsgs datastore.Datastore
 
 	netName dtypes.NetworkName
+
+	sigValCache *lru.TwoQueueCache
 }
 
 type msgSet struct {
@@ -180,6 +183,8 @@ func (mpp *mpoolProvider) LoadTipSet(tsk types.TipSetKey) (*types.TipSet, error)
 
 func New(api Provider, ds dtypes.MetadataDS, netName dtypes.NetworkName) (*MessagePool, error) {
 	cache, _ := lru.New2Q(build.BlsSignatureCacheSize)
+	verifcache, _ := lru.New2Q(build.VerifSigCacheSize)
+
 	mp := &MessagePool{
 		closer:        make(chan struct{}),
 		repubTk:       time.NewTicker(build.BlockDelay * 10 * time.Second),
@@ -188,6 +193,7 @@ func New(api Provider, ds dtypes.MetadataDS, netName dtypes.NetworkName) (*Messa
 		minGasPrice:   types.NewInt(0),
 		maxTxPoolSize: 5000,
 		blsSigCache:   cache,
+		sigValCache:   verifcache,
 		changes:       lps.New(50),
 		localMsgs:     namespace.Wrap(ds, datastore.NewKey(localMsgsDs)),
 		api:           api,
@@ -313,12 +319,6 @@ func (mp *MessagePool) Push(m *types.SignedMessage) (cid.Cid, error) {
 }
 
 func (mp *MessagePool) Add(m *types.SignedMessage) error {
-	mp.curTsLk.Lock()
-	defer mp.curTsLk.Unlock()
-	return mp.addTs(m, mp.curTs)
-}
-
-func (mp *MessagePool) addTs(m *types.SignedMessage, curTs *types.TipSet) error {
 	// big messages are bad, anti DOS
 	if m.Size() > 32*1024 {
 		return xerrors.Errorf("mpool message too large (%dB): %w", m.Size(), ErrMessageTooBig)
@@ -332,11 +332,53 @@ func (mp *MessagePool) addTs(m *types.SignedMessage, curTs *types.TipSet) error
 		return ErrMessageValueTooHigh
 	}
 
-	if err := sigs.Verify(&m.Signature, m.Message.From, m.Message.Cid().Bytes()); err != nil {
+	if err := mp.VerifyMsgSig(m); err != nil {
 		log.Warnf("mpooladd signature verification failed: %s", err)
 		return err
 	}
 
+	mp.curTsLk.Lock()
+	defer mp.curTsLk.Unlock()
+	return mp.addTs(m, mp.curTs)
+}
+
+func sigCacheKey(m *types.SignedMessage) (string, error) {
+	switch m.Signature.Type {
+	case crypto.SigTypeBLS:
+		if len(m.Signature.Data) < 90 {
+			return "", fmt.Errorf("bls signature too short")
+		}
+
+		return string(m.Cid().Bytes()) + string(m.Signature.Data[64:]), nil
+	case crypto.SigTypeSecp256k1:
+		return string(m.Cid().Bytes()), nil
+	default:
+		return "", xerrors.Errorf("unrecognized signature type: %d", m.Signature.Type)
+	}
+}
+
+func (mp *MessagePool) VerifyMsgSig(m *types.SignedMessage) error {
+	sck, err := sigCacheKey(m)
+	if err != nil {
+		return err
+	}
+
+	_, ok := mp.sigValCache.Get(sck)
+	if ok {
+		// already validated, great
+		return nil
+	}
+
+	if err := sigs.Verify(&m.Signature, m.Message.From, m.Message.Cid().Bytes()); err != nil {
+		return err
+	}
+
+	mp.sigValCache.Add(sck, struct{}{})
+
+	return nil
+}
+
+func (mp *MessagePool) addTs(m *types.SignedMessage, curTs *types.TipSet) error {
 	snonce, err := mp.getStateNonce(m.Message.From, curTs)
 	if err != nil {
 		return xerrors.Errorf("failed to look up actor state nonce: %s: %w", err, ErrBroadcastAnyway)

chain/messagepool/messagepool_test.go

@@ -3,7 +3,9 @@ package messagepool
 import (
 	"context"
 	"fmt"
+	"math/rand"
 	"testing"
+	"time"
 
 	"github.com/filecoin-project/go-address"
 	"github.com/filecoin-project/lotus/chain/types"
@@ -233,3 +235,61 @@ func TestRevertMessages(t *testing.T) {
 	}
 
 }
+
+func TestMpoolStress(t *testing.T) {
+	tma := newTestMpoolApi()
+
+	w, err := wallet.NewWallet(wallet.NewMemKeyStore())
+	if err != nil {
+		t.Fatal(err)
+	}
+
+	ds := datastore.NewMapDatastore()
+
+	mp, err := New(tma, ds, "mptest")
+	if err != nil {
+		t.Fatal(err)
+	}
+
+	var messages []*types.SignedMessage
+
+	for i := 0; i < 100; i++ {
+		sender, err := w.GenerateKey(crypto.SigTypeBLS)
+		if err != nil {
+			t.Fatal(err)
+		}
+		target := mock.Address(1001)
+
+		for i := 0; i < 50; i++ {
+			messages = append(messages, mock.MkMessage(sender, target, uint64(i), w))
+		}
+	}
+
+	numAdds := 1000
+	averageAddTimings := make(chan time.Duration)
+	for i := 0; i < 20; i++ {
+		go func() {
+			var sum time.Duration
+			for j := 0; j < numAdds; j++ {
+				b := time.Now()
+				if err := mp.Add(messages[rand.Intn(len(messages))]); err != nil {
+					t.Error(err)
+				}
+				m := time.Since(b)
+				sum += m
+				fmt.Println(m)
+			}
+			averageAddTimings <- (sum / time.Duration(numAdds))
+		}()
+	}
+
+	var timings []time.Duration
+	var tsum time.Duration
+	for i := 0; i < 20; i++ {
+		t := <-averageAddTimings
+		timings = append(timings, t)
+		tsum += t
+	}
+
+	fmt.Println("average add time: ", tsum/20)
+}