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

extract blocksync into its own package

Browse Source
This commit is contained in:
whyrusleeping 2019-11-09 15:00:22 -08:00
parent bda6d7e119
commit 904fdad4c6
10 changed files with 95 additions and 79 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

5
chain/blocksync.go → chain/blocksync/blocksync.go
View File

@ -1,4 +1,4 @@
package chain
package blocksync
import (
"bufio"
@ -21,10 +21,13 @@ import (
blocks "github.com/ipfs/go-block-format"
"github.com/ipfs/go-cid"
cbor "github.com/ipfs/go-ipld-cbor"
logging "github.com/ipfs/go-log"
inet "github.com/libp2p/go-libp2p-core/network"
"github.com/libp2p/go-libp2p-core/peer"
)
var log = logging.Logger("blocksync")
type NewStreamFunc func(context.Context, peer.ID, ...protocol.ID) (inet.Stream, error)
const BlockSyncProtocolID = "/fil/sync/blk/0.0.1"

122
chain/cbor_gen.go → chain/blocksync/cbor_gen.go
View File

@ -1,11 +1,11 @@
package chain
package blocksync
import (
"fmt"
"io"
"github.com/filecoin-project/lotus/chain/types"
cid "github.com/ipfs/go-cid"
"github.com/ipfs/go-cid"
cbg "github.com/whyrusleeping/cbor-gen"
xerrors "golang.org/x/xerrors"
)
@ -15,11 +15,15 @@ import (
var _ = xerrors.Errorf
func (t *BlockSyncRequest) MarshalCBOR(w io.Writer) error {
if t == nil {
_, err := w.Write(cbg.CborNull)
return err
}
if _, err := w.Write([]byte{131}); err != nil {
return err
}
// t.t.Start ([]cid.Cid)
// t.t.Start ([]cid.Cid) (slice)
if _, err := w.Write(cbg.CborEncodeMajorType(cbg.MajArray, uint64(len(t.Start)))); err != nil {
return err
}
@ -29,19 +33,20 @@ func (t *BlockSyncRequest) MarshalCBOR(w io.Writer) error {
}
}
// t.t.RequestLength (uint64)
if _, err := w.Write(cbg.CborEncodeMajorType(cbg.MajUnsignedInt, t.RequestLength)); err != nil {
// t.t.RequestLength (uint64) (uint64)
if _, err := w.Write(cbg.CborEncodeMajorType(cbg.MajUnsignedInt, uint64(t.RequestLength))); err != nil {
return err
}
// t.t.Options (uint64)
if _, err := w.Write(cbg.CborEncodeMajorType(cbg.MajUnsignedInt, t.Options)); err != nil {
// t.t.Options (uint64) (uint64)
if _, err := w.Write(cbg.CborEncodeMajorType(cbg.MajUnsignedInt, uint64(t.Options))); err != nil {
return err
}
return nil
}
func (t *BlockSyncRequest) UnmarshalCBOR(br io.Reader) error {
func (t *BlockSyncRequest) UnmarshalCBOR(r io.Reader) error {
br := cbg.GetPeeker(r)
maj, extra, err := cbg.CborReadHeader(br)
if err != nil {
@ -55,14 +60,14 @@ func (t *BlockSyncRequest) UnmarshalCBOR(br io.Reader) error {
return fmt.Errorf("cbor input had wrong number of fields")
}
// t.t.Start ([]cid.Cid)
// t.t.Start ([]cid.Cid) (slice)
maj, extra, err = cbg.CborReadHeader(br)
if err != nil {
return err
}
if extra > 8192 {
return fmt.Errorf("array too large")
return fmt.Errorf("t.Start: array too large (%d)", extra)
}
if maj != cbg.MajArray {
@ -80,7 +85,7 @@ func (t *BlockSyncRequest) UnmarshalCBOR(br io.Reader) error {
t.Start[i] = c
}
// t.t.RequestLength (uint64)
// t.t.RequestLength (uint64) (uint64)
maj, extra, err = cbg.CborReadHeader(br)
if err != nil {
@ -89,8 +94,8 @@ func (t *BlockSyncRequest) UnmarshalCBOR(br io.Reader) error {
if maj != cbg.MajUnsignedInt {
return fmt.Errorf("wrong type for uint64 field")
}
t.RequestLength = extra
// t.t.Options (uint64)
t.RequestLength = uint64(extra)
// t.t.Options (uint64) (uint64)
maj, extra, err = cbg.CborReadHeader(br)
if err != nil {
@ -99,16 +104,20 @@ func (t *BlockSyncRequest) UnmarshalCBOR(br io.Reader) error {
if maj != cbg.MajUnsignedInt {
return fmt.Errorf("wrong type for uint64 field")
}
t.Options = extra
t.Options = uint64(extra)
return nil
}
func (t *BlockSyncResponse) MarshalCBOR(w io.Writer) error {
if t == nil {
_, err := w.Write(cbg.CborNull)
return err
}
if _, err := w.Write([]byte{131}); err != nil {
return err
}
// t.t.Chain ([]*chain.BSTipSet)
// t.t.Chain ([]*blocksync.BSTipSet) (slice)
if _, err := w.Write(cbg.CborEncodeMajorType(cbg.MajArray, uint64(len(t.Chain)))); err != nil {
return err
}
@ -118,12 +127,12 @@ func (t *BlockSyncResponse) MarshalCBOR(w io.Writer) error {
}
}
// t.t.Status (uint64)
if _, err := w.Write(cbg.CborEncodeMajorType(cbg.MajUnsignedInt, t.Status)); err != nil {
// t.t.Status (uint64) (uint64)
if _, err := w.Write(cbg.CborEncodeMajorType(cbg.MajUnsignedInt, uint64(t.Status))); err != nil {
return err
}
// t.t.Message (string)
// t.t.Message (string) (string)
if _, err := w.Write(cbg.CborEncodeMajorType(cbg.MajTextString, uint64(len(t.Message)))); err != nil {
return err
}
@ -133,7 +142,8 @@ func (t *BlockSyncResponse) MarshalCBOR(w io.Writer) error {
return nil
}
func (t *BlockSyncResponse) UnmarshalCBOR(br io.Reader) error {
func (t *BlockSyncResponse) UnmarshalCBOR(r io.Reader) error {
br := cbg.GetPeeker(r)
maj, extra, err := cbg.CborReadHeader(br)
if err != nil {
@ -147,14 +157,14 @@ func (t *BlockSyncResponse) UnmarshalCBOR(br io.Reader) error {
return fmt.Errorf("cbor input had wrong number of fields")
}
// t.t.Chain ([]*chain.BSTipSet)
// t.t.Chain ([]*blocksync.BSTipSet) (slice)
maj, extra, err = cbg.CborReadHeader(br)
if err != nil {
return err
}
if extra > 8192 {
return fmt.Errorf("array too large")
return fmt.Errorf("t.Chain: array too large (%d)", extra)
}
if maj != cbg.MajArray {
@ -164,6 +174,7 @@ func (t *BlockSyncResponse) UnmarshalCBOR(br io.Reader) error {
t.Chain = make([]*BSTipSet, extra)
}
for i := 0; i < int(extra); i++ {
var v BSTipSet
if err := v.UnmarshalCBOR(br); err != nil {
return err
@ -172,7 +183,7 @@ func (t *BlockSyncResponse) UnmarshalCBOR(br io.Reader) error {
t.Chain[i] = &v
}
// t.t.Status (uint64)
// t.t.Status (uint64) (uint64)
maj, extra, err = cbg.CborReadHeader(br)
if err != nil {
@ -181,39 +192,30 @@ func (t *BlockSyncResponse) UnmarshalCBOR(br io.Reader) error {
if maj != cbg.MajUnsignedInt {
return fmt.Errorf("wrong type for uint64 field")
}
t.Status = extra
// t.t.Message (string)
maj, extra, err = cbg.CborReadHeader(br)
if err != nil {
return err
}
if maj != cbg.MajTextString {
return fmt.Errorf("expected cbor type 'text string' in input")
}
if extra > 256*1024 {
return fmt.Errorf("string in cbor input too long")
}
t.Status = uint64(extra)
// t.t.Message (string) (string)
{
buf := make([]byte, extra)
if _, err := io.ReadFull(br, buf); err != nil {
sval, err := cbg.ReadString(br)
if err != nil {
return err
}
t.Message = string(buf)
t.Message = string(sval)
}
return nil
}
func (t *BSTipSet) MarshalCBOR(w io.Writer) error {
if t == nil {
_, err := w.Write(cbg.CborNull)
return err
}
if _, err := w.Write([]byte{133}); err != nil {
return err
}
// t.t.Blocks ([]*types.BlockHeader)
// t.t.Blocks ([]*types.BlockHeader) (slice)
if _, err := w.Write(cbg.CborEncodeMajorType(cbg.MajArray, uint64(len(t.Blocks)))); err != nil {
return err
}
@ -223,7 +225,7 @@ func (t *BSTipSet) MarshalCBOR(w io.Writer) error {
}
}
// t.t.BlsMessages ([]*types.Message)
// t.t.BlsMessages ([]*types.Message) (slice)
if _, err := w.Write(cbg.CborEncodeMajorType(cbg.MajArray, uint64(len(t.BlsMessages)))); err != nil {
return err
}
@ -233,7 +235,7 @@ func (t *BSTipSet) MarshalCBOR(w io.Writer) error {
}
}
// t.t.BlsMsgIncludes ([][]uint64)
// t.t.BlsMsgIncludes ([][]uint64) (slice)
if _, err := w.Write(cbg.CborEncodeMajorType(cbg.MajArray, uint64(len(t.BlsMsgIncludes)))); err != nil {
return err
}
@ -248,7 +250,7 @@ func (t *BSTipSet) MarshalCBOR(w io.Writer) error {
}
}
// t.t.SecpkMessages ([]*types.SignedMessage)
// t.t.SecpkMessages ([]*types.SignedMessage) (slice)
if _, err := w.Write(cbg.CborEncodeMajorType(cbg.MajArray, uint64(len(t.SecpkMessages)))); err != nil {
return err
}
@ -258,7 +260,7 @@ func (t *BSTipSet) MarshalCBOR(w io.Writer) error {
}
}
// t.t.SecpkMsgIncludes ([][]uint64)
// t.t.SecpkMsgIncludes ([][]uint64) (slice)
if _, err := w.Write(cbg.CborEncodeMajorType(cbg.MajArray, uint64(len(t.SecpkMsgIncludes)))); err != nil {
return err
}
@ -275,7 +277,8 @@ func (t *BSTipSet) MarshalCBOR(w io.Writer) error {
return nil
}
func (t *BSTipSet) UnmarshalCBOR(br io.Reader) error {
func (t *BSTipSet) UnmarshalCBOR(r io.Reader) error {
br := cbg.GetPeeker(r)
maj, extra, err := cbg.CborReadHeader(br)
if err != nil {
@ -289,14 +292,14 @@ func (t *BSTipSet) UnmarshalCBOR(br io.Reader) error {
return fmt.Errorf("cbor input had wrong number of fields")
}
// t.t.Blocks ([]*types.BlockHeader)
// t.t.Blocks ([]*types.BlockHeader) (slice)
maj, extra, err = cbg.CborReadHeader(br)
if err != nil {
return err
}
if extra > 8192 {
return fmt.Errorf("array too large")
return fmt.Errorf("t.Blocks: array too large (%d)", extra)
}
if maj != cbg.MajArray {
@ -306,6 +309,7 @@ func (t *BSTipSet) UnmarshalCBOR(br io.Reader) error {
t.Blocks = make([]*types.BlockHeader, extra)
}
for i := 0; i < int(extra); i++ {
var v types.BlockHeader
if err := v.UnmarshalCBOR(br); err != nil {
return err
@ -314,14 +318,14 @@ func (t *BSTipSet) UnmarshalCBOR(br io.Reader) error {
t.Blocks[i] = &v
}
// t.t.BlsMessages ([]*types.Message)
// t.t.BlsMessages ([]*types.Message) (slice)
maj, extra, err = cbg.CborReadHeader(br)
if err != nil {
return err
}
if extra > 8192 {
return fmt.Errorf("array too large")
return fmt.Errorf("t.BlsMessages: array too large (%d)", extra)
}
if maj != cbg.MajArray {
@ -331,6 +335,7 @@ func (t *BSTipSet) UnmarshalCBOR(br io.Reader) error {
t.BlsMessages = make([]*types.Message, extra)
}
for i := 0; i < int(extra); i++ {
var v types.Message
if err := v.UnmarshalCBOR(br); err != nil {
return err
@ -339,14 +344,14 @@ func (t *BSTipSet) UnmarshalCBOR(br io.Reader) error {
t.BlsMessages[i] = &v
}
// t.t.BlsMsgIncludes ([][]uint64)
// t.t.BlsMsgIncludes ([][]uint64) (slice)
maj, extra, err = cbg.CborReadHeader(br)
if err != nil {
return err
}
if extra > 8192 {
return fmt.Errorf("array too large")
return fmt.Errorf("t.BlsMsgIncludes: array too large (%d)", extra)
}
if maj != cbg.MajArray {
@ -366,7 +371,7 @@ func (t *BSTipSet) UnmarshalCBOR(br io.Reader) error {
return err
}
if extra > 8192 {
return fmt.Errorf("array too large")
return fmt.Errorf("t.BlsMsgIncludes[i]: array too large (%d)", extra)
}
if maj != cbg.MajArray {
@ -392,14 +397,14 @@ func (t *BSTipSet) UnmarshalCBOR(br io.Reader) error {
}
}
// t.t.SecpkMessages ([]*types.SignedMessage)
// t.t.SecpkMessages ([]*types.SignedMessage) (slice)
maj, extra, err = cbg.CborReadHeader(br)
if err != nil {
return err
}
if extra > 8192 {
return fmt.Errorf("array too large")
return fmt.Errorf("t.SecpkMessages: array too large (%d)", extra)
}
if maj != cbg.MajArray {
@ -409,6 +414,7 @@ func (t *BSTipSet) UnmarshalCBOR(br io.Reader) error {
t.SecpkMessages = make([]*types.SignedMessage, extra)
}
for i := 0; i < int(extra); i++ {
var v types.SignedMessage
if err := v.UnmarshalCBOR(br); err != nil {
return err
@ -417,14 +423,14 @@ func (t *BSTipSet) UnmarshalCBOR(br io.Reader) error {
t.SecpkMessages[i] = &v
}
// t.t.SecpkMsgIncludes ([][]uint64)
// t.t.SecpkMsgIncludes ([][]uint64) (slice)
maj, extra, err = cbg.CborReadHeader(br)
if err != nil {
return err
}
if extra > 8192 {
return fmt.Errorf("array too large")
return fmt.Errorf("t.SecpkMsgIncludes: array too large (%d)", extra)
}
if maj != cbg.MajArray {
@ -444,7 +450,7 @@ func (t *BSTipSet) UnmarshalCBOR(br io.Reader) error {
return err
}
if extra > 8192 {
return fmt.Errorf("array too large")
return fmt.Errorf("t.SecpkMsgIncludes[i]: array too large (%d)", extra)
}
if maj != cbg.MajArray {

7
chain/sync.go
View File

@ -12,6 +12,7 @@ import (
"github.com/filecoin-project/lotus/build"
"github.com/filecoin-project/lotus/chain/actors"
"github.com/filecoin-project/lotus/chain/address"
"github.com/filecoin-project/lotus/chain/blocksync"
"github.com/filecoin-project/lotus/chain/state"
"github.com/filecoin-project/lotus/chain/stmgr"
"github.com/filecoin-project/lotus/chain/store"
@ -49,7 +50,7 @@ type Syncer struct {
bad *BadBlockCache
// handle to the block sync service
Bsync *BlockSync
Bsync *blocksync.BlockSync
self peer.ID
@ -61,7 +62,7 @@ type Syncer struct {
peerHeadsLk sync.Mutex
}
func NewSyncer(sm *stmgr.StateManager, bsync *BlockSync, self peer.ID) (*Syncer, error) {
func NewSyncer(sm *stmgr.StateManager, bsync *blocksync.BlockSync, self peer.ID) (*Syncer, error) {
gen, err := sm.ChainStore().GetGenesis()
if err != nil {
return nil, err
@ -882,7 +883,7 @@ func (syncer *Syncer) iterFullTipsets(ctx context.Context, headers []*types.TipS
return nil
}
func persistMessages(bs bstore.Blockstore, bst *BSTipSet) error {
func persistMessages(bs bstore.Blockstore, bst *blocksync.BSTipSet) error {
for _, m := range bst.BlsMessages {
//log.Infof("putting BLS message: %s", m.Cid())
if _, err := store.PutMessage(bs, m); err != nil {

2
chain/types/cbor_gen.go
View File

@ -5,7 +5,7 @@ import (
"io"
"math"
cid "github.com/ipfs/go-cid"
"github.com/ipfs/go-cid"
cbg "github.com/whyrusleeping/cbor-gen"
xerrors "golang.org/x/xerrors"
)

21
gen/main.go
View File

@ -8,6 +8,7 @@ import (
"github.com/filecoin-project/lotus/api"
"github.com/filecoin-project/lotus/chain/actors"
"github.com/filecoin-project/lotus/chain/blocksync"
"github.com/filecoin-project/lotus/chain/deals"
"github.com/filecoin-project/lotus/chain/types"
"github.com/filecoin-project/lotus/paych"
@ -71,17 +72,15 @@ func main() {
os.Exit(1)
}
/*
err = gen.WriteTupleEncodersToFile("./chain/cbor_gen.go", "chain",
chain.BlockSyncRequest{},
chain.BlockSyncResponse{},
chain.BSTipSet{},
)
if err != nil {
fmt.Println(err)
os.Exit(1)
}
*/
err = gen.WriteTupleEncodersToFile("./chain/blocksync/cbor_gen.go", "blocksync",
blocksync.BlockSyncRequest{},
blocksync.BlockSyncResponse{},
blocksync.BSTipSet{},
)
if err != nil {
fmt.Println(err)
os.Exit(1)
}
err = gen.WriteTupleEncodersToFile("./chain/actors/cbor_gen.go", "actors",
actors.InitActorState{},

1
go.mod
View File

@ -50,6 +50,7 @@ require (
github.com/libp2p/go-libp2p-peer v0.2.0
github.com/libp2p/go-libp2p-peerstore v0.1.3
github.com/libp2p/go-libp2p-pnet v0.1.0
github.com/libp2p/go-libp2p-protocol v0.1.0
github.com/libp2p/go-libp2p-pubsub v0.1.0
github.com/libp2p/go-libp2p-quic-transport v0.1.1
github.com/libp2p/go-libp2p-record v0.1.1

2
go.sum
View File

@ -325,6 +325,8 @@ github.com/libp2p/go-libp2p-peerstore v0.1.3 h1:wMgajt1uM2tMiqf4M+4qWKVyyFc8SfA+
github.com/libp2p/go-libp2p-peerstore v0.1.3/go.mod h1:BJ9sHlm59/80oSkpWgr1MyY1ciXAXV397W6h1GH/uKI=
github.com/libp2p/go-libp2p-pnet v0.1.0 h1:kRUES28dktfnHNIRW4Ro78F7rKBHBiw5MJpl0ikrLIA=
github.com/libp2p/go-libp2p-pnet v0.1.0/go.mod h1:ZkyZw3d0ZFOex71halXRihWf9WH/j3OevcJdTmD0lyE=
github.com/libp2p/go-libp2p-protocol v0.1.0 h1:HdqhEyhg0ToCaxgMhnOmUO8snQtt/kQlcjVk3UoJU3c=
github.com/libp2p/go-libp2p-protocol v0.1.0/go.mod h1:KQPHpAabB57XQxGrXCNvbL6UEXfQqUgC/1adR2Xtflk=
github.com/libp2p/go-libp2p-pubsub v0.1.0 h1:SmQeMa7IUv5vadh0fYgYsafWCBA1sCy5d/68kIYqGcU=
github.com/libp2p/go-libp2p-pubsub v0.1.0/go.mod h1:ZwlKzRSe1eGvSIdU5bD7+8RZN/Uzw0t1Bp9R1znpR/Q=
github.com/libp2p/go-libp2p-quic-transport v0.1.1 h1:MFMJzvsxIEDEVKzO89BnB/FgvMj9WI4GDGUW2ArDPUA=

5
node/builder.go
View File

@ -19,6 +19,7 @@ import (
"github.com/filecoin-project/lotus/api"
"github.com/filecoin-project/lotus/chain"
"github.com/filecoin-project/lotus/chain/blocksync"
"github.com/filecoin-project/lotus/chain/deals"
"github.com/filecoin-project/lotus/chain/market"
"github.com/filecoin-project/lotus/chain/metrics"
@ -197,14 +198,14 @@ func Online() Option {
// Filecoin services
Override(new(*chain.Syncer), chain.NewSyncer),
Override(new(*chain.BlockSync), chain.NewBlockSyncClient),
Override(new(*blocksync.BlockSync), blocksync.NewBlockSyncClient),
Override(new(*chain.MessagePool), chain.NewMessagePool),
Override(new(modules.Genesis), modules.ErrorGenesis),
Override(SetGenesisKey, modules.SetGenesis),
Override(new(*hello.Service), hello.NewHelloService),
Override(new(*chain.BlockSyncService), chain.NewBlockSyncService),
Override(new(*blocksync.BlockSyncService), blocksync.NewBlockSyncService),
Override(new(*peermgr.PeerMgr), peermgr.NewPeerMgr),
Override(RunHelloKey, modules.RunHello),

4
node/hello/hello.go
View File

@ -12,6 +12,7 @@ import (
"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-protocol"
"github.com/filecoin-project/lotus/chain"
"github.com/filecoin-project/lotus/chain/store"
@ -34,8 +35,9 @@ type Message struct {
GenesisHash cid.Cid
}
type NewStreamFunc func(context.Context, peer.ID, ...protocol.ID) (inet.Stream, error)
type Service struct {
newStream chain.NewStreamFunc
newStream NewStreamFunc
cs *store.ChainStore
syncer *chain.Syncer

5
node/modules/services.go
View File

@ -9,6 +9,7 @@ import (
"go.uber.org/fx"
"github.com/filecoin-project/lotus/chain"
"github.com/filecoin-project/lotus/chain/blocksync"
"github.com/filecoin-project/lotus/chain/deals"
"github.com/filecoin-project/lotus/chain/sub"
"github.com/filecoin-project/lotus/node/hello"
@ -37,8 +38,8 @@ func RunPeerMgr(mctx helpers.MetricsCtx, lc fx.Lifecycle, pmgr *peermgr.PeerMgr)
go pmgr.Run(helpers.LifecycleCtx(mctx, lc))
}
func RunBlockSync(h host.Host, svc *chain.BlockSyncService) {
h.SetStreamHandler(chain.BlockSyncProtocolID, svc.HandleStream)
func RunBlockSync(h host.Host, svc *blocksync.BlockSyncService) {
h.SetStreamHandler(blocksync.BlockSyncProtocolID, svc.HandleStream)
}
func HandleIncomingBlocks(mctx helpers.MetricsCtx, lc fx.Lifecycle, pubsub *pubsub.PubSub, s *chain.Syncer) {