forks/go-ethereum
1
0
Fork 1
mirror of https://github.com/ethereum/go-ethereum.git synced 2026-08-19 10:22:23 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions

swarm, swarm/network/stream, swarm/netork/simulations,, swarm/pss: adapt to new Kad API

Browse source
This commit is contained in:
zelig 2018-09-06 13:13:31 +02:00
parent e748d370ac
commit b7286a9909
8 changed files with 53 additions and 85 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

4
swarm/network/simulations/discovery/discovery_test.go
View file

@ -556,8 +556,8 @@ func newService(ctx *adapters.ServiceContext) (node.Service, error) {
kp.MinProxBinSize = testMinProxBinSize
if ctx.Config.Reachable != nil {
kp.Reachable = func(o network.OverlayAddr) bool {
return ctx.Config.Reachable(o.(*network.BzzAddr).ID())
kp.Reachable = func(o *network.BzzAddr) bool {
return ctx.Config.Reachable(o.ID())
}
}
kad := network.NewKademlia(addr.Over(), kp)

12
swarm/network/stream/delivery.go
View file

@ -47,15 +47,15 @@ var (
type Delivery struct {
db *storage.DBAPI
overlay network.Overlay
kad *network.Kademlia
receiveC chan *ChunkDeliveryMsg
getPeer func(discover.NodeID) *Peer
}
func NewDelivery(overlay network.Overlay, db *storage.DBAPI) *Delivery {
func NewDelivery(kad *network.Kademlia, db *storage.DBAPI) *Delivery {
d := &Delivery{
db: db,
overlay: overlay,
kad: kad,
receiveC: make(chan *ChunkDeliveryMsg, deliveryCap),
}
@ -172,7 +172,7 @@ func (d *Delivery) handleRetrieveRequestMsg(ctx context.Context, sp *Peer, req *
t := time.NewTimer(10 * time.Minute)
defer t.Stop()
log.Debug("waiting delivery", "peer", sp.ID(), "hash", req.Addr, "node", common.Bytes2Hex(d.overlay.BaseAddr()), "created", created)
log.Debug("waiting delivery", "peer", sp.ID(), "hash", req.Addr, "node", common.Bytes2Hex(d.kad.BaseAddr()), "created", created)
start := time.Now()
select {
case <-chunk.ReqC:
@ -269,8 +269,8 @@ func (d *Delivery) RequestFromPeers(ctx context.Context, hash []byte, skipCheck
var err error
requestFromPeersCount.Inc(1)
d.overlay.EachConn(hash, 255, func(p network.OverlayConn, po int, nn bool) bool {
spId := p.(network.Peer).ID()
d.kad.EachConn(hash, 255, func(p *network.Peer, po int, nn bool) bool {
spId := p.ID()
for _, p := range peersToSkip {
if p == spId {
log.Trace("Delivery.RequestFromPeers: skip peer", "peer", spId)

9
swarm/network/stream/snapshot_sync_test.go
View file

@ -457,15 +457,10 @@ func testSyncingViaDirectSubscribe(chunkCount int, nodeCount int) error {
//returns the number of subscriptions requested
func startSyncing(r *Registry, conf *synctestConfig) (int, error) {
var err error
kad, ok := r.delivery.overlay.(*network.Kademlia)
if !ok {
return 0, fmt.Errorf("Not a Kademlia!")
}
kad := r.delivery.kad
subCnt := 0
//iterate over each bin and solicit needed subscription to bins
kad.EachBin(r.addr.Over(), pof, 0, func(conn network.OverlayConn, po int) bool {
kad.EachBin(r.addr.Over(), pof, 0, func(conn *network.Peer, po int) bool {
//identify begin and start index of the bin(s) we want to subscribe to
histRange := &Range{}

18
swarm/network/stream/stream.go
View file

@ -130,7 +130,7 @@ func NewRegistry(addr *network.BzzAddr, delivery *Delivery, db *storage.DBAPI, i
// wait for kademlia table to be healthy
time.Sleep(options.SyncUpdateDelay)
kad := streamer.delivery.overlay.(*network.Kademlia)
kad := streamer.delivery.kad
depthC := latestIntC(kad.NeighbourhoodDepthC())
addressBookSizeC := latestIntC(kad.AddrCountC())
@ -398,9 +398,7 @@ func (r *Registry) Run(p *network.BzzPeer) error {
// and they are no longer required after iteration, request to Quit
// them will be send to appropriate peers.
func (r *Registry) updateSyncing() {
// if overlay in not Kademlia, panic
kad := r.delivery.overlay.(*network.Kademlia)
kad := r.delivery.kad
// map of all SYNC streams for all peers
// used at the and of the function to remove servers
// that are not needed anymore
@ -421,8 +419,7 @@ func (r *Registry) updateSyncing() {
r.peersMu.RUnlock()
// request subscriptions for all nodes and bins
kad.EachBin(r.addr.Over(), pot.DefaultPof(256), 0, func(conn network.OverlayConn, bin int) bool {
p := conn.(network.Peer)
kad.EachBin(r.addr.Over(), pot.DefaultPof(256), 0, func(p *network.Peer, bin int) bool {
log.Debug(fmt.Sprintf("Requesting subscription by: registry %s from peer %s for bin: %d", r.addr.ID(), p.ID(), bin))
// bin is always less then 256 and it is safe to convert it to type uint8
@ -461,10 +458,11 @@ func (r *Registry) updateSyncing() {
func (r *Registry) runProtocol(p *p2p.Peer, rw p2p.MsgReadWriter) error {
peer := protocols.NewPeer(p, rw, Spec)
bzzPeer := network.NewBzzTestPeer(peer, r.addr)
r.delivery.overlay.On(bzzPeer)
defer r.delivery.overlay.Off(bzzPeer)
return r.Run(bzzPeer)
bp := network.NewBzzTestPeer(peer, r.addr)
np := network.NewPeer(bp, r.delivery.kad)
r.delivery.kad.On(np)
defer r.delivery.kad.Off(np)
return r.Run(bp)
}
// HandleMsg is the message handler that delegates incoming messages

3
swarm/network_test.go
View file

@ -34,7 +34,6 @@ import (
"github.com/ethereum/go-ethereum/p2p/discover"
"github.com/ethereum/go-ethereum/p2p/simulations/adapters"
"github.com/ethereum/go-ethereum/swarm/api"
"github.com/ethereum/go-ethereum/swarm/network"
"github.com/ethereum/go-ethereum/swarm/network/simulation"
"github.com/ethereum/go-ethereum/swarm/storage"
colorable "github.com/mattn/go-colorable"
@ -293,7 +292,7 @@ func testSwarmNetwork(t *testing.T, o *testSwarmNetworkOptions, steps ...testSwa
if err != nil {
return nil, cleanup, err
}
bucket.Store(simulation.BucketKeyKademlia, swarm.bzz.Hive.Overlay.(*network.Kademlia))
bucket.Store(simulation.BucketKeyKademlia, swarm.bzz.Hive.Kademlia)
log.Info("new swarm", "bzzKey", config.BzzKey, "baseAddr", fmt.Sprintf("%x", swarm.bzz.BaseAddr()))
return swarm, cleanup, nil
},

43
swarm/pss/pss.go
View file

@ -110,10 +110,10 @@ func (params *PssParams) WithPrivateKey(privatekey *ecdsa.PrivateKey) *PssParams
//
// Implements node.Service
type Pss struct {
network.Overlay // we can get the overlayaddress from this
privateKey *ecdsa.PrivateKey // pss can have it's own independent key
w *whisper.Whisper // key and encryption backend
auxAPIs []rpc.API // builtins (handshake, test) can add APIs
*network.Kademlia // we can get the Kademliaaddress from this
privateKey *ecdsa.PrivateKey // pss can have it's own independent key
w *whisper.Whisper // key and encryption backend
auxAPIs []rpc.API // builtins (handshake, test) can add APIs
// sending and forwarding
fwdPool map[string]*protocols.Peer // keep track of all peers sitting on the pssmsg routing layer
@ -151,9 +151,9 @@ func (p *Pss) String() string {
// Creates a new Pss instance.
//
// In addition to params, it takes a swarm network overlay
// In addition to params, it takes a swarm network Kademlia
// and a FileStore storage for message cache storage.
func NewPss(k network.Overlay, params *PssParams) (*Pss, error) {
func NewPss(k *network.Kademlia, params *PssParams) (*Pss, error) {
if params.privateKey == nil {
return nil, errors.New("missing private key for pss")
}
@ -162,7 +162,7 @@ func NewPss(k network.Overlay, params *PssParams) (*Pss, error) {
Version: pssVersion,
}
ps := &Pss{
Overlay: k,
Kademlia: k,
privateKey: params.privateKey,
w: whisper.New(&whisper.DefaultConfig),
quitC: make(chan struct{}),
@ -290,9 +290,9 @@ func (p *Pss) addAPI(api rpc.API) {
p.auxAPIs = append(p.auxAPIs, api)
}
// Returns the swarm overlay address of the pss node
// Returns the swarm Kademlia address of the pss node
func (p *Pss) BaseAddr() []byte {
return p.Overlay.BaseAddr()
return p.Kademlia.BaseAddr()
}
// Returns the pss node's public key
@ -356,11 +356,11 @@ func (p *Pss) handlePssMsg(ctx context.Context, msg interface{}) error {
}
if int64(pssmsg.Expire) < time.Now().Unix() {
metrics.GetOrRegisterCounter("pss.expire", nil).Inc(1)
log.Warn("pss filtered expired message", "from", common.ToHex(p.Overlay.BaseAddr()), "to", common.ToHex(pssmsg.To))
log.Warn("pss filtered expired message", "from", common.ToHex(p.Kademlia.BaseAddr()), "to", common.ToHex(pssmsg.To))
return nil
}
if p.checkFwdCache(pssmsg) {
log.Trace("pss relay block-cache match (process)", "from", common.ToHex(p.Overlay.BaseAddr()), "to", (common.ToHex(pssmsg.To)))
log.Trace("pss relay block-cache match (process)", "from", common.ToHex(p.Kademlia.BaseAddr()), "to", (common.ToHex(pssmsg.To)))
return nil
}
p.addFwdCache(pssmsg)
@ -442,12 +442,12 @@ func (p *Pss) executeHandlers(topic Topic, payload []byte, from *PssAddress, asy
// will return false if using partial address
func (p *Pss) isSelfRecipient(msg *PssMsg) bool {
return bytes.Equal(msg.To, p.Overlay.BaseAddr())
return bytes.Equal(msg.To, p.Kademlia.BaseAddr())
}
// test match of leftmost bytes in given message to node's overlay address
// test match of leftmost bytes in given message to node's Kademlia address
func (p *Pss) isSelfPossibleRecipient(msg *PssMsg) bool {
local := p.Overlay.BaseAddr()
local := p.Kademlia.BaseAddr()
return bytes.Equal(msg.To[:], local[:len(msg.To)])
}
@ -816,14 +816,7 @@ func (p *Pss) forward(msg *PssMsg) error {
// send with kademlia
// find the closest peer to the recipient and attempt to send
sent := 0
p.Overlay.EachConn(to, 256, func(op network.OverlayConn, po int, isproxbin bool) bool {
// we need p2p.protocols.Peer.Send
// cast and resolve
sp, ok := op.(senderPeer)
if !ok {
log.Crit("Pss cannot use kademlia peer type")
return false
}
p.Kademlia.EachConn(to, 256, func(sp *network.Peer, po int, isproxbin bool) bool {
info := sp.Info()
// check if the peer is running pss
@ -840,7 +833,7 @@ func (p *Pss) forward(msg *PssMsg) error {
}
// get the protocol peer from the forwarding peer cache
sendMsg := fmt.Sprintf("MSG TO %x FROM %x VIA %x", to, p.BaseAddr(), op.Address())
sendMsg := fmt.Sprintf("MSG TO %x FROM %x VIA %x", to, p.BaseAddr(), sp.Address())
p.fwdPoolMu.RLock()
pp := p.fwdPool[sp.Info().ID]
p.fwdPoolMu.RUnlock()
@ -859,11 +852,11 @@ func (p *Pss) forward(msg *PssMsg) error {
// - if the peer is end recipient but the full address has not been disclosed
// - if the peer address matches the partial address fully
// - if the peer is in proxbin
if len(msg.To) < addressLength && bytes.Equal(msg.To, op.Address()[:len(msg.To)]) {
if len(msg.To) < addressLength && bytes.Equal(msg.To, sp.Address()[:len(msg.To)]) {
log.Trace(fmt.Sprintf("Pss keep forwarding: Partial address + full partial match"))
return true
} else if isproxbin {
log.Trace(fmt.Sprintf("%x is in proxbin, keep forwarding", common.ToHex(op.Address())))
log.Trace(fmt.Sprintf("%x is in proxbin, keep forwarding", common.ToHex(sp.Address())))
return true
}
// at this point we stop forwarding, and the state is as follows:

47
swarm/pss/pss_test.go
View file

@ -556,23 +556,6 @@ OUTER:
}
}
type pssTestPeer struct {
*protocols.Peer
addr []byte
}
func (t *pssTestPeer) Address() []byte {
return t.addr
}
func (t *pssTestPeer) Update(addr network.OverlayAddr) network.OverlayAddr {
return addr
}
func (t *pssTestPeer) Off() network.OverlayAddr {
return &pssTestPeer{}
}
// forwarding should skip peers that do not have matching pss capabilities
func TestMismatch(t *testing.T) {
@ -582,7 +565,7 @@ func TestMismatch(t *testing.T) {
t.Fatal(err)
}
// initialize overlay
// initialize kad
baseaddr := network.RandomAddr()
kad := network.NewKademlia((baseaddr).Over(), network.NewKadParams())
rw := &p2p.MsgPipeRW{}
@ -594,10 +577,10 @@ func TestMismatch(t *testing.T) {
Version: 0,
}
nid, _ := discover.HexID("0x01")
wrongpsspeer := &pssTestPeer{
Peer: protocols.NewPeer(p2p.NewPeer(nid, common.ToHex(wrongpssaddr.Over()), []p2p.Cap{wrongpsscap}), rw, nil),
addr: wrongpssaddr.Over(),
}
wrongpsspeer := network.NewPeer(&network.BzzPeer{
Peer: protocols.NewPeer(p2p.NewPeer(nid, common.ToHex(wrongpssaddr.Over()), []p2p.Cap{wrongpsscap}), rw, nil),
BzzAddr: &network.BzzAddr{OAddr: wrongpssaddr.Over(), UAddr: nil},
}, kad)
// one peer doesn't even have pss (boo!)
nopssaddr := network.RandomAddr()
@ -606,16 +589,16 @@ func TestMismatch(t *testing.T) {
Version: 1,
}
nid, _ = discover.HexID("0x02")
nopsspeer := &pssTestPeer{
Peer: protocols.NewPeer(p2p.NewPeer(nid, common.ToHex(nopssaddr.Over()), []p2p.Cap{nopsscap}), rw, nil),
addr: nopssaddr.Over(),
}
nopsspeer := network.NewPeer(&network.BzzPeer{
Peer: protocols.NewPeer(p2p.NewPeer(nid, common.ToHex(nopssaddr.Over()), []p2p.Cap{nopsscap}), rw, nil),
BzzAddr: &network.BzzAddr{OAddr: nopssaddr.Over(), UAddr: nil},
}, kad)
// add peers to kademlia and activate them
// it's safe so don't check errors
kad.Register([]network.OverlayAddr{wrongpsspeer})
kad.Register(wrongpsspeer.BzzAddr)
kad.On(wrongpsspeer)
kad.Register([]network.OverlayAddr{nopsspeer})
kad.Register(nopsspeer.BzzAddr)
kad.On(nopsspeer)
// create pss
@ -1636,17 +1619,17 @@ func newServices(allowRaw bool) adapters.Services {
}
}
func newTestPss(privkey *ecdsa.PrivateKey, overlay network.Overlay, ppextra *PssParams) *Pss {
func newTestPss(privkey *ecdsa.PrivateKey, kad *network.Kademlia, ppextra *PssParams) *Pss {
var nid discover.NodeID
copy(nid[:], crypto.FromECDSAPub(&privkey.PublicKey))
addr := network.NewAddrFromNodeID(nid)
// set up routing if kademlia is not passed to us
if overlay == nil {
if kad == nil {
kp := network.NewKadParams()
kp.MinProxBinSize = 3
overlay = network.NewKademlia(addr.Over(), kp)
kad = network.NewKademlia(addr.Over(), kp)
}
// create pss
@ -1654,7 +1637,7 @@ func newTestPss(privkey *ecdsa.PrivateKey, overlay network.Overlay, ppextra *Pss
if ppextra != nil {
pp.SymKeyCacheCapacity = ppextra.SymKeyCacheCapacity
}
ps, err := NewPss(overlay, pp)
ps, err := NewPss(kad, pp)
if err != nil {
return nil
}

2
swarm/swarm.go
View file

@ -356,7 +356,7 @@ func (self *Swarm) Start(srv *p2p.Server) error {
log.Error("bzz failed", "err", err)
return err
}
log.Info("Swarm network started", "bzzaddr", fmt.Sprintf("%x", self.bzz.Hive.Overlay.BaseAddr()))
log.Info("Swarm network started", "bzzaddr", fmt.Sprintf("%x", self.bzz.Hive.BaseAddr()))
if self.ps != nil {
self.ps.Start(srv)