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

Use new hello protocol

Browse Source 
Using map encoder.
Types in LatencyMessage are int64 in spec, butl 1. tupled generator
does not support int64 right now; 2. there is no harm in using uint64 here.

Signed-off-by: Jakub Sztandera <kubuxu@protocol.ai>
This commit is contained in:
Jakub Sztandera 2020-02-12 19:08:49 +01:00
parent 3aa521600a
commit 50702f309f
7 changed files with 353 additions and 39 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

24
chain/actors/cbor_gen.go
View File

@ -597,7 +597,7 @@ func (t *SectorPreCommitInfo) MarshalCBOR(w io.Writer) error {
return err
}
for _, v := range t.DealIDs {
if err := cbg.CborWriteHeader(w, cbg.MajUnsignedInt, v); err != nil {
if err := cbg.CborWriteHeader(w, cbg.MajUnsignedInt, uint64(v)); err != nil {
return err
}
}
@ -684,7 +684,7 @@ func (t *SectorPreCommitInfo) UnmarshalCBOR(r io.Reader) error {
return xerrors.Errorf("value read for array t.DealIDs was not a uint, instead got %d", maj)
}
t.DealIDs[i] = val
t.DealIDs[i] = uint64(val)
}
return nil
@ -2681,7 +2681,7 @@ func (t *PublishStorageDealResponse) MarshalCBOR(w io.Writer) error {
return err
}
for _, v := range t.DealIDs {
if err := cbg.CborWriteHeader(w, cbg.MajUnsignedInt, v); err != nil {
if err := cbg.CborWriteHeader(w, cbg.MajUnsignedInt, uint64(v)); err != nil {
return err
}
}
@ -2731,7 +2731,7 @@ func (t *PublishStorageDealResponse) UnmarshalCBOR(r io.Reader) error {
return xerrors.Errorf("value read for array t.DealIDs was not a uint, instead got %d", maj)
}
t.DealIDs[i] = val
t.DealIDs[i] = uint64(val)
}
return nil
@ -2755,7 +2755,7 @@ func (t *ActivateStorageDealsParams) MarshalCBOR(w io.Writer) error {
return err
}
for _, v := range t.Deals {
if err := cbg.CborWriteHeader(w, cbg.MajUnsignedInt, v); err != nil {
if err := cbg.CborWriteHeader(w, cbg.MajUnsignedInt, uint64(v)); err != nil {
return err
}
}
@ -2805,7 +2805,7 @@ func (t *ActivateStorageDealsParams) UnmarshalCBOR(r io.Reader) error {
return xerrors.Errorf("value read for array t.Deals was not a uint, instead got %d", maj)
}
t.Deals[i] = val
t.Deals[i] = uint64(val)
}
return nil
@ -2829,7 +2829,7 @@ func (t *ProcessStorageDealsPaymentParams) MarshalCBOR(w io.Writer) error {
return err
}
for _, v := range t.DealIDs {
if err := cbg.CborWriteHeader(w, cbg.MajUnsignedInt, v); err != nil {
if err := cbg.CborWriteHeader(w, cbg.MajUnsignedInt, uint64(v)); err != nil {
return err
}
}
@ -2879,7 +2879,7 @@ func (t *ProcessStorageDealsPaymentParams) UnmarshalCBOR(r io.Reader) error {
return xerrors.Errorf("value read for array t.DealIDs was not a uint, instead got %d", maj)
}
t.DealIDs[i] = val
t.DealIDs[i] = uint64(val)
}
return nil
@ -3077,7 +3077,7 @@ func (t *ComputeDataCommitmentParams) MarshalCBOR(w io.Writer) error {
return err
}
for _, v := range t.DealIDs {
if err := cbg.CborWriteHeader(w, cbg.MajUnsignedInt, v); err != nil {
if err := cbg.CborWriteHeader(w, cbg.MajUnsignedInt, uint64(v)); err != nil {
return err
}
}
@ -3132,7 +3132,7 @@ func (t *ComputeDataCommitmentParams) UnmarshalCBOR(r io.Reader) error {
return xerrors.Errorf("value read for array t.DealIDs was not a uint, instead got %d", maj)
}
t.DealIDs[i] = val
t.DealIDs[i] = uint64(val)
}
// t.SectorSize (uint64) (uint64)
@ -3183,7 +3183,7 @@ func (t *SectorProveCommitInfo) MarshalCBOR(w io.Writer) error {
return err
}
for _, v := range t.DealIDs {
if err := cbg.CborWriteHeader(w, cbg.MajUnsignedInt, v); err != nil {
if err := cbg.CborWriteHeader(w, cbg.MajUnsignedInt, uint64(v)); err != nil {
return err
}
}
@ -3260,7 +3260,7 @@ func (t *SectorProveCommitInfo) UnmarshalCBOR(r io.Reader) error {
return xerrors.Errorf("value read for array t.DealIDs was not a uint, instead got %d", maj)
}
t.DealIDs[i] = val
t.DealIDs[i] = uint64(val)
}
return nil

