lotus/cli/mpool.go

231 lines
4.1 KiB
Go
Raw Normal View History

2019-07-09 22:58:51 +00:00
package cli
import (
"encoding/json"
2019-07-09 22:58:51 +00:00
"fmt"
2020-03-10 00:44:08 +00:00
"sort"
2019-07-09 22:58:51 +00:00
"github.com/urfave/cli/v2"
2020-06-05 22:59:01 +00:00
"golang.org/x/xerrors"
2019-11-23 01:26:32 +00:00
"github.com/filecoin-project/go-address"
2020-03-10 00:26:10 +00:00
2019-11-23 01:26:32 +00:00
"github.com/filecoin-project/lotus/chain/types"
2019-07-09 22:58:51 +00:00
)
var mpoolCmd = &cli.Command{
Name: "mpool",
Usage: "Manage message pool",
Subcommands: []*cli.Command{
mpoolPending,
2019-11-17 07:44:06 +00:00
mpoolSub,
2019-11-23 01:26:32 +00:00
mpoolStat,
2019-07-09 22:58:51 +00:00
},
}
var mpoolPending = &cli.Command{
Name: "pending",
Usage: "Get pending messages",
2020-04-29 19:52:04 +00:00
Flags: []cli.Flag{
&cli.BoolFlag{
2020-04-30 11:19:37 +00:00
Name: "local",
2020-04-29 19:52:04 +00:00
Usage: "print pending messages for addresses in local wallet only",
},
},
2019-07-09 22:58:51 +00:00
Action: func(cctx *cli.Context) error {
2019-10-03 18:12:30 +00:00
api, closer, err := GetFullNodeAPI(cctx)
2019-07-12 04:09:04 +00:00
if err != nil {
return err
}
2019-10-03 18:12:30 +00:00
defer closer()
2019-07-12 04:09:04 +00:00
2019-07-18 23:16:23 +00:00
ctx := ReqContext(cctx)
2019-07-09 22:58:51 +00:00
2020-04-29 19:52:04 +00:00
var filter map[address.Address]struct{}
if cctx.Bool("local") {
filter = map[address.Address]struct{}{}
addrss, err := api.WalletList(ctx)
if err != nil {
return xerrors.Errorf("getting local addresses: %w", err)
}
for _, a := range addrss {
filter[a] = struct{}{}
}
}
msgs, err := api.MpoolPending(ctx, types.EmptyTSK)
2019-07-09 22:58:51 +00:00
if err != nil {
return err
}
for _, msg := range msgs {
2020-04-30 11:19:37 +00:00
if filter != nil {
2020-04-29 19:52:04 +00:00
if _, has := filter[msg.Message.From]; !has {
continue
}
}
out, err := json.MarshalIndent(msg, "", " ")
if err != nil {
return err
}
fmt.Println(string(out))
2019-07-09 22:58:51 +00:00
}
return nil
},
}
2019-11-17 07:44:06 +00:00
var mpoolSub = &cli.Command{
Name: "sub",
Usage: "Subscibe to mpool changes",
Action: func(cctx *cli.Context) error {
api, closer, err := GetFullNodeAPI(cctx)
if err != nil {
return err
}
defer closer()
ctx := ReqContext(cctx)
sub, err := api.MpoolSub(ctx)
if err != nil {
return err
}
for {
select {
case update := <-sub:
out, err := json.MarshalIndent(update, "", " ")
if err != nil {
return err
}
fmt.Println(string(out))
case <-ctx.Done():
return nil
}
}
},
}
2019-11-23 01:26:32 +00:00
type statBucket struct {
msgs map[uint64]*types.SignedMessage
}
2020-03-10 00:44:08 +00:00
type mpStat struct {
addr string
past, cur, future uint64
}
2019-11-23 01:26:32 +00:00
var mpoolStat = &cli.Command{
Name: "stat",
Usage: "print mempool stats",
2020-04-29 19:52:04 +00:00
Flags: []cli.Flag{
&cli.BoolFlag{
2020-04-30 11:19:37 +00:00
Name: "local",
2020-04-29 19:52:04 +00:00
Usage: "print stats for addresses in local wallet only",
},
},
2019-11-23 01:26:32 +00:00
Action: func(cctx *cli.Context) error {
api, closer, err := GetFullNodeAPI(cctx)
if err != nil {
return err
}
defer closer()
ctx := ReqContext(cctx)
ts, err := api.ChainHead(ctx)
if err != nil {
return xerrors.Errorf("getting chain head: %w", err)
}
2020-04-29 19:52:04 +00:00
var filter map[address.Address]struct{}
if cctx.Bool("local") {
filter = map[address.Address]struct{}{}
addrss, err := api.WalletList(ctx)
if err != nil {
return xerrors.Errorf("getting local addresses: %w", err)
}
for _, a := range addrss {
filter[a] = struct{}{}
}
}
msgs, err := api.MpoolPending(ctx, types.EmptyTSK)
2019-11-23 01:26:32 +00:00
if err != nil {
return err
}
buckets := map[address.Address]*statBucket{}
for _, v := range msgs {
2020-04-30 11:19:37 +00:00
if filter != nil {
2020-04-29 19:52:04 +00:00
if _, has := filter[v.Message.From]; !has {
continue
}
}
2019-11-23 01:26:32 +00:00
bkt, ok := buckets[v.Message.From]
if !ok {
bkt = &statBucket{
msgs: map[uint64]*types.SignedMessage{},
}
buckets[v.Message.From] = bkt
}
bkt.msgs[v.Message.Nonce] = v
}
2020-03-10 00:44:08 +00:00
var out []mpStat
2019-11-23 01:26:32 +00:00
for a, bkt := range buckets {
act, err := api.StateGetActor(ctx, a, ts.Key())
2019-11-23 01:26:32 +00:00
if err != nil {
fmt.Printf("%s, err: %s\n", a, err)
continue
2019-11-23 01:26:32 +00:00
}
cur := act.Nonce
for {
_, ok := bkt.msgs[cur]
if !ok {
break
}
cur++
}
2020-03-10 00:44:08 +00:00
past := uint64(0)
future := uint64(0)
2019-11-23 19:01:56 +00:00
for _, m := range bkt.msgs {
if m.Message.Nonce < act.Nonce {
past++
}
if m.Message.Nonce > cur {
future++
}
}
2020-03-10 00:44:08 +00:00
out = append(out, mpStat{
addr: a.String(),
past: past,
cur: cur - act.Nonce,
future: future,
})
}
sort.Slice(out, func(i, j int) bool {
return out[i].addr < out[j].addr
})
for _, stat := range out {
fmt.Printf("%s, past: %d, cur: %d, future: %d\n", stat.addr, stat.past, stat.cur, stat.future)
2019-11-23 01:26:32 +00:00
}
return nil
},
}