Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Jump to bottom

feat: add send epoch, time, elapsed epoch and elapsed time for each message in mpool to UI #1523

Merged
merged 5 commits into from
Jun 15, 2023
Merged

feat: add send epoch, time, elapsed epoch and elapsed time for each message in mpool to UI #1523

Show file tree
Hide file tree
Changes from 2 commits
Commits
Show all changes
5 commits
Select commit Hold shift + click to select a range
15aa362
add message epoch/time details
Jun 14, 2023
6ae47d6
implement suggestion
Jun 14, 2023
33309fc
use moment lib
Jun 15, 2023
d0a25ec
fix alerting bug
Jun 15, 2023
a5a9f53
update polling interval
Jun 15, 2023
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
30 changes: 17 additions & 13 deletions gql/resolver_mpool.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -8,6 +8,7 @@ import (
"reflect"

gqltypes "github.com/filecoin-project/boost/gql/types"
"github.com/filecoin-project/boost/lib/mpoolmonitor"
"github.com/filecoin-project/go-address"
"github.com/filecoin-project/go-state-types/abi"
"github.com/filecoin-project/lotus/chain/consensus"
Expand All @@ -16,6 +17,7 @@ import (
)

type msg struct {
SentEpoch gqltypes.Uint64
To string
From string
Nonce gqltypes.Uint64
Expand All @@ -31,12 +33,13 @@ type msg struct {
// query: mpool(local): [Message]
func (r *resolver) Mpool(ctx context.Context, args struct{ Local bool }) ([]*msg, error) {
var ret []*msg
var msgs []*types.SignedMessage
var msgs []*mpoolmonitor.TimeStampedMsg

ts, err := r.fullNode.ChainHead(ctx)
if err != nil {
return nil, fmt.Errorf("failed to get chain head: %w", err)
}

baseFee := ts.Blocks()[0].ParentBaseFee

if args.Local {
Expand All @@ -54,17 +57,17 @@ func (r *resolver) Mpool(ctx context.Context, args struct{ Local bool }) ([]*msg
// Convert params to human-readable and get method name
for _, m := range msgs {
var params string
methodName := m.Message.Method.String()
toact, err := r.fullNode.StateGetActor(ctx, m.Message.To, types.EmptyTSK)
methodName := m.SignedMessage.Message.Method.String()
toact, err := r.fullNode.StateGetActor(ctx, m.SignedMessage.Message.To, types.EmptyTSK)
if err == nil {
method, ok := consensus.NewActorRegistry().Methods[toact.Code][m.Message.Method]
method, ok := consensus.NewActorRegistry().Methods[toact.Code][m.SignedMessage.Message.Method]
if ok {
methodName = method.Name

params = string(m.Message.Params)
params = string(m.SignedMessage.Message.Params)
p, ok := reflect.New(method.Params.Elem()).Interface().(cbg.CBORUnmarshaler)
if ok {
if err := p.UnmarshalCBOR(bytes.NewReader(m.Message.Params)); err == nil {
if err := p.UnmarshalCBOR(bytes.NewReader(m.SignedMessage.Message.Params)); err == nil {
b, err := json.MarshalIndent(p, "", " ")
if err == nil {
params = string(b)
Expand All @@ -75,13 +78,14 @@ func (r *resolver) Mpool(ctx context.Context, args struct{ Local bool }) ([]*msg
}

ret = append(ret, &msg{
To: m.Message.To.String(),
From: m.Message.From.String(),
Nonce: gqltypes.Uint64(m.Message.Nonce),
Value: gqltypes.BigInt{Int: m.Message.Value},
GasFeeCap: gqltypes.BigInt{Int: m.Message.GasFeeCap},
GasLimit: gqltypes.Uint64(uint64(m.Message.GasLimit)),
GasPremium: gqltypes.BigInt{Int: m.Message.GasPremium},
SentEpoch: gqltypes.Uint64(m.Added),
To: m.SignedMessage.Message.To.String(),
From: m.SignedMessage.Message.From.String(),
Nonce: gqltypes.Uint64(m.SignedMessage.Message.Nonce),
Value: gqltypes.BigInt{Int: m.SignedMessage.Message.Value},
GasFeeCap: gqltypes.BigInt{Int: m.SignedMessage.Message.GasFeeCap},
GasLimit: gqltypes.Uint64(uint64(m.SignedMessage.Message.GasLimit)),
GasPremium: gqltypes.BigInt{Int: m.SignedMessage.Message.GasPremium},
Method: methodName,
Params: params,
BaseFee: gqltypes.BigInt{Int: baseFee},
Expand Down
1 change: 1 addition & 0 deletions gql/schema.graphql
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -363,6 +363,7 @@ type TransferStats {
}

type MpoolMessage {
SentEpoch: Uint64!
From: String!
To: String!
Nonce: Uint64!
Expand Down
43 changes: 23 additions & 20 deletions lib/mpoolmonitor/mpoolmonitor.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -16,10 +16,11 @@ import (

var log = logging.Logger("mpoolmonitor")

// timeStampedMsg wraps the pending msg with a chainEpoch
type timeStampedMsg struct {
m types.SignedMessage
added abi.ChainEpoch // Epoch when message was first noticed in mpool
// TimeStampedMsg wraps the pending msg with a chainEpoch
type TimeStampedMsg struct {
SignedMessage types.SignedMessage
Added abi.ChainEpoch // Epoch when message was first noticed in mpool
AddedTime time.Time
}

type MpoolMonitor struct {
Expand All @@ -28,14 +29,14 @@ type MpoolMonitor struct {
fullNode v1api.FullNode
lk sync.Mutex
mpoolAlertEpochs abi.ChainEpoch
msgs map[cid.Cid]*timeStampedMsg
msgs map[cid.Cid]*TimeStampedMsg
}

func NewMonitor(fullNode v1api.FullNode, mpoolAlertEpochs int64) *MpoolMonitor {
return &MpoolMonitor{
fullNode: fullNode,
mpoolAlertEpochs: abi.ChainEpoch(mpoolAlertEpochs),
msgs: make(map[cid.Cid]*timeStampedMsg),
msgs: make(map[cid.Cid]*TimeStampedMsg),
}
}

Expand Down Expand Up @@ -71,10 +72,11 @@ func (mm *MpoolMonitor) startMonitoring(ctx context.Context) {
}

// update gets the current pending messages from mpool. It updated the local
// copy of a message(not timeStampedMsg.added) if CID is already present in MpoolMonitor.msgs
// copy of a message(not TimeStampedMsg.added) if CID is already present in MpoolMonitor.msgs
// Otherwise, it inserts a new key value pair for the message. It removed any msgs not found in the
// latest output of MpoolPending
func (mm *MpoolMonitor) update(ctx context.Context) error {
t := time.Now()
ts, err := mm.fullNode.ChainHead(ctx)
if err != nil {
return fmt.Errorf("failed to get chain head: %w", err)
Expand All @@ -84,12 +86,13 @@ func (mm *MpoolMonitor) update(ctx context.Context) error {
return fmt.Errorf("getting mpool messages: %w", err)
}

newMsgs := make(map[cid.Cid]*timeStampedMsg)
newMsgs := make(map[cid.Cid]*TimeStampedMsg)

for _, pm := range msgs {
newMsgs[pm.Cid()] = &timeStampedMsg{
m: *pm,
added: ts.Height(),
newMsgs[pm.Cid()] = &TimeStampedMsg{
SignedMessage: *pm,
Added: ts.Height(),
AddedTime: t,
LexLuthr marked this conversation as resolved.
Show resolved Hide resolved
}
}

Expand All @@ -100,7 +103,7 @@ func (mm *MpoolMonitor) update(ctx context.Context) error {
for k, v := range newMsgs {
_, ok := mm.msgs[k]
if ok {
mm.msgs[k].m = v.m
mm.msgs[k].SignedMessage = v.SignedMessage
} else {
mm.msgs[k] = v
}
Expand All @@ -117,9 +120,9 @@ func (mm *MpoolMonitor) update(ctx context.Context) error {
return nil
}

func (mm *MpoolMonitor) PendingLocal(ctx context.Context) ([]*types.SignedMessage, error) {
func (mm *MpoolMonitor) PendingLocal(ctx context.Context) ([]*TimeStampedMsg, error) {
localAddr := make(map[string]struct{})
var ret []*types.SignedMessage
var ret []*TimeStampedMsg

addrs, err := mm.fullNode.WalletList(ctx)
if err != nil {
Expand All @@ -134,23 +137,23 @@ func (mm *MpoolMonitor) PendingLocal(ctx context.Context) ([]*types.SignedMessag
defer mm.lk.Unlock()

for _, msg := range mm.msgs {
if _, has := localAddr[msg.m.Message.From.String()]; !has {
if _, has := localAddr[msg.SignedMessage.Message.From.String()]; !has {
continue
}
ret = append(ret, &msg.m)
ret = append(ret, msg)
}

return ret, nil
}

func (mm *MpoolMonitor) PendingAll() ([]*types.SignedMessage, error) {
var ret []*types.SignedMessage
func (mm *MpoolMonitor) PendingAll() ([]*TimeStampedMsg, error) {
var ret []*TimeStampedMsg

mm.lk.Lock()
defer mm.lk.Unlock()

for _, msg := range mm.msgs {
ret = append(ret, &msg.m)
ret = append(ret, msg)
}

return ret, nil
Expand All @@ -167,7 +170,7 @@ func (mm *MpoolMonitor) Alerts(ctx context.Context) ([]cid.Cid, error) {
defer mm.lk.Unlock()

for mcid := range mm.msgs {
if mm.msgs[mcid].added+mm.mpoolAlertEpochs <= ts.Height() {
if mm.msgs[mcid].Added+mm.mpoolAlertEpochs <= ts.Height() {
ret = append(ret, mcid)
}
}
Expand Down
22 changes: 21 additions & 1 deletion react/src/Mpool.js
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -4,6 +4,8 @@ import {React, useState} from "react";
import {humanFIL} from "./util";
import './Mpool.css'
import {PageContainer} from "./Components";
import {EpochQuery} from "./gql";
import {addCommas} from "./util";

export function MpoolPage(props) {
return <PageContainer pageType="mpool" title="Message Pool">
Expand All @@ -13,7 +15,7 @@ export function MpoolPage(props) {

function MpoolContent(props) {
const [local, setLocal] = useState(true)
const {loading, error, data} = useQuery(MpoolQuery, { variables: { local } })
const {loading, error, data} = useQuery(MpoolQuery, { variables: { local } , pollInterval: 4000, fetchPolicy: "network-only"})
Copy link
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Why pool every 4 seconds? This should be at least 15sec. given that block time is 30sec.

Copy link
Collaborator Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Actually, this accounts for devnet as well. Ideally, I wanted to use build.BlockDelaySecs but that comes from MpoolQuery so I went with shortest available number. Even with 4 seconds, it should not be too aggressive as we are not polling directly to lotus node.

Copy link
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Why do we care about this feature at all on a devnet? There should be no alerts or backlogged messages on a devnet. I'd rather we don't spam Lotus with unnecessary API calls -- all the pages start to add up at some point.

Copy link
Collaborator Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

I am pushing one more commit to remove all messages page and everything related to that. I will also update the polling time.

This page does not make a call to lotus API directly. It uses cached entries in mpoolManager with Boost.

Copy link
Collaborator Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

@nonsense I have adjusted the polling time and moving the other changes (Removing all messages) to a new PR.


if (loading) {
return <div>Loading...</div>
Expand Down Expand Up @@ -44,11 +46,29 @@ function MpoolMessage(props) {
const i = props.i
const msg = props.msg

const {data} = useQuery(EpochQuery)
let currEpoch = data.epoch.Epoch+''
console.log(currEpoch)
let sentEpoch = msg.SentEpoch+''
console.log(sentEpoch)
let t = (currEpoch - sentEpoch)*data.epoch.SecondsPerEpoch+''

const handleParamsClick = (el) => {
el.target.classList.toggle('expanded')
}

let hh = Math.floor(t / 3600)
t %= 3600
let mm = Math.floor(t / 60)
t %= 60
let etime = `${hh} hours and ${mm} minutes ${t} seconds`
let epochInfo = `${sentEpoch} (${currEpoch - sentEpoch} epochs / ${etime} ago)`

return <>
<tr key={"sentepoch"}>
<td>Sent At Epoch</td>
<td>{epochInfo}</td>
</tr>
<tr key={"to"}>
<td>To</td>
<td className="address">{msg.To}</td>
Expand Down
1 change: 1 addition & 0 deletions react/src/gql.js
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -620,6 +620,7 @@ const StorageAskUpdate = gql`
const MpoolQuery = gql`
query AppMpoolQuery($local: Boolean!) {
mpool(local: $local) {
SentEpoch
From
To
Nonce
Expand Down