8
chain/blocksync/cbor_gen.go
View File

@ -274,7 +274,7 @@ func (t *BSTipSet) MarshalCBOR(w io.Writer) error {
return err
}
for _, v := range v {
if err := cbg.CborWriteHeader(w, cbg.MajUnsignedInt, v); err != nil {
if err := cbg.CborWriteHeader(w, cbg.MajUnsignedInt, uint64(v)); err != nil {
return err
}
}
@ -311,7 +311,7 @@ func (t *BSTipSet) MarshalCBOR(w io.Writer) error {
return err
}
for _, v := range v {
if err := cbg.CborWriteHeader(w, cbg.MajUnsignedInt, v); err != nil {
if err := cbg.CborWriteHeader(w, cbg.MajUnsignedInt, uint64(v)); err != nil {
return err
}
}
@ -437,7 +437,7 @@ func (t *BSTipSet) UnmarshalCBOR(r io.Reader) error {
return xerrors.Errorf("value read for array t.BlsMsgIncludes[i] was not a uint, instead got %d", maj)
}
t.BlsMsgIncludes[i][j] = val
t.BlsMsgIncludes[i][j] = uint64(val)
}
}
@ -519,7 +519,7 @@ func (t *BSTipSet) UnmarshalCBOR(r io.Reader) error {
return xerrors.Errorf("value read for array t.SecpkMsgIncludes[i] was not a uint, instead got %d", maj)
}
t.SecpkMsgIncludes[i][j] = val
t.SecpkMsgIncludes[i][j] = uint64(val)
}
}

10
gen/main.go
View File

