implement dumb pruning

chain/messagepool/gasguess/guessgas.go

@@ -1,4 +1,4 @@
-package miner
+package gasguess
 
 import (
 	"context"
@@ -13,15 +13,17 @@ import (
 	"github.com/filecoin-project/specs-actors/actors/builtin"
 )
 
+type ActorLookup func(context.Context, address.Address, types.TipSetKey) (*types.Actor, error)
+
 const failedGasGuessRatio = 0.5
 const failedGasGuessMax = 25_000_000
 
-type costKey struct {
+type CostKey struct {
-	code cid.Cid
+	Code cid.Cid
-	m    abi.MethodNum
+	M    abi.MethodNum
 }
 
-var costs = map[costKey]int64{
+var Costs = map[CostKey]int64{
 	{builtin.InitActorCodeID, 2}:          8916753,
 	{builtin.StorageMarketActorCodeID, 2}: 6955002,
 	{builtin.StorageMarketActorCodeID, 4}: 245436108,
@@ -63,7 +65,7 @@ func GuessGasUsed(ctx context.Context, tsk types.TipSetKey, msg *types.SignedMes
 		return failedGuess(msg), xerrors.Errorf("could not lookup actor: %w", err)
 	}
 
-	guess, ok := costs[costKey{to.Code, msg.Message.Method}]
+	guess, ok := Costs[CostKey{to.Code, msg.Message.Method}]
 	if !ok {
 		return failedGuess(msg), xerrors.Errorf("unknown code-method combo")
 	}

chain/messagepool/messagepool.go

@@ -242,6 +242,12 @@ func (mp *MessagePool) Close() error {
 	return nil
 }
 
+func (mp *MessagePool) Prune() {
+	mp.pruneTrigger <- struct{}{}
+	mp.pruneTrigger <- struct{}{}
+	mp.pruneTrigger <- struct{}{}
+}
+
 func (mp *MessagePool) runLoop() {
 	for {
 		select {
@@ -627,6 +633,10 @@ func (mp *MessagePool) Remove(from address.Address, nonce uint64) {
 	mp.lk.Lock()
 	defer mp.lk.Unlock()
 
+	mp.remove(from, nonce)
+}
+
+func (mp *MessagePool) remove(from address.Address, nonce uint64) {
 	mset, ok := mp.pending[from]
 	if !ok {
 		return

chain/messagepool/messagepool_test.go

@@ -233,3 +233,49 @@ func TestRevertMessages(t *testing.T) {
 	}
 
 }
+
+func TestPruningSimple(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)
+	}
+
+	sender, err := w.GenerateKey(crypto.SigTypeBLS)
+	if err != nil {
+		t.Fatal(err)
+	}
+	target := mock.Address(1001)
+
+	for i := 0; i < 5; i++ {
+		smsg := mock.MkMessage(sender, target, uint64(i), w)
+		if err := mp.Add(smsg); err != nil {
+			t.Fatal(err)
+		}
+	}
+
+	for i := 10; i < 50; i++ {
+		smsg := mock.MkMessage(sender, target, uint64(i), w)
+		if err := mp.Add(smsg); err != nil {
+			t.Fatal(err)
+		}
+	}
+
+	mp.maxTxPoolSizeHi = 40
+	mp.maxTxPoolSizeLo = 10
+
+	mp.Prune()
+
+	msgs, _ := mp.Pending()
+	if len(msgs) != 5 {
+		t.Fatal("expected only 5 messages in pool")
+	}
+}

chain/messagepool/pruning.go

@@ -1,10 +1,20 @@
 package messagepool
 
 import (
+	"bytes"
+	"context"
+	big2 "math/big"
 	"sort"
 	"time"
 
+	"github.com/filecoin-project/go-address"
+	"github.com/filecoin-project/lotus/build"
+	"github.com/filecoin-project/lotus/chain/messagepool/gasguess"
 	"github.com/filecoin-project/lotus/chain/types"
+	"github.com/filecoin-project/lotus/chain/vm"
+	"github.com/filecoin-project/specs-actors/actors/abi"
+	"github.com/filecoin-project/specs-actors/actors/abi/big"
+	"github.com/ipfs/go-cid"
 	"golang.org/x/xerrors"
 )
 
@@ -40,7 +50,229 @@ func (mp *MessagePool) pruneExcessMessages() error {
 	return nil
 }
 
+// just copied from miner/ SelectMessages
+func (mp *MessagePool) pruneMessages(ctx context.Context) error {
+	al := func(ctx context.Context, addr address.Address, tsk types.TipSetKey) (*types.Actor, error) {
+		return mp.api.StateGetActor(addr, mp.curTs)
+	}
+
+	msgs := make([]*types.SignedMessage, 0, mp.currentSize)
+	for a := range mp.pending {
+		msgs = append(msgs, mp.pendingFor(a)...)
+	}
+
+	type senderMeta struct {
+		lastReward   abi.TokenAmount
+		lastGasLimit int64
+
+		gasReward []abi.TokenAmount
+		gasLimit  []int64
+
+		msgs []*types.SignedMessage
+	}
+
+	inclNonces := make(map[address.Address]uint64)
+	inclBalances := make(map[address.Address]big.Int)
+	outBySender := make(map[address.Address]*senderMeta)
+
+	tooLowFundMsgs := 0
+	tooHighNonceMsgs := 0
+
+	start := build.Clock.Now()
+	vmValid := time.Duration(0)
+	getbal := time.Duration(0)
+	guessGasDur := time.Duration(0)
+
+	mp.Pending()
+	sort.Slice(msgs, func(i, j int) bool {
+		return msgs[i].Message.Nonce < msgs[j].Message.Nonce
+	})
+
+	for _, msg := range msgs {
+		vmstart := build.Clock.Now()
+
+		minGas := vm.PricelistByEpoch(mp.curTs.Height()).OnChainMessage(msg.ChainLength()) // TODO: really should be doing just msg.ChainLength() but the sync side of this code doesnt seem to have access to that
+		if err := msg.VMMessage().ValidForBlockInclusion(minGas.Total()); err != nil {
+			log.Warnf("invalid message in message pool: %s", err)
+			continue
+		}
+
+		vmValid += build.Clock.Since(vmstart)
+
+		// TODO: this should be in some more general 'validate message' call
+		if msg.Message.GasLimit > build.BlockGasLimit {
+			log.Warnf("message in mempool had too high of a gas limit (%d)", msg.Message.GasLimit)
+			continue
+		}
+
+		if msg.Message.To == address.Undef {
+			log.Warnf("message in mempool had bad 'To' address")
+			continue
+		}
+
+		from := msg.Message.From
+
+		getBalStart := build.Clock.Now()
+		if _, ok := inclNonces[from]; !ok {
+			act, err := mp.api.StateGetActor(from, nil)
+			if err != nil {
+				log.Warnf("failed to check message sender balance, skipping message: %+v", err)
+				continue
+			}
+
+			inclNonces[from] = act.Nonce
+			inclBalances[from] = act.Balance
+		}
+		getbal += build.Clock.Since(getBalStart)
+
+		if inclBalances[from].LessThan(msg.Message.RequiredFunds()) {
+			tooLowFundMsgs++
+			// todo: drop from mpool
+			continue
+		}
+
+		if msg.Message.Nonce > inclNonces[from] {
+			tooHighNonceMsgs++
+			continue
+		}
+
+		if msg.Message.Nonce < inclNonces[from] {
+			continue
+		}
+
+		inclNonces[from] = msg.Message.Nonce + 1
+		inclBalances[from] = types.BigSub(inclBalances[from], msg.Message.RequiredFunds())
+		sm := outBySender[from]
+		if sm == nil {
+			sm = &senderMeta{
+				lastReward: big.Zero(),
+			}
+		}
+
+		sm.gasLimit = append(sm.gasLimit, sm.lastGasLimit+msg.Message.GasLimit)
+		sm.lastGasLimit = sm.gasLimit[len(sm.gasLimit)-1]
+
+		guessGasStart := build.Clock.Now()
+		guessedGas, err := gasguess.GuessGasUsed(ctx, types.EmptyTSK, msg, al)
+		guessGasDur += build.Clock.Since(guessGasStart)
+		if err != nil {
+			log.Infow("failed to guess gas", "to", msg.Message.To, "method", msg.Message.Method, "err", err)
+		}
+
+		estimatedReward := big.Mul(types.NewInt(uint64(guessedGas)), msg.Message.GasPrice)
+
+		sm.gasReward = append(sm.gasReward, big.Add(sm.lastReward, estimatedReward))
+		sm.lastReward = sm.gasReward[len(sm.gasReward)-1]
+
+		sm.msgs = append(sm.msgs, msg)
+
+		outBySender[from] = sm
+	}
+
+	gasLimitLeft := int64(build.BlockGasLimit)
+
+	orderedSenders := make([]address.Address, 0, len(outBySender))
+	for k := range outBySender {
+		orderedSenders = append(orderedSenders, k)
+	}
+	sort.Slice(orderedSenders, func(i, j int) bool {
+		return bytes.Compare(orderedSenders[i].Bytes(), orderedSenders[j].Bytes()) == -1
+	})
+
+	out := make([]*types.SignedMessage, 0, build.BlockMessageLimit)
+	{
+		for {
+			var bestSender address.Address
+			var nBest int
+			var bestGasToReward float64
+
+			// TODO: This is O(n^2)-ish, could use something like container/heap to cache this math
+			for _, sender := range orderedSenders {
+				meta, ok := outBySender[sender]
+				if !ok {
+					continue
+				}
+				for n := range meta.msgs {
+					if meta.gasLimit[n] > gasLimitLeft {
+						break
+					}
+
+					if n+len(out) > build.BlockMessageLimit {
+						break
+					}
+
+					gasToReward, _ := new(big2.Float).SetInt(meta.gasReward[n].Int).Float64()
+					gasToReward /= float64(meta.gasLimit[n])
+
+					if gasToReward >= bestGasToReward {
+						bestSender = sender
+						nBest = n + 1
+						bestGasToReward = gasToReward
+					}
+				}
+			}
+
+			if nBest == 0 {
+				break // block gas limit reached
+			}
+
+			{
+				out = append(out, outBySender[bestSender].msgs[:nBest]...)
+				gasLimitLeft -= outBySender[bestSender].gasLimit[nBest-1]
+
+				outBySender[bestSender].msgs = outBySender[bestSender].msgs[nBest:]
+				outBySender[bestSender].gasLimit = outBySender[bestSender].gasLimit[nBest:]
+				outBySender[bestSender].gasReward = outBySender[bestSender].gasReward[nBest:]
+
+				if len(outBySender[bestSender].msgs) == 0 {
+					delete(outBySender, bestSender)
+				}
+			}
+
+			if len(out) >= build.BlockMessageLimit {
+				break
+			}
+		}
+	}
+
+	if tooLowFundMsgs > 0 {
+		log.Warnf("%d messages in mempool does not have enough funds", tooLowFundMsgs)
+	}
+
+	if tooHighNonceMsgs > 0 {
+		log.Warnf("%d messages in mempool had too high nonce", tooHighNonceMsgs)
+	}
+
+	sm := build.Clock.Now()
+	if sm.Sub(start) > time.Second {
+		log.Warnw("SelectMessages took a long time",
+			"duration", sm.Sub(start),
+			"vmvalidate", vmValid,
+			"getbalance", getbal,
+			"guessgas", guessGasDur,
+			"msgs", len(msgs))
+	}
+
+	if len(out) > mp.maxTxPoolSizeLo {
+		out = out[:mp.maxTxPoolSizeLo]
+	}
+
+	good := make(map[cid.Cid]bool)
+	for _, m := range out {
+		good[m.Cid()] = true
+	}
+
+	for _, m := range msgs {
+		if !good[m.Cid()] {
+			mp.remove(m.Message.From, m.Message.Nonce)
+		}
+	}
+
+	return nil
+}
+
 func (mp *MessagePool) pruneFutureMessages() (int, error) {
+	var pruned int
 	for addr, ms := range mp.pending {
 		if _, ok := mp.localAddrs[addr]; ok {
 			continue
@@ -67,6 +299,7 @@ func (mp *MessagePool) pruneFutureMessages() (int, error) {
 			} else {
 				ms.nextNonce = start
 				for ; i < len(allmsgs); i++ {
+					pruned++
 					delete(ms.msgs, allmsgs[i].Message.Nonce)
 				}
 				break
@@ -74,4 +307,5 @@ func (mp *MessagePool) pruneFutureMessages() (int, error) {
 		}
 
 	}
+	return pruned, nil
 }

miner/miner.go

@@ -17,6 +17,7 @@ import (
 	"github.com/filecoin-project/lotus/api"
 	"github.com/filecoin-project/lotus/build"
 	"github.com/filecoin-project/lotus/chain/gen"
+	"github.com/filecoin-project/lotus/chain/messagepool/gasguess"
 	"github.com/filecoin-project/lotus/chain/store"
 	"github.com/filecoin-project/lotus/chain/types"
 
@@ -448,7 +449,7 @@ type actCacheEntry struct {
 type cachedActorLookup struct {
 	tsk      types.TipSetKey
 	cache    map[address.Address]actCacheEntry
-	fallback ActorLookup
+	fallback gasguess.ActorLookup
 }
 
 func (c *cachedActorLookup) StateGetActor(ctx context.Context, a address.Address, tsk types.TipSetKey) (*types.Actor, error) {

miner/miner_test.go

@@ -8,6 +8,7 @@ import (
 	"github.com/filecoin-project/specs-actors/actors/builtin"
 
 	"github.com/filecoin-project/lotus/build"
+	"github.com/filecoin-project/lotus/chain/messagepool/gasguess"
 	"github.com/filecoin-project/lotus/chain/types"
 	"github.com/stretchr/testify/assert"
 )
@@ -49,7 +50,7 @@ func TestSelectNotOverLimited(t *testing.T) {
 		return actors[addr], nil
 	}
 
-	gasUsed := costs[costKey{builtin.StorageMinerActorCodeID, 4}]
+	gasUsed := gasguess.Costs[gasguess.CostKey{builtin.StorageMinerActorCodeID, 4}]
 
 	var goodMsgs []types.Message
 	for i := int64(0); i < build.BlockGasLimit/gasUsed+10; i++ {

miner/selectmessages.go

@@ -9,13 +9,14 @@ import (
 
 	"github.com/filecoin-project/go-address"
 	"github.com/filecoin-project/lotus/build"
+	"github.com/filecoin-project/lotus/chain/messagepool/gasguess"
 	"github.com/filecoin-project/lotus/chain/types"
 	"github.com/filecoin-project/lotus/chain/vm"
 	"github.com/filecoin-project/specs-actors/actors/abi"
 	"github.com/filecoin-project/specs-actors/actors/abi/big"
 )
 
-func SelectMessages(ctx context.Context, al ActorLookup, ts *types.TipSet, msgs []*types.SignedMessage) ([]*types.SignedMessage, error) {
+func SelectMessages(ctx context.Context, al gasguess.ActorLookup, ts *types.TipSet, msgs []*types.SignedMessage) ([]*types.SignedMessage, error) {
 	al = (&cachedActorLookup{
 		tsk:      ts.Key(),
 		cache:    map[address.Address]actCacheEntry{},
@@ -114,7 +115,7 @@ func SelectMessages(ctx context.Context, al ActorLookup, ts *types.TipSet, msgs
 		sm.lastGasLimit = sm.gasLimit[len(sm.gasLimit)-1]
 
 		guessGasStart := build.Clock.Now()
-		guessedGas, err := GuessGasUsed(ctx, ts.Key(), msg, al)
+		guessedGas, err := gasguess.GuessGasUsed(ctx, ts.Key(), msg, al)
 		guessGasDur += build.Clock.Since(guessGasStart)
 		if err != nil {
 			log.Infow("failed to guess gas", "to", msg.Message.To, "method", msg.Message.Method, "err", err)