lotus/node/hello/hello.go

package hello

import (
	"context"
	"time"

	"github.com/ipfs/go-cid"
	cbor "github.com/ipfs/go-ipld-cbor"
	logging "github.com/ipfs/go-log"
	"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/lotus/chain"
	"github.com/filecoin-project/lotus/chain/store"
	"github.com/filecoin-project/lotus/chain/types"
	"github.com/filecoin-project/lotus/lib/cborutil"
	"github.com/filecoin-project/lotus/peermgr"
)

const ProtocolID = "/fil/hello/1.0.0"

var log = logging.Logger("hello")

func init() {
	cbor.RegisterCborType(Message{})
}

type Message struct {
	HeaviestTipSet       []cid.Cid
	HeaviestTipSetWeight types.BigInt
	GenesisHash          cid.Cid

	TArrial int64
	TSent   int64
}

type NewStreamFunc func(context.Context, peer.ID, ...protocol.ID) (inet.Stream, error)
type Service struct {
	newStream NewStreamFunc

	cs     *store.ChainStore
	syncer *chain.Syncer
	pmgr   *peermgr.PeerMgr
}

func NewHelloService(h host.Host, cs *store.ChainStore, syncer *chain.Syncer, pmgr peermgr.MaybePeerMgr) *Service {
	if pmgr.Mgr == nil {
		log.Warn("running without peer manager")
	}

	return &Service{
		newStream: h.NewStream,

		cs:     cs,
		syncer: syncer,
		pmgr:   pmgr.Mgr,
	}
}

func (hs *Service) HandleStream(s inet.Stream) {
	defer s.Close()

	var hmsg Message
	if err := cborutil.ReadCborRPC(s, &hmsg); err != nil {
		log.Infow("failed to read hello message", "error", err)
		return
	}
	arrived := time.Now()

	log.Debugw("genesis from hello",
		"tipset", hmsg.HeaviestTipSet,
		"peer", s.Conn().RemotePeer(),
		"hash", hmsg.GenesisHash)

	if hmsg.GenesisHash != hs.syncer.Genesis.Cids()[0] {
		log.Warnf("other peer has different genesis! (%s)", hmsg.GenesisHash)
		s.Conn().Close()
		return
	}
	go func() {
		sent := time.Now()
		msg := &Message{
			TArrial: arrived.UnixNano(),
			TSent:   sent.UnixNano(),
		}
		if err := cborutil.WriteCborRPC(s, msg); err != nil {
			log.Debugf("error while responding to latency: %v", err)
		}
	}()

	ts, err := hs.syncer.FetchTipSet(context.Background(), s.Conn().RemotePeer(), hmsg.HeaviestTipSet)
	if err != nil {
		log.Errorf("failed to fetch tipset from peer during hello: %s", err)
		return
	}

	log.Infof("Got new tipset through Hello: %s from %s", ts.Cids(), s.Conn().RemotePeer())
	hs.syncer.InformNewHead(s.Conn().RemotePeer(), ts)
	if hs.pmgr != nil {
		hs.pmgr.AddFilecoinPeer(s.Conn().RemotePeer())
	}

}

func (hs *Service) SayHello(ctx context.Context, pid peer.ID) error {
	s, err := hs.newStream(ctx, pid, ProtocolID)
	if err != nil {
		return err
	}
	defer s.Close()

	hts := hs.cs.GetHeaviestTipSet()
	weight, err := hs.cs.Weight(ctx, hts)
	if err != nil {
		return err
	}

	gen, err := hs.cs.GetGenesis()
	if err != nil {
		return err
	}

	hmsg := &Message{
		HeaviestTipSet:       hts.Cids(),
		HeaviestTipSetWeight: weight,
		GenesisHash:          gen.Cid(),
	}
	log.Info("Sending hello message: ", hts.Cids(), hts.Height(), gen.Cid())

	t0 := time.Now()
	if err := cborutil.WriteCborRPC(s, hmsg); err != nil {
		return err
	}

	go func() {
		hmsg = &Message{}
		s.SetReadDeadline(time.Now().Add(10 * time.Second))
		err := cborutil.ReadCborRPC(s, hmsg) // ignore error
		ok := err != nil

		t3 := time.Now()
		lat := t3.Sub(t0)
		// add to peer tracker
		if hs.pmgr != nil {
			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)
				offset := t0.Sub(t1) + t3.Sub(t2)
				offset /= 2
				log.Infow("time offset", "offset", offset.Seconds(), "peerid", pid.String())
			}
		}
	}()

	return nil
}