@ -11,6 +11,7 @@ import (
"github.com/filecoin-project/lotus/chain/blocksync"
"github.com/filecoin-project/lotus/chain/types"
"github.com/filecoin-project/lotus/lib/statemachine"
"github.com/filecoin-project/lotus/node/hello"
"github.com/filecoin-project/lotus/paych"
"github.com/filecoin-project/lotus/storage/sealing"
)
@ -58,6 +59,15 @@ func main() {
os.Exit(1)
}
err = gen.WriteMapEncodersToFile("./node/hello/cbor_gen.go", "hello",
hello.HelloMessage{},
hello.LatencyMessage{},
)
if err != nil {
fmt.Println(err)
os.Exit(1)
}
err = gen.WriteTupleEncodersToFile("./chain/blocksync/cbor_gen.go", "blocksync",
blocksync.BlockSyncRequest{},
blocksync.BlockSyncResponse{},

2
go.mod
View File

@ -86,7 +86,7 @@ require (
github.com/prometheus/common v0.4.0
github.com/stretchr/testify v1.4.0
github.com/whyrusleeping/bencher v0.0.0-20190829221104-bb6607aa8bba
github.com/whyrusleeping/cbor-gen v0.0.0-20200123233031-1cdf64d27158
github.com/whyrusleeping/cbor-gen v0.0.0-20200206220010-03c9665e2a66
github.com/whyrusleeping/multiaddr-filter v0.0.0-20160516205228-e903e4adabd7
github.com/whyrusleeping/pubsub v0.0.0-20131020042734-02de8aa2db3d
go.opencensus.io v0.22.2

2
go.sum
View File

@ -745,6 +745,8 @@ github.com/whyrusleeping/cbor-gen v0.0.0-20191212224538-d370462a7e8a/go.mod h1:x
github.com/whyrusleeping/cbor-gen v0.0.0-20191216205031-b047b6acb3c0/go.mod h1:xdlJQaiqipF0HW+Mzpg7XRM3fWbGvfgFlcppuvlkIvY=
github.com/whyrusleeping/cbor-gen v0.0.0-20200123233031-1cdf64d27158 h1:WXhVOwj2USAXB5oMDwRl3piOux2XMV9TANaYxXHdkoE=
github.com/whyrusleeping/cbor-gen v0.0.0-20200123233031-1cdf64d27158/go.mod h1:Xj/M2wWU+QdTdRbu/L/1dIZY8/Wb2K9pAhtroQuxJJI=
github.com/whyrusleeping/cbor-gen v0.0.0-20200206220010-03c9665e2a66 h1:LolR9FiEfQNn5U031bAhn/46po2JgWHKadYbcWFIJ+0=
github.com/whyrusleeping/cbor-gen v0.0.0-20200206220010-03c9665e2a66/go.mod h1:Xj/M2wWU+QdTdRbu/L/1dIZY8/Wb2K9pAhtroQuxJJI=
github.com/whyrusleeping/chunker v0.0.0-20181014151217-fe64bd25879f h1:jQa4QT2UP9WYv2nzyawpKMOCl+Z/jW7djv2/J50lj9E=
github.com/whyrusleeping/chunker v0.0.0-20181014151217-fe64bd25879f/go.mod h1:p9UJB6dDgdPgMJZs7UjUOdulKyRr9fqkS+6JKAInPy8=
github.com/whyrusleeping/go-keyspace v0.0.0-20160322163242-5b898ac5add1 h1:EKhdznlJHPMoKr0XTrX+IlJs1LH3lyx2nfr1dOlZ79k=

302
node/hello/cbor_gen.go Normal file
View File

@ -0,0 +1,302 @@
// Code generated by github.com/whyrusleeping/cbor-gen. DO NOT EDIT.
package hello
import (
"fmt"
"io"
"github.com/ipfs/go-cid"
cbg "github.com/whyrusleeping/cbor-gen"
xerrors "golang.org/x/xerrors"
)
var _ = xerrors.Errorf
func (t *HelloMessage) MarshalCBOR(w io.Writer) error {
if t == nil {
_, err := w.Write(cbg.CborNull)
return err
}
if _, err := w.Write([]byte{164}); err != nil {
return err
}
// t.HeaviestTipSet ([]cid.Cid) (slice)
if len("HeaviestTipSet") > cbg.MaxLength {
return xerrors.Errorf("Value in field \"HeaviestTipSet\" was too long")
}
if _, err := w.Write(cbg.CborEncodeMajorType(cbg.MajTextString, uint64(len("HeaviestTipSet")))); err != nil {
return err
}
if _, err := w.Write([]byte("HeaviestTipSet")); err != nil {
return err
}
if len(t.HeaviestTipSet) > cbg.MaxLength {
return xerrors.Errorf("Slice value in field t.HeaviestTipSet was too long")
}
if _, err := w.Write(cbg.CborEncodeMajorType(cbg.MajArray, uint64(len(t.HeaviestTipSet)))); err != nil {
return err
}
for _, v := range t.HeaviestTipSet {
if err := cbg.WriteCid(w, v); err != nil {
return xerrors.Errorf("failed writing cid field t.HeaviestTipSet: %w", err)
}
}
// t.HeaviestTipSetHeight (uint64) (uint64)
if len("HeaviestTipSetHeight") > cbg.MaxLength {
return xerrors.Errorf("Value in field \"HeaviestTipSetHeight\" was too long")
}
if _, err := w.Write(cbg.CborEncodeMajorType(cbg.MajTextString, uint64(len("HeaviestTipSetHeight")))); err != nil {
return err
}
if _, err := w.Write([]byte("HeaviestTipSetHeight")); err != nil {
return err
}
if _, err := w.Write(cbg.CborEncodeMajorType(cbg.MajUnsignedInt, uint64(t.HeaviestTipSetHeight))); err != nil {
return err
}
// t.HeaviestTipSetWeight (types.BigInt) (struct)
if len("HeaviestTipSetWeight") > cbg.MaxLength {
return xerrors.Errorf("Value in field \"HeaviestTipSetWeight\" was too long")
}
if _, err := w.Write(cbg.CborEncodeMajorType(cbg.MajTextString, uint64(len("HeaviestTipSetWeight")))); err != nil {
return err
}
if _, err := w.Write([]byte("HeaviestTipSetWeight")); err != nil {
return err
}
if err := t.HeaviestTipSetWeight.MarshalCBOR(w); err != nil {
return err
}
// t.GenesisHash (cid.Cid) (struct)
if len("GenesisHash") > cbg.MaxLength {
return xerrors.Errorf("Value in field \"GenesisHash\" was too long")
}
if _, err := w.Write(cbg.CborEncodeMajorType(cbg.MajTextString, uint64(len("GenesisHash")))); err != nil {
return err
}
if _, err := w.Write([]byte("GenesisHash")); err != nil {
return err
}
if err := cbg.WriteCid(w, t.GenesisHash); err != nil {
return xerrors.Errorf("failed to write cid field t.GenesisHash: %w", err)
}
return nil
}
func (t *HelloMessage) UnmarshalCBOR(r io.Reader) error {
br := cbg.GetPeeker(r)
maj, extra, err := cbg.CborReadHeader(br)
if err != nil {
return err
}
if maj != cbg.MajMap {
return fmt.Errorf("cbor input should be of type map")
}
if extra > cbg.MaxLength {
return fmt.Errorf("HelloMessage: map struct too large (%d)", extra)
}
var name string
n := extra
for i := uint64(0); i < n; i++ {
{
sval, err := cbg.ReadString(br)
if err != nil {
return err
}
name = string(sval)
}
switch name {
// t.HeaviestTipSet ([]cid.Cid) (slice)
case "HeaviestTipSet":
maj, extra, err = cbg.CborReadHeader(br)
if err != nil {
return err
}
if extra > cbg.MaxLength {
return fmt.Errorf("t.HeaviestTipSet: array too large (%d)", extra)
}
if maj != cbg.MajArray {
return fmt.Errorf("expected cbor array")
}
if extra > 0 {
t.HeaviestTipSet = make([]cid.Cid, extra)
}
for i := 0; i < int(extra); i++ {
c, err := cbg.ReadCid(br)
if err != nil {
return xerrors.Errorf("reading cid field t.HeaviestTipSet failed: %w", err)
}
t.HeaviestTipSet[i] = c
}
// t.HeaviestTipSetHeight (uint64) (uint64)
case "HeaviestTipSetHeight":
maj, extra, err = cbg.CborReadHeader(br)
if err != nil {
return err
}
if maj != cbg.MajUnsignedInt {
return fmt.Errorf("wrong type for uint64 field")
}
t.HeaviestTipSetHeight = uint64(extra)
// t.HeaviestTipSetWeight (types.BigInt) (struct)
case "HeaviestTipSetWeight":
{
if err := t.HeaviestTipSetWeight.UnmarshalCBOR(br); err != nil {
return err
}
}
// t.GenesisHash (cid.Cid) (struct)
case "GenesisHash":
{
c, err := cbg.ReadCid(br)
if err != nil {
return xerrors.Errorf("failed to read cid field t.GenesisHash: %w", err)
}
t.GenesisHash = c
}
default:
return fmt.Errorf("unknown struct field %d: '%s'", i, name)
}
}
return nil
}
func (t *LatencyMessage) MarshalCBOR(w io.Writer) error {
if t == nil {
_, err := w.Write(cbg.CborNull)
return err
}
if _, err := w.Write([]byte{162}); err != nil {
return err
}
// t.TArrial (uint64) (uint64)
if len("TArrial") > cbg.MaxLength {
return xerrors.Errorf("Value in field \"TArrial\" was too long")
}
if _, err := w.Write(cbg.CborEncodeMajorType(cbg.MajTextString, uint64(len("TArrial")))); err != nil {
return err
}
if _, err := w.Write([]byte("TArrial")); err != nil {
return err
}
if _, err := w.Write(cbg.CborEncodeMajorType(cbg.MajUnsignedInt, uint64(t.TArrial))); err != nil {
return err
}
// t.TSent (uint64) (uint64)
if len("TSent") > cbg.MaxLength {
return xerrors.Errorf("Value in field \"TSent\" was too long")
}
if _, err := w.Write(cbg.CborEncodeMajorType(cbg.MajTextString, uint64(len("TSent")))); err != nil {
return err
}
if _, err := w.Write([]byte("TSent")); err != nil {
return err
}
if _, err := w.Write(cbg.CborEncodeMajorType(cbg.MajUnsignedInt, uint64(t.TSent))); err != nil {
return err
}
return nil
}
func (t *LatencyMessage) UnmarshalCBOR(r io.Reader) error {
br := cbg.GetPeeker(r)
maj, extra, err := cbg.CborReadHeader(br)
if err != nil {
return err
}
if maj != cbg.MajMap {
return fmt.Errorf("cbor input should be of type map")
}
if extra > cbg.MaxLength {
return fmt.Errorf("LatencyMessage: map struct too large (%d)", extra)
}
var name string
n := extra
for i := uint64(0); i < n; i++ {
{
sval, err := cbg.ReadString(br)
if err != nil {
return err
}
name = string(sval)
}
switch name {
// t.TArrial (uint64) (uint64)
case "TArrial":
maj, extra, err = cbg.CborReadHeader(br)
if err != nil {
return err
}
if maj != cbg.MajUnsignedInt {
return fmt.Errorf("wrong type for uint64 field")
}
t.TArrial = uint64(extra)
// t.TSent (uint64) (uint64)
case "TSent":
maj, extra, err = cbg.CborReadHeader(br)
if err != nil {
return err
}
if maj != cbg.MajUnsignedInt {
return fmt.Errorf("wrong type for uint64 field")
}
t.TSent = uint64(extra)
default:
return fmt.Errorf("unknown struct field %d: '%s'", i, name)
}
}
return nil
}

44
node/hello/hello.go
View File

@ -5,14 +5,13 @@ import (
"time"
"github.com/ipfs/go-cid"
cbor "github.com/ipfs/go-ipld-cbor"
logging "github.com/ipfs/go-log/v2"
"github.com/libp2p/go-libp2p-core/host"
inet "github.com/libp2p/go-libp2p-core/network"
"github.com/libp2p/go-libp2p-core/peer"
protocol "github.com/libp2p/go-libp2p-core/protocol"
"github.com/filecoin-project/go-cbor-util"
cborutil "github.com/filecoin-project/go-cbor-util"
"github.com/filecoin-project/lotus/chain"
"github.com/filecoin-project/lotus/chain/store"
"github.com/filecoin-project/lotus/chain/types"
@ -23,17 +22,15 @@ const ProtocolID = "/fil/hello/1.0.0"
var log = logging.Logger("hello")
func init() {
cbor.RegisterCborType(Message{})
}
type Message struct {
type HelloMessage struct {
HeaviestTipSet []cid.Cid
HeaviestTipSetHeight uint64
HeaviestTipSetWeight types.BigInt
GenesisHash cid.Cid
TArrial int64
TSent int64
}
type LatencyMessage struct {
TArrial uint64
TSent uint64
}
type NewStreamFunc func(context.Context, peer.ID, ...protocol.ID) (inet.Stream, error)
@ -61,7 +58,7 @@ func NewHelloService(h host.Host, cs *store.ChainStore, syncer *chain.Syncer, pm
func (hs *Service) HandleStream(s inet.Stream) {
var hmsg Message
var hmsg HelloMessage
if err := cborutil.ReadCborRPC(s, &hmsg); err != nil {
log.Infow("failed to read hello message, diconnecting", "error", err)
s.Conn().Close()
@ -83,9 +80,9 @@ func (hs *Service) HandleStream(s inet.Stream) {
defer s.Close()
sent := time.Now()
msg := &Message{
TArrial: arrived.UnixNano(),
TSent: sent.UnixNano(),
msg := &LatencyMessage{
TArrial: uint64(arrived.UnixNano()),
TSent: uint64(sent.UnixNano()),
}
if err := cborutil.WriteCborRPC(s, msg); err != nil {
log.Debugf("error while responding to latency: %v", err)
@ -128,8 +125,9 @@ func (hs *Service) SayHello(ctx context.Context, pid peer.ID) error {
return err
}
hmsg := &Message{
hmsg := &HelloMessage{
HeaviestTipSet: hts.Cids(),
HeaviestTipSetHeight: hts.Height(),
HeaviestTipSetWeight: weight,
GenesisHash: gen.Cid(),
}
@ -143,10 +141,12 @@ func (hs *Service) SayHello(ctx context.Context, pid peer.ID) error {
go func() {
defer s.Close()
hmsg = &Message{}
lmsg := &LatencyMessage{}
s.SetReadDeadline(time.Now().Add(10 * time.Second))
err := cborutil.ReadCborRPC(s, hmsg)
ok := err == nil
err := cborutil.ReadCborRPC(s, lmsg)
if err != nil {
log.Infow("reading latency message", "error", err)
}
t3 := time.Now()
lat := t3.Sub(t0)
@ -155,10 +155,10 @@ func (hs *Service) SayHello(ctx context.Context, pid peer.ID) error {
hs.pmgr.SetPeerLatency(pid, lat)
}
if ok {
if hmsg.TArrial != 0 && hmsg.TSent != 0 {
t1 := time.Unix(0, hmsg.TArrial)
t2 := time.Unix(0, hmsg.TSent)
if err == nil {
if lmsg.TArrial != 0 && lmsg.TSent != 0 {
t1 := time.Unix(0, int64(lmsg.TArrial))
t2 := time.Unix(0, int64(lmsg.TSent))
offset := t0.Sub(t1) + t3.Sub(t2)
offset /= 2
log.Infow("time offset", "offset", offset.Seconds(), "peerid", pid.String())