cerc-io/lotus
16
0
Fork 0
Code Issues Pull Requests Packages Projects Releases 27 Wiki Activity

Merge pull request #2766 from filecoin-project/feat/mpool-pruning

Browse Source 
Implement some basic pruning strategy
This commit is contained in:
Łukasz Magiera 2020-08-02 01:47:48 +02:00 committed by GitHub
commit fa240a37db
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
8 changed files with 377 additions and 30 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

14
miner/guessgas.go → chain/messagepool/gasguess/guessgas.go
View File

@ -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 {
code cid.Cid
m abi.MethodNum
type CostKey struct {
Code cid.Cid
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")
}

77
chain/messagepool/messagepool.go
View File

@ -44,6 +44,9 @@ const ReplaceByFeeRatio = 1.25
const repubMsgLimit = 5
var MemPoolSizeLimitHiDefault = 50000
var MemPoolSizeLimitLoDefault = 40000
var (
rbfNum = types.NewInt(uint64((ReplaceByFeeRatio - 1) * 256))
rbfDenom = types.NewInt(256)
@ -86,7 +89,12 @@ type MessagePool struct {
minGasPrice types.BigInt
maxTxPoolSize int
currentSize int
maxTxPoolSizeHi int
maxTxPoolSizeLo int
// pruneTrigger is a channel used to trigger a mempool pruning
pruneTrigger chan struct{}
blsSigCache *lru.TwoQueueCache
@ -110,7 +118,7 @@ func newMsgSet() *msgSet {
}
}
func (ms *msgSet) add(m *types.SignedMessage) error {
func (ms *msgSet) add(m *types.SignedMessage) (bool, error) {
if len(ms.msgs) == 0 || m.Message.Nonce >= ms.nextNonce {
ms.nextNonce = m.Message.Nonce + 1
}
@ -126,7 +134,7 @@ func (ms *msgSet) add(m *types.SignedMessage) error {
"newprice", m.Message.GasPrice, "addr", m.Message.From, "nonce", m.Message.Nonce)
} else {
log.Info("add with duplicate nonce")
return xerrors.Errorf("message from %s with nonce %d already in mpool,"+
return false, xerrors.Errorf("message from %s with nonce %d already in mpool,"+
" increase GasPrice to %s from %s to trigger replace by fee",
m.Message.From, m.Message.Nonce, minPrice, m.Message.GasPrice)
}
@ -134,7 +142,7 @@ func (ms *msgSet) add(m *types.SignedMessage) error {
}
ms.msgs[m.Message.Nonce] = m
return nil
return !has, nil
}
type Provider interface {
@ -196,25 +204,27 @@ func New(api Provider, ds dtypes.MetadataDS, netName dtypes.NetworkName) (*Messa
verifcache, _ := lru.New2Q(build.VerifSigCacheSize)
mp := &MessagePool{
closer: make(chan struct{}),
repubTk: build.Clock.Ticker(time.Duration(build.BlockDelaySecs) * 10 * time.Second),
localAddrs: make(map[address.Address]struct{}),
pending: make(map[address.Address]*msgSet),
minGasPrice: types.NewInt(0),
maxTxPoolSize: 5000,
blsSigCache: cache,
sigValCache: verifcache,
changes: lps.New(50),
localMsgs: namespace.Wrap(ds, datastore.NewKey(localMsgsDs)),
api: api,
netName: netName,
closer: make(chan struct{}),
repubTk: build.Clock.Ticker(time.Duration(build.BlockDelaySecs) * 10 * time.Second),
localAddrs: make(map[address.Address]struct{}),
pending: make(map[address.Address]*msgSet),
minGasPrice: types.NewInt(0),
maxTxPoolSizeHi: MemPoolSizeLimitHiDefault,
maxTxPoolSizeLo: MemPoolSizeLimitLoDefault,
pruneTrigger: make(chan struct{}, 1),
blsSigCache: cache,
sigValCache: verifcache,
changes: lps.New(50),
localMsgs: namespace.Wrap(ds, datastore.NewKey(localMsgsDs)),
api: api,
netName: netName,
}
if err := mp.loadLocal(); err != nil {
log.Errorf("loading local messages: %+v", err)
}
go mp.repubLocal()
go mp.runLoop()
mp.curTs = api.SubscribeHeadChanges(func(rev, app []*types.TipSet) error {
err := mp.HeadChange(rev, app)
@ -232,7 +242,13 @@ func (mp *MessagePool) Close() error {
return nil
}
func (mp *MessagePool) repubLocal() {
func (mp *MessagePool) Prune() {
mp.pruneTrigger <- struct{}{}
mp.pruneTrigger <- struct{}{}
mp.pruneTrigger <- struct{}{}
}
func (mp *MessagePool) runLoop() {
for {
select {
case <-mp.repubTk.C:
@ -294,6 +310,10 @@ func (mp *MessagePool) repubLocal() {
if errout != nil {
log.Errorf("errors while republishing: %+v", errout)
}
case <-mp.pruneTrigger:
if err := mp.pruneExcessMessages(); err != nil {
log.Errorf("failed to prune excess messages from mempool: %s", err)
}
case <-mp.closer:
mp.repubTk.Stop()
return
@ -466,8 +486,21 @@ func (mp *MessagePool) addLocked(m *types.SignedMessage) error {
mp.pending[m.Message.From] = mset
}
if err := mset.add(m); err != nil {
incr, err := mset.add(m)
if err != nil {
log.Info(err)
return err // TODO(review): this error return was dropped at some point, was it on purpose?
}
if incr {
mp.currentSize++
if mp.currentSize > mp.maxTxPoolSizeHi {
// send signal to prune messages if it hasnt already been sent
select {
case mp.pruneTrigger <- struct{}{}:
default:
}
}
}
mp.changes.Pub(api.MpoolUpdate{
@ -600,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
@ -610,6 +647,8 @@ func (mp *MessagePool) Remove(from address.Address, nonce uint64) {
Type: api.MpoolRemove,
Message: m,
}, localUpdates)
mp.currentSize--
}
// NB: This deletes any message with the given nonce. This makes sense

54
chain/messagepool/messagepool_test.go
View File

@ -14,8 +14,13 @@ import (
"github.com/filecoin-project/specs-actors/actors/crypto"
"github.com/ipfs/go-cid"
"github.com/ipfs/go-datastore"
logging "github.com/ipfs/go-log/v2"
)
func init() {
_ = logging.SetLogLevel("*", "INFO")
}
type testMpoolAPI struct {
cb func(rev, app []*types.TipSet) error
@ -233,3 +238,52 @@ 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)
}
a := mock.MkBlock(nil, 1, 1)
tma.applyBlock(t, a)
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, got: ", len(msgs))
}
}

249
chain/messagepool/pruning.go Normal file
View File

@ -0,0 +1,249 @@
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"
)
func (mp *MessagePool) pruneExcessMessages() error {
start := time.Now()
defer func() {
log.Infow("message pruning complete", "took", time.Since(start))
}()
mp.curTsLk.Lock()
ts := mp.curTs
mp.curTsLk.Unlock()
mp.lk.Lock()
defer mp.lk.Unlock()
if mp.currentSize < mp.maxTxPoolSizeHi {
return nil
}
return mp.pruneMessages(context.TODO(), ts)
}
// just copied from miner/ SelectMessages
func (mp *MessagePool) pruneMessages(ctx context.Context, ts *types.TipSet) error {
al := func(ctx context.Context, addr address.Address, tsk types.TipSetKey) (*types.Actor, error) {
return mp.api.StateGetActor(addr, ts)
}
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)
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(ts.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
}
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, mp.maxTxPoolSizeLo)
{
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 n+len(out) >= mp.maxTxPoolSizeLo {
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]...)
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) >= mp.maxTxPoolSizeLo {
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))
}
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
}

2
chain/types/mock/chain.go
View File

@ -27,7 +27,7 @@ func MkMessage(from, to address.Address, nonce uint64, w *wallet.Wallet) *types.
From: from,
Value: types.NewInt(1),
Nonce: nonce,
GasLimit: 1,
GasLimit: 1000000,
GasPrice: types.NewInt(0),
}

3
miner/miner.go
View File

@ -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) {

3
miner/miner_test.go
View File

@ -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++ {

5
miner/selectmessages.go
View File

@ -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)