forks/go-ethereum
1
0
Fork 1
mirror of https://github.com/ethereum/go-ethereum.git synced 2026-07-27 15:16:43 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions 3

swarm, swarm/network, swarm/pss, p2p/simulations: cleanup + doc

Browse source
This commit is contained in:
lash 2017-06-20 21:30:46 +02:00 committed by Lewis Marshall
parent 61b45b767a
commit 65023d7d4b
12 changed files with 331 additions and 414 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

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

@ -4,6 +4,7 @@ import (
"context"
"encoding/json"
"fmt"
"flag"
"io/ioutil"
"os"
"testing"
@ -27,13 +28,23 @@ var services = adapters.Services{
serviceName: newService,
}
var (
snapshotFile = flag.String("snapshot", "", "create snapshot")
nodeCount = flag.Int("nodes", 10, "number of nodes to create (default 10)")
verbose = flag.Bool("verbose", false, "output extra logs")
)
func init() {
flag.Parse()
// register the discovery service which will run as a devp2p
// protocol when using the exec adapter
adapters.RegisterServices(services)
// log.Root().SetHandler(log.LvlFilterHandler(log.LvlError, log.StreamHandler(os.Stderr, log.TerminalFormat(false))))
log.Root().SetHandler(log.LvlFilterHandler(log.LvlTrace, log.StreamHandler(os.Stderr, log.TerminalFormat(false))))
if *verbose {
log.Root().SetHandler(log.LvlFilterHandler(log.LvlTrace, log.StreamHandler(os.Stderr, log.TerminalFormat(false))))
} else {
log.Root().SetHandler(log.LvlFilterHandler(log.LvlError, log.StreamHandler(os.Stderr, log.TerminalFormat(false))))
}
}
func TestDiscoverySimulationDockerAdapter(t *testing.T) {
@ -59,15 +70,14 @@ func TestDiscoverySimulationSimAdapter(t *testing.T) {
func testDiscoverySimulation(t *testing.T, adapter adapters.NodeAdapter) {
// create network
nodeCount := 10
net := simulations.NewNetwork(adapter, &simulations.NetworkConfig{
ID: "0",
DefaultService: serviceName,
})
defer net.Shutdown()
trigger := make(chan discover.NodeID)
ids := make([]discover.NodeID, nodeCount)
for i := 0; i < nodeCount; i++ {
ids := make([]discover.NodeID, *nodeCount)
for i := 0; i < *nodeCount; i++ {
node, err := net.NewNode()
if err != nil {
t.Fatalf("error starting node: %s", err)
@ -136,13 +146,22 @@ func testDiscoverySimulation(t *testing.T, adapter adapters.NodeAdapter) {
t.Fatalf("simulation failed: %s", result.Error)
}
snap, err := net.Snapshot()
if err != nil {
t.Fatalf("no shapshot dude")
if *snapshotFile != "" {
snap, err := net.Snapshot()
if err != nil {
t.Fatalf("no shapshot dude")
}
jsonsnapshot, err := json.Marshal(snap)
if err != nil {
t.Fatalf("corrupt json snapshot: %v", err)
}
log.Info("writing snapshot", "file", *snapshotFile)
err = ioutil.WriteFile(*snapshotFile, jsonsnapshot, os.ModePerm)
if err != nil {
t.Fatal(err)
}
}
jsonsnapshot, err := json.Marshal(snap)
ioutil.WriteFile("jsonsnapshot.txt", jsonsnapshot, os.ModePerm)
t.Log("Simulation Passed:")
t.Logf("Duration: %s", result.FinishedAt.Sub(result.StartedAt))
for _, id := range ids {

34
swarm/pss/client/client.go
View file

@ -18,12 +18,12 @@
// "github.com/ethereum/go-ethereum/pot"
// "github.com/ethereum/go-ethereum/log"
// )
//
//
// type FooMsg struct {
// Bar int
// }
//
//
//
//
// func fooHandler (msg interface{}) error {
// foomsg, ok := msg.(*FooMsg)
// if ok {
@ -31,7 +31,7 @@
// }
// return fmt.Errorf("Unknown message")
// }
//
//
// spec := &protocols.Spec{
// Name: "foo",
// Version: 1,
@ -40,7 +40,7 @@
// FooMsg{},
// },
// }
//
//
// proto := &p2p.Protocol{
// Name: spec.Name,
// Version: spec.Version,
@ -51,7 +51,7 @@
// },
// }
//
// func implementation() {
// func implementation() {
// cfg := pss.NewClientConfig()
// psc := pss.NewClient(context.Background(), nil, cfg)
// err := psc.Start()
@ -61,21 +61,21 @@
// }
//
// log.Debug("connected to pss node", "bzz addr", psc.BaseAddr)
//
// err = psc.RunProtocol(proto)
//
// err = psc.RunProtocol(proto)
// if err != nil {
// log.Crit("can't start protocol on pss websocket")
// os.Exit(1)
// }
//
//
// addr := pot.RandomAddress() // should be a real address, of course
// psc.AddPssPeer(addr, spec)
//
//
// // use the protocol for something
//
//
// psc.Stop()
// }
//
// }
//
// BUG(test): TestIncoming test times out due to deadlock issues in the swarm hive
package client
@ -86,11 +86,11 @@ import (
"github.com/ethereum/go-ethereum/common"
"github.com/ethereum/go-ethereum/log"
"github.com/ethereum/go-ethereum/node"
"github.com/ethereum/go-ethereum/p2p"
"github.com/ethereum/go-ethereum/p2p/discover"
"github.com/ethereum/go-ethereum/p2p/protocols"
"github.com/ethereum/go-ethereum/pot"
"github.com/ethereum/go-ethereum/node"
"github.com/ethereum/go-ethereum/rlp"
"github.com/ethereum/go-ethereum/rpc"
"github.com/ethereum/go-ethereum/swarm/pss"
@ -125,7 +125,7 @@ func NewClientConfig() *ClientConfig {
// After a successful connection with Client.Start, BaseAddr contains the swarm overlay address of the pss node
type Client struct {
BaseAddr []byte
BaseAddr []byte
localuri string
remoteuri string
ctx context.Context
@ -234,7 +234,7 @@ func NewClientWithRPC(ctx context.Context, rpcclient *rpc.Client) (*Client, erro
protos: make(map[pss.Topic]*p2p.Protocol),
ws: rpcclient,
ctx: ctx,
BaseAddr: oaddr,
BaseAddr: oaddr,
}, nil
}
@ -267,7 +267,7 @@ func (self *Client) Start() error {
}
// Mounts a new devp2p protcool on the pss connection
// the protocol is aliased as a "pss topic"
// the protocol is aliased as a "pss topic"
// uses normal devp2p Send and incoming message handler routines from the p2p/protocols package
//
// when an incoming message is received from a peer that is not yet known to the client, this peer object is instantiated, and the protocol is run on it.

183
swarm/pss/doc.go Normal file
View file

@ -0,0 +1,183 @@
// pss provides devp2p functionality for swarm nodes without the need for a direct tcp connection between them.
//
// It uses swarm kademlia routing to send and receive messages. Routing is deterministic and will seek the shortest route available on the network.
//
// Messages are encapsulated in a devp2p message structure `PssMsg`. These capsules are forwarded from node to node using ordinary tcp devp2p until it reaches it's destination. The destination address is hinted in `PssMsg.To`
//
// The content of a PssMsg can be anything at all, down to a simple, non-descript byte-slices. But convenience methods are made available to implement devp2p protocol functionality on top of it.
//
// In its final implementation, pss is intended to become "shh over bzz," that is; "whisper over swarm." Specifically, this means that the emphemeral encryption envelopes of whisper will be used to obfuscate the correspondance. Ideally, the unencrypted content of the PssMsg will only contain a part of the address of the recipient, where the final recipient is the one who matches this partial address *and* successfully can encrypt the message.
//
// For the current state and roadmap of pss development please see https://github.com/ethersphere/swarm/wiki/swarm-dev-progress.
//
// Please report issues on https://github.com/ethersphere/go-ethereum
//
// Feel free to ask questions in https://gitter.im/ethersphere/pss
//
// TLDR IMPLEMENTATION
//
// Most developers will most probably want to use the protocol-wrapping convenience client in swarm/pss/client. Documentation and a minimal code example for the latter is found in the package documentation. The pss API can of course also be used directly. The client implementation provides a clear illustration of its intended usage.
//
// pss implements the node.Service interface. This means that the API methods will be auto-magically exposed to any RPC layer the node activates. In particular, pss provides subscription to incoming messages using the go-ethereum rpc websocket layer.
//
// The important API methods are:
// - Receive() - start a subscription to receive new incoming messages matching specific "topics"
// - Send() - send content over pss to a specified recipient
//
//
// LOWLEVEL IMPLEMENTATION
//
// code speaks louder than words:
//
// import (
// "io/ioutil"
// "os"
// "github.com/ethereum/go-ethereum/p2p"
// "github.com/ethereum/go-ethereum/log"
// "github.com/ethereum/go-ethereum/swarm/pss"
// "github.com/ethereum/go-ethereum/swarm/network"
// "github.com/ethereum/go-ethereum/swarm/storage"
// )
//
// var (
// righttopic = pss.NewTopic("foo", 4)
// wrongtopic = pss.NewTopic("bar", 2)
// )
//
// func init() {
// hs := log.StreamHandler(os.Stderr, log.TerminalFormat(true))
// hf := log.LvlFilterHandler(log.LvlTrace, hs)
// h := log.CallerFileHandler(hf)
// log.Root().SetHandler(h)
// }
//
//
// // Pss.Handler type
// func handler(msg []byte, p *p2p.Peer, from []byte) error {
// log.Debug("received", "msg", msg, "from", from, "forwarder", p.ID())
// return nil
// }
//
// func implementation() {
//
// // bogus addresses for illustration purposes
// meaddr := network.RandomAddr()
// toaddr := network.RandomAddr()
// fwdaddr := network.RandomAddr()
//
// // new kademlia for routing
// kp := network.NewKadParams()
// to := network.NewKademlia(meaddr.Over(), kp)
//
// // new (local) storage for cache
// cachedir, err := ioutil.TempDir("", "pss-cache")
// if err != nil {
// panic("overlay")
// }
// dpa, err := storage.NewLocalDPA(cachedir)
// if err != nil {
// panic("storage")
// }
//
// // setup pss
// psp := pss.NewPssParams(false)
// ps := pss.NewPss(to, dpa, psp)
//
// // does nothing but please include it
// ps.Start(nil)
//
// dereg := ps.Register(&righttopic, handler)
//
// // in its simplest form a message is just a byteslice
// payload := []byte("foobar")
//
// // send a raw message
// err = ps.SendRaw(toaddr.Over(), righttopic, payload)
// log.Error("Fails. Not connect, so nothing in kademlia. But it illustrates the point.", "err", err)
//
// // forward a full message
// envfwd := pss.NewEnvelope(fwdaddr.Over(), righttopic, payload)
// msgfwd := &pss.PssMsg{
// To: toaddr.Over(),
// Payload: envfwd,
// }
// err = ps.Forward(msgfwd)
// log.Error("Also fails, same reason. I wish, I wish, I wish there was somebody out there.", "err", err)
//
// // process an incoming message
// // (this is the first step after the devp2p PssMsg message handler)
// envme := pss.NewEnvelope(toaddr.Over(), righttopic, payload)
// msgme := &pss.PssMsg{
// To: meaddr.Over(),
// Payload: envme,
// }
// err = ps.Process(msgme)
// if err == nil {
// log.Info("this works :)")
// }
//
// // if we don't have a registered topic it fails
// dereg() // remove the previously registered topic-handler link
// ps.Process(msgme)
// log.Error("It fails as we expected", "err", err)
//
// // does nothing but please include it
// ps.Stop()
// }
//
// MESSAGE STRUCTURE
//
// NOTE! This part is subject to change. In particular the envelope structure will be re-implemented using whisper.
//
// A pss message has the following layers:
//
// PssMsg
// Contains (eventually only part of) recipient address, and (eventually) encrypted Envelope.
//
// Envelope
// Currently rlp-encoded. Contains the Payload, along with sender address, topic and expiry information.
//
// Payload
// Byte-slice of arbitrary data
//
// ProtocolMsg
// An optional convenience structure for implementation of devp2p protocols. Contains Code, Size and Payload analogous to the p2p.Msg structure, where the payload is a rlp-encoded byteslice. For transport, this struct is serialized and used as the "payload" above.
//
// TOPICS AND PROTOCOLS
//
// Pure pss is protocol agnostic. Instead it uses the notion of Topic. This is NOT the "subject" of a message. Instead this type is used to internally register handlers for messages matching respective Topics.
//
// Topic in this context virtually mean anything; protocols, chatrooms, or social media groups.
//
// When implementing devp2p protocols, topics are direct mappings to protocols name and version. The pss package provides the PssProtocol convenience structure, and a generic Handler that can be passed to Pss.Register. This makes it possible to use the same message handler code for pss that are used for direct connected peers.
//
// CONNECTIONS
//
// A "connection" in pss is a purely virtual construct. There is no mechanisms in place to ensure that the remote peer actually is there. In fact, "adding" a peer involves merely the node's opinion that the peer is there. It may issue messages to that remote peer to a directly connected peer, which in turn passes it on. But if it is not present on the network - or if there is no route to it - the message will never reach its destination through mere forwarding.
//
// When implementing the devp2p protocol stack, the "adding" of a remote peer is a prerequisite for the side actually initiating the protocol communication. Adding a peer in effect "runs" the protocol on that peer, and adds an internal mapping between a topic and that peer. It also enables sending and receiving messages using the main io-construct in devp2p - the p2p.MsgReadWriter.
//
// Under the hood, pss implements its own MsgReadWriter, which bridges MsgReadWriter.WriteMsg with Pss.SendRaw, and deftly adds an InjectMsg method which pipes incoming messages to appear on the MsgReadWriter.ReadMsg channel.
//
// An incoming connection is nothing more than an actual PssMsg appearing with a certain Topic. If a Handler har been registered to that Topic, the message will be passed to it. This constitutes a "new" connection if:
//
// - The pss node never called AddPeer with this combination of remote peer address and topic, and
//
// - The pss node never received a PssMsg from this remote peer with this specific Topic before.
//
// If it is a "new" connection, the protocol will be "run" on the remote peer, in the same manner as if it was pre-emptively added.
//
// ROUTING AND CACHING
//
// (please refer to swarm kademlia routing for an explanation of the routing algorithm used for pss)
//
// pss implements a simple caching mechanism, using the swarm DPA for storage of the messages and generation of the digest keys used in the cache table. The caching is intended to alleviate the following:
//
// - save messages so that they can be delivered later if the recipient was not online at the time of sending.
//
// - drop an identical message to the same recipient if received within a given time interval
//
// - prevent backwards routing of messages
//
// the latter may occur if only one entry is in the receiving node's kademlia. In this case the forwarder will be provided as the "nearest node" to the final recipient. The cache keeps the address of who the message was forwarded from, and if the cache lookup matches, the message will be dropped.
package pss

202
swarm/pss/pss.go
View file

@ -1,185 +1,3 @@
// pss provides devp2p functionality for swarm nodes without the need for a direct tcp connection between them.
//
// It uses swarm kademlia routing to send and receive messages. Routing is deterministic and will seek the shortest route available on the network.
//
// Messages are encapsulated in a devp2p message structure `PssMsg`. These capsules are forwarded from node to node using ordinary tcp devp2p until it reaches it's destination. The destination address is hinted in `PssMsg.To`
//
// The content of a PssMsg can be anything at all, down to a simple, non-descript byte-slices. But convenience methods are made available to implement devp2p protocol functionality on top of it.
//
// In its final implementation, pss is intended to become "shh over bzz," that is; "whisper over swarm." Specifically, this means that the emphemeral encryption envelopes of whisper will be used to obfuscate the correspondance. Ideally, the unencrypted content of the PssMsg will only contain a part of the address of the recipient, where the final recipient is the one who matches this partial address *and* successfully can encrypt the message.
//
// For the current state and roadmap of pss development please see https://github.com/ethersphere/swarm/wiki/swarm-dev-progress.
//
// Please report issues on https://github.com/ethersphere/go-ethereum
//
// Feel free to ask questions in https://gitter.im/ethersphere/pss
//
// TLDR IMPLEMENTATION
//
// Most developers will most probably want to use the protocol-wrapping convenience client in swarm/pss/client. Documentation and a minimal code example for the latter is found in the package documentation. The pss API can of course also be used directly. The client implementation provides a clear illustration of its intended usage.
//
// pss implements the node.Service interface. This means that the API methods will be auto-magically exposed to any RPC layer the node activates. In particular, pss provides subscription to incoming messages using the go-ethereum rpc websocket layer.
//
// The important API methods are:
// - Receive() - start a subscription to receive new incoming messages matching specific "topics"
// - Send() - send content over pss to a specified recipient
//
//
// LOWLEVEL IMPLEMENTATION
//
// code speaks louder than words:
//
// import (
// "io/ioutil"
// "os"
// "github.com/ethereum/go-ethereum/p2p"
// "github.com/ethereum/go-ethereum/log"
// "github.com/ethereum/go-ethereum/swarm/pss"
// "github.com/ethereum/go-ethereum/swarm/network"
// "github.com/ethereum/go-ethereum/swarm/storage"
// )
//
// var (
// righttopic = pss.NewTopic("foo", 4)
// wrongtopic = pss.NewTopic("bar", 2)
// )
//
// func init() {
// hs := log.StreamHandler(os.Stderr, log.TerminalFormat(true))
// hf := log.LvlFilterHandler(log.LvlTrace, hs)
// h := log.CallerFileHandler(hf)
// log.Root().SetHandler(h)
// }
//
//
// // Pss.Handler type
// func handler(msg []byte, p *p2p.Peer, from []byte) error {
// log.Debug("received", "msg", msg, "from", from, "forwarder", p.ID())
// return nil
// }
//
// func implementation() {
//
// // bogus addresses for illustration purposes
// meaddr := network.RandomAddr()
// toaddr := network.RandomAddr()
// fwdaddr := network.RandomAddr()
//
// // new kademlia for routing
// kp := network.NewKadParams()
// to := network.NewKademlia(meaddr.Over(), kp)
//
// // new (local) storage for cache
// cachedir, err := ioutil.TempDir("", "pss-cache")
// if err != nil {
// panic("overlay")
// }
// dpa, err := storage.NewLocalDPA(cachedir)
// if err != nil {
// panic("storage")
// }
//
// // setup pss
// psp := pss.NewPssParams(false)
// ps := pss.NewPss(to, dpa, psp)
//
// // does nothing but please include it
// ps.Start(nil)
//
// dereg := ps.Register(&righttopic, handler)
//
// // in its simplest form a message is just a byteslice
// payload := []byte("foobar")
//
// // send a raw message
// err = ps.SendRaw(toaddr.Over(), righttopic, payload)
// log.Error("Fails. Not connect, so nothing in kademlia. But it illustrates the point.", "err", err)
//
// // forward a full message
// envfwd := pss.NewEnvelope(fwdaddr.Over(), righttopic, payload)
// msgfwd := &pss.PssMsg{
// To: toaddr.Over(),
// Payload: envfwd,
// }
// err = ps.Forward(msgfwd)
// log.Error("Also fails, same reason. I wish, I wish, I wish there was somebody out there.", "err", err)
//
// // process an incoming message
// // (this is the first step after the devp2p PssMsg message handler)
// envme := pss.NewEnvelope(toaddr.Over(), righttopic, payload)
// msgme := &pss.PssMsg{
// To: meaddr.Over(),
// Payload: envme,
// }
// err = ps.Process(msgme)
// if err == nil {
// log.Info("this works :)")
// }
//
// // if we don't have a registered topic it fails
// dereg() // remove the previously registered topic-handler link
// ps.Process(msgme)
// log.Error("It fails as we expected", "err", err)
//
// // does nothing but please include it
// ps.Stop()
// }
//
// MESSAGE STRUCTURE
//
// NOTE! This part is subject to change. In particular the envelope structure will be re-implemented using whisper.
//
// A pss message has the following layers:
//
// PssMsg
// Contains (eventually only part of) recipient address, and (eventually) encrypted Envelope.
//
// Envelope
// Currently rlp-encoded. Contains the Payload, along with sender address, topic and expiry information.
//
// Payload
// Byte-slice of arbitrary data
//
// ProtocolMsg
// An optional convenience structure for implementation of devp2p protocols. Contains Code, Size and Payload analogous to the p2p.Msg structure, where the payload is a rlp-encoded byteslice. For transport, this struct is serialized and used as the "payload" above.
//
// TOPICS AND PROTOCOLS
//
// Pure pss is protocol agnostic. Instead it uses the notion of Topic. This is NOT the "subject" of a message. Instead this type is used to internally register handlers for messages matching respective Topics.
//
// Topic in this context virtually mean anything; protocols, chatrooms, or social media groups.
//
// When implementing devp2p protocols, topics are direct mappings to protocols name and version. The pss package provides the PssProtocol convenience structure, and a generic Handler that can be passed to Pss.Register. This makes it possible to use the same message handler code for pss that are used for direct connected peers.
//
// CONNECTIONS
//
// A "connection" in pss is a purely virtual construct. There is no mechanisms in place to ensure that the remote peer actually is there. In fact, "adding" a peer involves merely the node's opinion that the peer is there. It may issue messages to that remote peer to a directly connected peer, which in turn passes it on. But if it is not present on the network - or if there is no route to it - the message will never reach its destination through mere forwarding.
//
// When implementing the devp2p protocol stack, the "adding" of a remote peer is a prerequisite for the side actually initiating the protocol communication. Adding a peer in effect "runs" the protocol on that peer, and adds an internal mapping between a topic and that peer. It also enables sending and receiving messages using the main io-construct in devp2p - the p2p.MsgReadWriter.
//
// Under the hood, pss implements its own MsgReadWriter, which bridges MsgReadWriter.WriteMsg with Pss.SendRaw, and deftly adds an InjectMsg method which pipes incoming messages to appear on the MsgReadWriter.ReadMsg channel.
//
// An incoming connection is nothing more than an actual PssMsg appearing with a certain Topic. If a Handler har been registered to that Topic, the message will be passed to it. This constitutes a "new" connection if:
//
// - The pss node never called AddPeer with this combination of remote peer address and topic, and
//
// - The pss node never received a PssMsg from this remote peer with this specific Topic before.
//
// If it is a "new" connection, the protocol will be "run" on the remote peer, in the same manner as if it was pre-emptively added.
//
// ROUTING AND CACHING
//
// (please refer to swarm kademlia routing for an explanation of the routing algorithm used for pss)
//
// pss implements a simple caching mechanism, using the swarm DPA for storage of the messages and generation of the digest keys used in the cache table. The caching is intended to alleviate the following:
//
// - save messages so that they can be delivered later if the recipient was not online at the time of sending.
//
// - drop an identical message to the same recipient if received within a given time interval
//
// - prevent backwards routing of messages
//
// the latter may occur if only one entry is in the receiving node's kademlia. In this case the forwarder will be provided as the "nearest node" to the final recipient. The cache keeps the address of who the message was forwarded from, and if the cache lookup matches, the message will be dropped.
package pss
import (
@ -203,8 +21,8 @@ import (
const (
PssPeerCapacity = 256 // limit of peers kept in cache. (not implemented)
PssPeerTopicDefaultCapacity = 8 // limit of topics kept per peer. (not implemented)
digestLength = 32 // byte length of digest used for pss cache (currently same as swarm chunk hash)
PssPeerTopicDefaultCapacity = 8 // limit of topics kept per peer. (not implemented)
digestLength = 32 // byte length of digest used for pss cache (currently same as swarm chunk hash)
digestCapacity = 256 // cache entry limit (not implement)
)
@ -242,7 +60,7 @@ type pssDigest [digestLength]byte
type Pss struct {
network.Overlay // we can get the overlayaddress from this
peerPool map[pot.Address]map[Topic]p2p.MsgReadWriter // keep track of all virtual p2p.Peers we are currently speaking to
fwdPool map[discover.NodeID]*protocols.Peer // keep track of all peers sitting on the pssmsg routing layer
fwdPool map[discover.NodeID]*protocols.Peer // keep track of all peers sitting on the pssmsg routing layer
handlers map[Topic]map[*Handler]bool // topic and version based pss payload handlers
fwdcache map[pssDigest]pssCacheEntry // checksum of unique fields from pssmsg mapped to expiry, cache to determine whether to drop msg
cachettl time.Duration // how long to keep messages in fwdcache
@ -292,13 +110,12 @@ func (self *Pss) Start(srv *p2p.Server) error {
return nil
}
// For node.Service implementation. Does nothing for now, but should be included in the code for backwards compatibility.
func (self *Pss) Stop() error {
return nil
}
// devp2p protocol object for the PssMsg struct.
// devp2p protocol object for the PssMsg struct.
//
// This represents the PssMsg capsule, and is the entry point for processing, receiving and sending pss messages between directly connected peers.
func (self *Pss) Protocols() []p2p.Protocol {
@ -507,10 +324,11 @@ func (self *Pss) Forward(msg *PssMsg) error {
log.Crit("Pss cannot use kademlia peer type")
return false
}
//sendMsg := fmt.Sprintf("MSG %x TO %x FROM %x VIA %x", digest, common.ByteLabel(msg.To), common.ByteLabel(self.BaseAddr()), common.ByteLabel(op.Address()))
sendMsg := fmt.Sprintf("TO %x FROM %x VIA %x", common.ByteLabel(msg.To), common.ByteLabel(self.BaseAddr()), common.ByteLabel(op.Address()))
pp := self.fwdPool[sp.ID()]
if self.checkFwdCache(op.Address(), digest) {
log.Info("%v: peer already forwarded to", sendMsg)
log.Info(fmt.Sprintf("%v: peer already forwarded to", sendMsg))
return true
}
err := pp.Send(msg)
@ -518,7 +336,7 @@ func (self *Pss) Forward(msg *PssMsg) error {
log.Warn(fmt.Sprintf("%v: failed forwarding: %v", sendMsg, err))
return true
}
log.Trace(fmt.Sprintf("%v: successfully forwarded", sendMsg))
log.Debug(fmt.Sprintf("%v: successfully forwarded", sendMsg))
sent++
// if equality holds, p is always the first peer given in the iterator
if bytes.Equal(msg.To, op.Address()) || !isproxbin {
@ -628,7 +446,6 @@ func (prw PssReadWriter) injectMsg(msg p2p.Msg) error {
return nil
}
// For devp2p protocol integration only.
//
// Convenience object for passing messages in and out of the p2p layer
@ -640,7 +457,7 @@ type PssProtocol struct {
}
// For devp2p protocol integration only.
//
//
// Maps a Topic to a devp2p protocol.
func RegisterPssProtocol(ps *Pss, topic *Topic, spec *protocols.Spec, targetprotocol *p2p.Protocol) *PssProtocol {
pp := &PssProtocol{
@ -654,7 +471,7 @@ func RegisterPssProtocol(ps *Pss, topic *Topic, spec *protocols.Spec, targetprot
// For devp2p protocol integration only.
//
// Generic handler for initiating devp2p-like protocol connections
// Generic handler for initiating devp2p-like protocol connections
//
// This handler should be passed to Pss.Register with the associated ropic.
func (self *PssProtocol) Handle(msg []byte, p *p2p.Peer, senderAddr []byte) error {
@ -680,4 +497,3 @@ func (self *PssProtocol) Handle(msg []byte, p *p2p.Peer, senderAddr []byte) erro
return nil
}

235
swarm/pss/pss_test.go
View file

@ -34,18 +34,33 @@ const (
var (
snapshotfile string
debugflag = flag.Bool("v", false, "verbose")
// custom logging
psslogmain log.Logger
)
var services = newServices()
func init() {
flag.Parse()
rand.Seed(time.Now().Unix())
adapters.RegisterServices(services)
loglevel := log.LvlDebug
if *debugflag {
loglevel = log.LvlTrace
}
psslogmain = log.New("psslog", "*")
hs := log.StreamHandler(os.Stderr, log.TerminalFormat(true))
hf := log.LvlFilterHandler(log.LvlTrace, hs)
hf := log.LvlFilterHandler(loglevel, hs)
h := log.CallerFileHandler(hf)
log.Root().SetHandler(h)
flag.StringVar(&snapshotfile, "file", "snapsnot.json", "snapshot file")
}
func TestCache(t *testing.T) {
@ -243,72 +258,49 @@ func TestSimpleLinear(t *testing.T) {
}
}
func TestFullRandom50n(t *testing.T) {
adapter := adapters.NewSimAdapter(services)
testFullRandom(t, adapter, 50, 50)
func TestSnapshot_50_5(t *testing.T) {
testSnapshot(t, "testdata/snapshot_50.json", 5, true)
}
func TestFullRandom25n(t *testing.T) {
adapter := adapters.NewSimAdapter(services)
testFullRandom(t, adapter, 25, 25)
func TestSnapshot_5_50(t *testing.T) {
testSnapshot(t, "testdata/snapshot_5.json", 50, true)
}
func TestFullRandom10n(t *testing.T) {
adapter := adapters.NewSimAdapter(services)
testFullRandom(t, adapter, 10, 10)
func TestSnapshot_5_5(t *testing.T) {
testSnapshot(t, "testdata/snapshot_5.json", 5, true)
}
func TestFullRandom5n(t *testing.T) {
baseDir, err := ioutil.TempDir("", "swarm-test")
if err != nil {
t.Fatal(err)
}
defer os.RemoveAll(baseDir)
adapter := adapters.NewExecAdapter(baseDir)
testFullRandom(t, adapter, 5, 5)
}
func testFullRandom(t *testing.T, adapter adapters.NodeAdapter, nodecount int, msgcount int) {
}
func TestFullRandomSnapshot50(t *testing.T) {
testFullRandomSnapshot(t, true, 5)
}
func testFullRandomSnapshot(t *testing.T, sim bool, msgcount int) {
var msgtoids []discover.NodeID
var msgreceived []discover.NodeID
var cancelmain func()
//var triggerptr *chan discover.NodeID
func testSnapshot(t *testing.T, snapshotfile string, msgcount int, sim bool) {
baseDir, err := ioutil.TempDir("", "swarm-test")
if err != nil {
t.Fatal(err)
}
defer os.RemoveAll(baseDir)
// choose the adapter to use
var adapter adapters.NodeAdapter
if sim {
adapter = adapters.NewSimAdapter(services)
} else {
baseDir, err := ioutil.TempDir("", "swarm-test")
if err != nil {
t.Fatal(err)
}
defer os.RemoveAll(baseDir)
adapter = adapters.NewExecAdapter(baseDir)
}
wg := sync.WaitGroup{}
wg.Add(msgcount)
psslog := make(map[discover.NodeID]log.Logger)
psslogmain := log.New("psslog", "*")
// process shapshot
jsonsnapshot, err := ioutil.ReadFile(snapshotfile)
if err != nil {
t.Fatalf("cant read snapshot: %s", snapshotfile)
}
snapshot := &simulations.Snapshot{}
err = json.Unmarshal(jsonsnapshot, snapshot)
if err != nil {
t.Fatalf("snapshot file unreadable: %v", err)
}
for _, node := range snapshot.Nodes {
node.Config.Services = []string{"bzz", "pss"}
}
// setup network with snapshot
net := simulations.NewNetwork(adapter, &simulations.NetworkConfig{
ID: "0",
})
@ -319,121 +311,30 @@ func testFullRandomSnapshot(t *testing.T, sim bool, msgcount int) {
t.Fatalf("invalid snapshot: %v", err)
}
//msgtoids := make([]discover.NodeID, msgcount)
timeout := 15 * time.Second
ctx, cancelmain := context.WithTimeout(context.Background(), timeout)
defer cancelmain()
trigger := make(chan discover.NodeID)
//triggerptr = &trigger
// nodes expecting messages
recvids := make([]discover.NodeID, msgcount)
// the overlay address map to recvids
recvaddrs := make(map[discover.NodeID][]byte)
// for i = 0; i < nodecount; i++ {
// nodeconfig := adapters.RandomNodeConfig()
// nodeconfig.Services = []string{"bzz", "pss"}
// node, err := net.NewNodeWithConfig(nodeconfig)
// if err != nil {
// t.Fatalf("error starting node: %s", err)
// }
//
// if err := net.Start(node.ID()); err != nil {
// t.Fatalf("error starting node %s: %s", node.ID().TerminalString(), err)
// }
//
// if err := triggerChecks(ctx, &wg, triggerptr, net, node.ID()); err != nil {
// t.Fatal("error triggering checks for node %s: %s", node.ID().TerminalString(), err)
// }
// ids[i] = node.ID()
// if i < fullnodecount {
// fullpeers[ids[i]] = network.ToOverlayAddr(node.ID().Bytes())
// psslog[ids[i]] = log.New("psslog", fmt.Sprintf("%x", fullpeers[ids[i]]))
// }
// log.Debug("psslog starting node", "id", nodeconfig.ID)
// }
//
// for i, id := range fullids {
// msgfromids = append(msgfromids, id)
// msgtoids[i] = fullids[(i+(len(fullids)/2)+1)%len(fullids)]
// }
// messages actually received (registered through trigger and test check)
var msgreceived []discover.NodeID
// run a simulation which connects the 10 nodes in a ring and waits
// for full peer discovery
// action := func(ctx context.Context) error {
// for i, id := range ids {
// peerID := ids[(i+1)%len(ids)]
// if net.GetConn(id, peerID) != nil {
// continue
// }
// if err := net.Connect(id, peerID); err != nil {
// return err
// }
// psslog[id].Debug("conn ok", "one", id, "other", peerID)
// }
// return nil
// }
// check := func(ctx context.Context, id discover.NodeID) (bool, error) {
// select {
// case <-ctx.Done():
// wg.Done()
// psslog[id].Error("conn failed!", "id", id)
// return false, ctx.Err()
// default:
// }
// var tgt []byte
// var fwd struct {
// Addr []byte
// Count int
// }
//
// for i, fid := range msgfromids {
// if id == fid {
// tgt = network.ToOverlayAddr(msgtoids[(i+(len(msgtoids)/2)+1)%len(msgtoids)].Bytes())
// break
// }
// }
// p := net.GetNode(id)
// if p == nil {
// return false, fmt.Errorf("Unknown node: %v", id)
// }
// c, err := p.Client()
// if err != nil {
// return false, err
// }
// for fwd.Count < 2 {
// c.CallContext(context.Background(), &fwd, "pss_getForwarder", tgt)
// time.Sleep(time.Microsecond * 250)
// }
// psslog[id].Debug("fwd check ok", "topaddr", fmt.Sprintf("%x", common.ByteLabel(fwd.Addr)), "kadcount", fwd.Count)
// return true, nil
// }
//
// result := simulations.NewSimulation(net).Run(ctx, &simulations.Step{
// Action: action,
// Trigger: trigger,
// Expect: &simulations.Expectation{
// Nodes: ids,
// Check: check,
// },
// })
// if result.Error != nil {
// t.Fatalf("simulation failed: %s", result.Error)
// cancelmain()
// }
//
// trigger = make(chan discover.NodeID)
// triggerptr = &trigger
// trigger for expect in test
trigger := make(chan discover.NodeID)
var ids []discover.NodeID
// one wait for every message
wg := sync.WaitGroup{}
wg.Add(msgcount)
action := func(ctx context.Context) error {
var rpcerr error
var rpcbyte []byte
//for ii, id := range msgfromids {
for _, simnode := range net.Nodes {
ids = append(ids, simnode.ID())
//node := net.GetNode(id)
if simnode == nil {
return fmt.Errorf("unknown node: %s", simnode.ID())
}
@ -452,17 +353,22 @@ func testFullRandomSnapshot(t *testing.T, sim bool, msgcount int) {
if err != nil {
t.Fatalf("cant get overlayaddr: %v", err)
}
err = triggerChecks(ctx, &wg, &trigger, net, simnode.ID())
if err != nil {
t.Fatalf("trigger setup failed: %v", err)
}
}
for i := 0; i < msgcount; i++ {
idx := rand.Intn(len(net.Nodes))
sendernode := net.Nodes[idx]
toidx := rand.Intn(len(net.Nodes)-1)
if idx >= toidx {
if toidx >= idx {
toidx++
}
recvnode := net.Nodes[toidx]
recvids[i] = recvnode.ID()
msg := PingMsg{Created: time.Now()}
code, _ := PingProtocol.GetCode(&PingMsg{})
pmsg, _ := NewProtocolMsg(code, msg)
@ -472,7 +378,6 @@ func testFullRandomSnapshot(t *testing.T, sim bool, msgcount int) {
return fmt.Errorf("error getting sendernode client: %s", err)
}
client.CallContext(ctx, &rpcerr, "pss_send", PingTopic, APIMsg{
//Addr: fullpeers[msgtoids[ii]],
Addr: recvaddrs[recvnode.ID()],
Msg: pmsg,
})
@ -483,15 +388,14 @@ func testFullRandomSnapshot(t *testing.T, sim bool, msgcount int) {
return nil
}
check := func(ctx context.Context, id discover.NodeID) (bool, error) {
select {
case <-ctx.Done():
wg.Done()
return false, ctx.Err()
default:
case <-ctx.Done():
wg.Done()
return false, ctx.Err()
default:
}
msgreceived = append(msgreceived, id)
psslog[id].Info("trigger received", "id", id, "len", len(msgreceived))
psslogmain.Info("trigger received", "id", id, "len", len(msgreceived))
wg.Done()
return true, nil
}
@ -500,8 +404,7 @@ func testFullRandomSnapshot(t *testing.T, sim bool, msgcount int) {
Action: action,
Trigger: trigger,
Expect: &simulations.Expectation{
//Nodes: msgtoids,
Nodes: ids,
Nodes: recvids,
Check: check,
},
})
@ -511,13 +414,14 @@ func testFullRandomSnapshot(t *testing.T, sim bool, msgcount int) {
t.Fatalf("simulation failed: %s", result.Error)
}
if len(msgreceived) != len(msgtoids) {
t.Fatalf("Simulation Failed, got %d of %d msgs", len(msgreceived), len(msgtoids))
wg.Wait()
if len(msgreceived) != msgcount {
t.Fatalf("Simulation Failed, got %d of %d msgs", len(msgreceived), msgcount)
}
wg.Wait()
psslogmain.Info("done!")
t.Logf("Simulation Passed, got %d of %d msgs", len(msgreceived), len(msgtoids))
t.Logf("Simulation Passed, got %d of %d msgs", len(msgreceived), msgcount)
//t.Logf("Duration: %s", result.FinishedAt.Sub(result.StartedAt))
}
@ -527,7 +431,6 @@ func testFullRandomSnapshot(t *testing.T, sim bool, msgcount int) {
func triggerChecks(ctx context.Context, wg *sync.WaitGroup, trigger *chan discover.NodeID, net *simulations.Network, id discover.NodeID) error {
quitC := make(chan struct{})
got := false
node := net.GetNode(id)
if node == nil {
@ -555,12 +458,8 @@ func triggerChecks(ctx context.Context, wg *sync.WaitGroup, trigger *chan discov
defer peersub.Unsubscribe()
for {
select {
case event := <-peerevents:
if event.Type == "add" && !got {
got = true
*trigger <- id
}
case <-msgevents:
psslogmain.Debug("incoming msg", "node", id)
*trigger <- id
case err := <-peersub.Err():
if err != nil {

3
swarm/pss/pssapi.go
View file

@ -24,7 +24,7 @@ func NewAPI(ps *Pss) *API {
//
// A new handler is registered in pss for the supplied topic
//
// All incoming messages to the node matching this topic will be encapsulated in the APIMsg struct and sent to the subscriber
// All incoming messages to the node matching this topic will be encapsulated in the APIMsg struct and sent to the subscriber
func (pssapi *API) Receive(ctx context.Context, topic Topic) (*rpc.Subscription, error) {
notifier, supported := rpc.NotifierFromContext(ctx)
if !supported {
@ -111,4 +111,3 @@ func (pssapitest *APITest) GetForwarder(addr []byte) (fwd struct {
})
return
}

1
swarm/pss/testdata/snapshot_10.json vendored Executable file
View file

File diff suppressed because one or more lines are too long

1
swarm/pss/testdata/snapshot_5.json vendored Executable file
View file

@ -0,0 +1 @@
{"nodes":[{"config":{"id":"78c9669ee7b6b66642c69547501bddc4ffb5b3887c2b889b30b761fe245fee05db56bc892c6e6502a4d036c9c86ce7f9f0975d7b8b1ffebfdb8109335f364b4a","private_key":"d264be17451dbbeec8cffa553fdc9717dcc76b4ad895db7bbc672016470b6e3a","name":"node01","services":["discovery"]},"up":true},{"config":{"id":"0eec1942b5c7bea584e03d9d0f8b3219161daa68635457fd33e703b1b1512b1dd2ae27bd575a72b4748f6c85b00006dd5a08be0b8eacd6cbd8d90124b3874c7a","private_key":"b5057fc754be217f79b594072c0ef2d35c306435c8b1de75358852e92ab33ca7","name":"node02","services":["discovery"]},"up":true},{"config":{"id":"8d84d276b03161f58af3911b10da2dc33cfc93391130ed89db0f7a84bd6323b0237dab2f3c8b5c7d9953cdc745125dd67ee1122cb73a583caa98359f01afb415","private_key":"a00eea5fae5de6ca8e8d7b622ab1ede0f8adeb6fbb55a00dcd9783b760812cdd","name":"node03","services":["discovery"]},"up":true},{"config":{"id":"fdeaa376c517fe1b4574eb859d92df4272d344409ae9ec42f1c3f21e386c88b8dddd5d2febb9851e450c7e2a72113d419a6e52bf63e6bfbdde00259062ed3098","private_key":"3de00ba717baa58589e0562051f4ce6a2650c0bdab330d9eb55786fecbab2ec7","name":"node04","services":["discovery"]},"up":true},{"config":{"id":"a26035f22d085ad5c8f3892f16ed4867b6a09952c589d0177af8372666e44a454a3e49b43456c25a8365015f450432f4a6b328810e740dd2eadb898857b83e7b","private_key":"1aa2ca63a68c03f8dbc1a5a50cd7994af15fe860e1d2c7e97c0b99cd9f295532","name":"node05","services":["discovery"]},"up":true}],"conns":[{"one":"78c9669ee7b6b66642c69547501bddc4ffb5b3887c2b889b30b761fe245fee05db56bc892c6e6502a4d036c9c86ce7f9f0975d7b8b1ffebfdb8109335f364b4a","other":"a26035f22d085ad5c8f3892f16ed4867b6a09952c589d0177af8372666e44a454a3e49b43456c25a8365015f450432f4a6b328810e740dd2eadb898857b83e7b","up":true,"reverse":true,"distance":65},{"one":"0eec1942b5c7bea584e03d9d0f8b3219161daa68635457fd33e703b1b1512b1dd2ae27bd575a72b4748f6c85b00006dd5a08be0b8eacd6cbd8d90124b3874c7a","other":"78c9669ee7b6b66642c69547501bddc4ffb5b3887c2b889b30b761fe245fee05db56bc892c6e6502a4d036c9c86ce7f9f0975d7b8b1ffebfdb8109335f364b4a","up":true,"reverse":false,"distance":69},{"one":"8d84d276b03161f58af3911b10da2dc33cfc93391130ed89db0f7a84bd6323b0237dab2f3c8b5c7d9953cdc745125dd67ee1122cb73a583caa98359f01afb415","other":"0eec1942b5c7bea584e03d9d0f8b3219161daa68635457fd33e703b1b1512b1dd2ae27bd575a72b4748f6c85b00006dd5a08be0b8eacd6cbd8d90124b3874c7a","up":true,"reverse":true,"distance":68},{"one":"fdeaa376c517fe1b4574eb859d92df4272d344409ae9ec42f1c3f21e386c88b8dddd5d2febb9851e450c7e2a72113d419a6e52bf63e6bfbdde00259062ed3098","other":"8d84d276b03161f58af3911b10da2dc33cfc93391130ed89db0f7a84bd6323b0237dab2f3c8b5c7d9953cdc745125dd67ee1122cb73a583caa98359f01afb415","up":true,"reverse":false,"distance":65},{"one":"a26035f22d085ad5c8f3892f16ed4867b6a09952c589d0177af8372666e44a454a3e49b43456c25a8365015f450432f4a6b328810e740dd2eadb898857b83e7b","other":"fdeaa376c517fe1b4574eb859d92df4272d344409ae9ec42f1c3f21e386c88b8dddd5d2febb9851e450c7e2a72113d419a6e52bf63e6bfbdde00259062ed3098","up":true,"reverse":true,"distance":69}]}

1
swarm/pss/testdata/snapshot_50.json vendored Executable file
View file

File diff suppressed because one or more lines are too long

6
swarm/pss/types.go
View file

@ -49,7 +49,6 @@ func (self *PssMsg) String() string {
return fmt.Sprintf("PssMsg: Recipient: %x", common.ByteLabel(self.To))
}
// Pre-Whisper placeholder, payload of PssMsg, sender address, Topic
type Envelope struct {
Topic Topic
@ -100,6 +99,11 @@ type APIMsg struct {
Addr []byte
}
// for debugging, show nice hex version
func (self *APIMsg) String() string {
return fmt.Sprintf("APIMsg: from: %s..., msg: %s...", common.ByteLabel(self.Msg), common.ByteLabel(self.Addr))
}
// Signature for a message handler function for a PssMsg
//
// Implementations of this type are passed to Pss.Register together with a topic,

20
swarm/swarm.go
View file

@ -30,6 +30,7 @@ import (
"github.com/ethereum/go-ethereum/contracts/chequebook"
"github.com/ethereum/go-ethereum/contracts/ens"
"github.com/ethereum/go-ethereum/crypto"
"github.com/ethereum/go-ethereum/p2p/discover"
"github.com/ethereum/go-ethereum/ethclient"
"github.com/ethereum/go-ethereum/log"
"github.com/ethereum/go-ethereum/node"
@ -41,6 +42,7 @@ import (
"github.com/ethereum/go-ethereum/swarm/network"
"github.com/ethereum/go-ethereum/swarm/pss"
"github.com/ethereum/go-ethereum/swarm/storage"
//psschat "github.com/ethereum/go-ethereum/swarm/pss/protocols/chat"
)
// the swarm stack
@ -110,9 +112,12 @@ func NewSwarm(ctx *node.ServiceContext, backend chequebook.Backend, config *api.
)
config.HiveParams.Discovery = true
nodeid := discover.PubkeyID(crypto.ToECDSAPub(common.FromHex(config.PublicKey)))
addr := network.NewAddrFromNodeID(nodeid)
bzzconfig := &network.BzzConfig{
UnderlayAddr: common.FromHex(config.PublicKey),
OverlayAddr: common.FromHex(config.BzzKey),
UnderlayAddr: addr.UAddr,
HiveParams: config.HiveParams,
}
self.bzz = network.NewBzz(bzzconfig, to, nil)
@ -189,6 +194,11 @@ Start is called when the stack is started
*/
// implements the node.Service interface
func (self *Swarm) Start(net *p2p.Server) error {
// update uaddr to correct enode
newaddr := self.bzz.UpdateLocalAddr([]byte(net.Self().String()))
log.Warn("Updated bzz local addr", "oaddr", fmt.Sprintf("%x", newaddr.OAddr), "uaddr", fmt.Sprintf("%x", newaddr.UAddr))
// set chequebook
if self.swapEnabled {
ctx := context.Background() // The initial setup has no deadline.
@ -213,6 +223,7 @@ func (self *Swarm) Start(net *p2p.Server) error {
if self.pss != nil {
self.pss.Start(net)
log.Info("Pss started")
}
self.dpa.Start()
@ -242,9 +253,9 @@ func (self *Swarm) Stop() error {
self.dpa.Stop()
//self.hive.Stop()
self.bzz.Stop()
if self.pss != nil {
self.pss.Stop()
}
//if self.pss != nil {
// self.pss.Stop()
//}
if ch := self.config.Swap.Chequebook(); ch != nil {
ch.Stop()
ch.Save()
@ -265,6 +276,7 @@ func (self *Swarm) Protocols() (protos []p2p.Protocol) {
}
if self.pss != nil {
log.Warn("adding pss protos")
for _, p := range self.pss.Protocols() {
protos = append(protos, p)
}

18
swarm/swarm_psschat.go
View file

@ -229,24 +229,6 @@ func (self *Swarm) Start(net *p2p.Server) error {
return nil
})
log.Info("Pss started ... with 'pssChat' :)")
// pss.RegisterPssProtocol(self.pss, &psschat.ChatTopic, psschat.ChatProtocol, psschat.New(nil, nil, nil, func(ctrl *psschat.ChatCtrl) {
// if ctrl.OutC != nil {
// go func() {
// for {
// select {
// case msg := <-ctrl.OutC:
// err := ctrl.Peer.Send(msg)
// if err != nil {
// // do something with ctrl.ConnC here
// //pp.Drop(err)
// //return
// }
// }
// }
// }()
// }
//
// }))
}
self.dpa.Start()