Local networked P2P! #222
Local networked P2P! #222
Conversation
prestonvanloon
requested review from
rauljordan,
terencechain,
rawfalafel and
nisdas
and removed request for
rauljordan
sharding/p2p/discovery.go
Outdated
| } | ||
|
|
||
| // topicPeerLister has a method to return connected peers on a given topic. | ||
| // This is implemented by floodsub.PubSub |
sharding/p2p/discovery.go
Outdated
| func (d *discovery) HandlePeerFound(pi ps.PeerInfo) { | ||
| d.host.Peerstore().AddAddrs(pi.ID, pi.Addrs, ps.PermanentAddrTTL) | ||
| if err := d.host.Connect(d.ctx, pi); err != nil { | ||
| log.Warn(fmt.Sprintf("Failed to connect to peer: %v", err)) |
sharding/p2p/discovery.go
Outdated
| log.Warn(fmt.Sprintf("Failed to connect to peer: %v", err)) | ||
| } | ||
|
|
||
| log.Debug(fmt.Sprintf("Peers now: %s", d.host.Peerstore().Peers())) |
sharding/p2p/proto/v1/messages.proto
Outdated
| @@ -0,0 +1,39 @@ | |||
| syntax = "proto3"; | |||
There was a problem hiding this comment.
Should we put this stuff inside of shared/p2p/proto?
sharding/p2p/service.go
Outdated
| }, nil | ||
| } | ||
|
|
||
| // Start the main routine for an p2p server. | ||
| func (s *Server) Start() { | ||
| log.Info("Starting shardp2p server") | ||
| if err := startDiscovery(s.ctx, s.host, s.gsub); err != nil { | ||
| log.Error(fmt.Sprintf("Could not start p2p discovery! %v", err)) |
sharding/proposer/service.go
Outdated
| @@ -76,18 +78,36 @@ func (p *Proposer) Stop() error { | |||
| return nil | |||
| } | |||
|
|
|||
| // TODO: Move this somewhere else | |||
There was a problem hiding this comment.
Let's put this in the utils package.
sharding/simulator/service.go
Outdated
| @@ -79,18 +78,18 @@ func (s *Simulator) simulateNotaryRequests(fetcher mainchain.RecordFetcher, read | |||
| } | |||
|
|
|||
| period := new(big.Int).Div(blockNumber.Number(), big.NewInt(s.config.PeriodLength)) | |||
| period = period.Sub(period, big.NewInt(1)) | |||
There was a problem hiding this comment.
Add a comment explaining why we're doing this
| s.p2p.Broadcast(req) | ||
| log.Debug("Sent request for collation body via a shardp2p broadcast") | ||
| } else { | ||
| log.Warn("Syncer generated nil CollationBodyRequest") |
There was a problem hiding this comment.
This is already captured by the log right before it that says "no collation found for shardid x, period y"
There was a problem hiding this comment.
I don't see that message?
There was a problem hiding this comment.
syncer/handlers.go log.Debugf("No collation exists for shard")
There was a problem hiding this comment.
I see, but from this context when we ask the syncer to generate a collation request and it gives us nothing back, something feels wrong. Without debug logging the simulator appears to do nothing.
We'll need to iterate on this to make it better.
sharding/simulator/service.go
Outdated
|
|
||
| s.p2p.Broadcast(tx) | ||
| log.Debug("Transaction broadcasted") | ||
| // s.p2p.Broadcast(messages.TransactionBroadcast{Transaction: tx}) |
There was a problem hiding this comment.
Can we just delete the comments beneath it?
| res, err := RespondCollationBody(req, collationFetcher) | ||
| if err != nil { | ||
| log.Errorf("Could not construct response: %v", err) | ||
| continue | ||
| } | ||
|
|
||
| if res == nil { | ||
| // TODO: Send that we don't have it? |
There was a problem hiding this comment.
Agree we should send a msg for this
|
Also, in order to try this out, we need to start a proposer first and then a simulator, as only one of these actors can access the local leveldb instance. Additionally, the simulator lags behind the proposer in terms of requesting collation data, as they are never synchronized on the same period for these requests. Just a few things to keep in mind @prestonvanloon. |
|
I addressed some concerns. I’ll move the protobuf definition to a better place, tomorrow. |
| // startDiscovery protocols. Currently, this supports discovery via multicast | ||
| // DNS peer discovery. | ||
| // | ||
| // TODO: add other discovery protocols such as DHT, etc. |
There was a problem hiding this comment.
Let's open up an issue for add other discovery protocols
| func (d *discovery) HandlePeerFound(pi ps.PeerInfo) { | ||
| d.host.Peerstore().AddAddrs(pi.ID, pi.Addrs, ps.PermanentAddrTTL) | ||
| if err := d.host.Connect(d.ctx, pi); err != nil { | ||
| log.Warnf("Failed to connect to peer: %v", err) |
There was a problem hiding this comment.
Wouldn't error be more appropriate?
Warning is like "you should probably take a look at this"
Error is like "something failed but I'm not quitting"
There was a problem hiding this comment.
I'm not sure it's really error level. Imagine that they discover a peer but then the peer goes offline or refuses the connection.
In geth, these are debug level logs so warning might even be too verbose.
sharding/p2p/discovery_test.go
Outdated
| func expectPeers(t *testing.T, h *bhost.BasicHost, n int) { | ||
| if len(h.Peerstore().Peers()) != n { | ||
| t.Errorf( | ||
| "Expected 2 peer for host %v, but has %d peers. They are: %v.", |
There was a problem hiding this comment.
instead of Expected 2 peers , should be Expected %d peers , with the argument being n
| var portRange int32 = 100 | ||
|
|
||
| // buildOptions for the libp2p host. | ||
| // TODO: Expand on these options and provide the option configuration via flags. |
Former-commit-id: 284a046
Depends on #221 (vendor libraries) for a smaller diff.
Preliminary p2p services
The p2p functionality here is rather simple, but is a means to test (locally) networked sharding nodes.
Key features:
Send"works" by broadcasting your message to all peers. Technically your intended receiver will receive the message, but so will everyone else for the time being.