From b8bf32bce397b04cb8d7a7f5035d01fe1c83c6d4 Mon Sep 17 00:00:00 2001 From: nolash Date: Mon, 20 Mar 2017 13:05:12 +0100 Subject: [PATCH] swarm/network: pss 0.1 General: - Message encapsulation is temporary until integration with whisper Incoming message handling: - Implements a dispatcher for registering handler functions to pss topics - Handlers can be any sort of function, and can send/receive on p2p.MsgReadWriter - Actual p2p.Protocol handling now as pluggable convenience function Sending and routing: - Added temporary cache solution based on message hash digest - Send and forward split up as separate methods - Send is now payload agnostic (p2p.Protocol replies massaged in p2p.MsgWriter) Tests: - TestPssRandom... Sends PSS between randomly selected nodes in different population magnitudes - TestPssFullLinearEcho A->B->C->B->A ping/pong over pss, where A and C are running pss - TestPssFullRandom50Pct 10 nodes where 5 are running pss, 5 sends to/from random nodes (messages currently get stuck in kad. routing, discarded by cache block) - Protocoltester tests status unknown --- cmd/swarm/main.go | 10 +- common/format.go | 15 + p2p/protocols/protocol.go | 27 +- p2p/testing/protocolsession.go | 44 +- swarm/api/config.go | 8 +- swarm/api/testapi.go | 12 - swarm/api/ws/server.go | 54 ++ swarm/api/ws/server_test.go | 62 ++ swarm/network/discovery.go | 21 +- swarm/network/hive.go | 5 +- swarm/network/hive_test.go | 3 +- swarm/network/kademlia.go | 16 +- swarm/network/pss.go | 478 ++++++++++++++- swarm/network/pss_test.go | 1041 +++++++++++++++++++++++++++----- swarm/network/test_overlay.go | 11 +- swarm/swarm.go | 94 ++- 16 files changed, 1650 insertions(+), 251 deletions(-) create mode 100644 swarm/api/ws/server.go create mode 100644 swarm/api/ws/server_test.go diff --git a/cmd/swarm/main.go b/cmd/swarm/main.go index 4772ad25a1..012a8c178d 100644 --- a/cmd/swarm/main.go +++ b/cmd/swarm/main.go @@ -116,6 +116,9 @@ var ( SwarmUploadMimeType = cli.StringFlag{ Name: "mime", Usage: "force mime type", + PssEnabledFlag = cli.BoolFlag{ + Name: "pss", + Usage: "Enable pss (message passing over swarm)", } CorsStringFlag = cli.StringFlag{ Name: "corsdomain", @@ -260,6 +263,8 @@ Cleans database of corrupted entries. SwarmUploadDefaultPath, SwarmUpFromStdinFlag, SwarmUploadMimeType, + // pss flags + PssEnabledFlag, } app.Flags = append(app.Flags, debug.Flags...) app.Before = func(ctx *cli.Context) error { @@ -347,7 +352,8 @@ func registerBzzService(ctx *cli.Context, stack *node.Node) { } swapEnabled := ctx.GlobalBool(SwarmSwapEnabledFlag.Name) syncEnabled := ctx.GlobalBoolT(SwarmSyncEnabledFlag.Name) - + pssEnabled := ctx.GlobalBool(PssEnabledFlag.Name) + ethapi := ctx.GlobalString(EthAPIFlag.Name) cors := ctx.GlobalString(CorsStringFlag.Name) @@ -361,7 +367,7 @@ func registerBzzService(ctx *cli.Context, stack *node.Node) { } else { swapEnabled = false } - return swarm.NewSwarm(ctx, client, bzzconfig, swapEnabled, syncEnabled, cors) + return swarm.NewSwarm(ctx, client, bzzconfig, swapEnabled, syncEnabled, cors, pssEnabled) } if err := stack.Register(boot); err != nil { utils.Fatalf("Failed to register the Swarm service: %v", err) diff --git a/common/format.go b/common/format.go index fccc299620..2932b86a82 100644 --- a/common/format.go +++ b/common/format.go @@ -23,6 +23,10 @@ import ( "time" ) +const ( + LabelLength = 8 +) + // PrettyDuration is a pretty printed version of a time.Duration value that cuts // the unnecessary precision off from the formatted textual representation. type PrettyDuration time.Duration @@ -38,3 +42,14 @@ func (d PrettyDuration) String() string { } return label } + +// useful for segfault safe reduction of log clutter +func ByteLabel(addr []byte) (b [LabelLength]byte) { + //b = make([LabelLength]byte) + if len(addr) < LabelLength { + copy(b[LabelLength - len(addr) - 1:len(addr)], addr[:]) + } else { + copy(b[:], addr[:LabelLength]) + } + return b +} diff --git a/p2p/protocols/protocol.go b/p2p/protocols/protocol.go index 417977b8ab..d9e9895f35 100644 --- a/p2p/protocols/protocol.go +++ b/p2p/protocols/protocol.go @@ -32,7 +32,6 @@ package protocols import ( "fmt" "reflect" - "time" "github.com/ethereum/go-ethereum/log" "github.com/ethereum/go-ethereum/p2p" @@ -122,6 +121,15 @@ type CodeMap struct { messages map[reflect.Type]uint64 // index of types to codes, for sending by type } +func (self *CodeMap) GetInterface(code uint64) (interface{}, bool) { + if int(code) > len(self.codes)-1 { + return nil, false + } + typ := self.codes[code] + val := reflect.New(typ) + return val.Interface(), true +} + func (self *CodeMap) GetCode(msg interface{}) (uint64, bool) { code, found := self.messages[reflect.TypeOf(msg)] return code, found @@ -268,21 +276,8 @@ func (self *Peer) Send(msg interface{}) error { return errorf(ErrInvalidMsgType, "%v", code) } log.Trace(fmt.Sprintf("=> msg #%d TO %v : %v", code, self.ID(), msg)) - go func() { - self.wErrc <- p2p.Send(self.rw, uint64(code), msg) - }() - var err error - select { - case err = <-self.wErrc: - if err == nil { - return nil - } - case <-time.NewTimer(3000 * time.Millisecond).C: - err = fmt.Errorf("write timeout") - } - err = errorf(ErrWrite, "(msg code: %v): %v", code, err) - self.Drop(err) - return err + go p2p.Send(self.rw, uint64(code), msg) + return nil } func (self *Peer) DisconnectHook(f func(error)) { diff --git a/p2p/testing/protocolsession.go b/p2p/testing/protocolsession.go index 723154e4f9..36705d3c19 100644 --- a/p2p/testing/protocolsession.go +++ b/p2p/testing/protocolsession.go @@ -2,6 +2,9 @@ package testing import ( "fmt" + "regexp" + "strconv" + "strings" "sync" "time" @@ -13,6 +16,7 @@ import ( type ProtocolSession struct { TestNodeAdapter Ids []*adapters.NodeId + ignore []uint64 } type TestMessenger interface { @@ -21,6 +25,7 @@ type TestMessenger interface { } type TestNodeAdapter interface { + p2p.Server GetPeer(id *adapters.NodeId) *adapters.Peer } @@ -64,6 +69,10 @@ func NewProtocolSession(na TestNodeAdapter, ids []*adapters.NodeId) *ProtocolSes return ps } +func (self *ProtocolSession) SetIgnoreCodes(ignore ...uint64) { + self.ignore = ignore +} + // trigger sends messages from peers func (self *ProtocolSession) trigger(trig Trigger) error { peer := self.GetPeer(trig.Peer) @@ -109,8 +118,39 @@ func (self *ProtocolSession) expect(exp Expect) error { errc := make(chan error) go func() { - log.Trace(fmt.Sprintf("waiting for msg, %v", exp.Msg)) - errc <- p2p.ExpectMsg(peer, exp.Code, exp.Msg) + var err error + ignored := true + log.Trace("waiting for msg", "code", exp.Code, "msg", exp.Msg) + for ignored { + ignored = false + err = p2p.ExpectMsg(peer, exp.Code, exp.Msg) + // frail, but we can't know what code expectmsg got otherwise + // can we do better error reporting in p2p.ExpectMsg()? + if err != nil { + if strings.Contains(err.Error(), "code") { + re, _ := regexp.Compile("got ([0-9]+),") + match := re.FindStringSubmatch(err.Error()) + if len(match) > 1 { + for _, codetoignore := range self.ignore { + codewegot, err := strconv.ParseUint(match[1], 10, 64) + if err == nil { + if codetoignore == codewegot { + ignored = true + log.Trace("ignore msg with wrong code", "received", codewegot, "expected", exp.Code) + break + } + } else { + log.Warn("expectmsg errormsg parse error?!") + } + } + } else { + log.Warn("expectmsg errormsg parse error?!") + break + } + } + } + } + errc <- err }() t := exp.Timeout diff --git a/swarm/api/config.go b/swarm/api/config.go index 23a8555007..4c4d5a3a8b 100644 --- a/swarm/api/config.go +++ b/swarm/api/config.go @@ -47,13 +47,14 @@ type Config struct { *storage.ChunkerParams *network.HiveParams Swap *swap.SwapParams - *network.SyncParams + //*network.SyncParams Path string Port string PublicKey string BzzKey string EnsRoot common.Address NetworkId uint64 + *network.PssParams } // config is agnostic to where private key is coming from @@ -72,10 +73,11 @@ func NewConfig(path string, contract common.Address, prvKey *ecdsa.PrivateKey, n keyhex := crypto.Keccak256Hash(pubkey).Hex() self = &Config{ - SyncParams: network.NewSyncParams(dirpath), - HiveParams: network.NewHiveParams(dirpath), + //SyncParams: network.NewSyncParams(dirpath), + HiveParams: network.NewHiveParams(), ChunkerParams: storage.NewChunkerParams(), StoreParams: storage.NewStoreParams(dirpath), + PssParams: network.NewPssParams(), Port: port, Path: dirpath, Swap: swap.DefaultSwapParams(contract, prvKey), diff --git a/swarm/api/testapi.go b/swarm/api/testapi.go index 6631196c17..95ba8da6af 100644 --- a/swarm/api/testapi.go +++ b/swarm/api/testapi.go @@ -29,18 +29,6 @@ func NewControl(api *Api, hive *network.Hive) *Control { return &Control{api, hive} } -func (self *Control) BlockNetworkRead(on bool) { - self.hive.BlockNetworkRead(on) -} - -func (self *Control) SyncEnabled(on bool) { - self.hive.SyncEnabled(on) -} - -func (self *Control) SwapEnabled(on bool) { - self.hive.SwapEnabled(on) -} - func (self *Control) Hive() string { return self.hive.String() } diff --git a/swarm/api/ws/server.go b/swarm/api/ws/server.go new file mode 100644 index 0000000000..71cac8c3cd --- /dev/null +++ b/swarm/api/ws/server.go @@ -0,0 +1,54 @@ +package ws + +import ( + "net" + + "github.com/ethereum/go-ethereum/logger" + "github.com/ethereum/go-ethereum/logger/glog" + "github.com/ethereum/go-ethereum/rpc" +) + +// Server is the basic configuration needs for the HTTP server and also +// includes CORS settings. +type Server struct { + //Addr string + CorsString string + Endpoint string +} + +// startWS initializes and starts the websocket RPC endpoint. +func StartWSServer(apis []rpc.API, server *Server) error { + + // Generate the whitelist based on the allowed modules + /*whitelist := make(map[string]bool) + for _, module := range modules { + whitelist[module] = true + }*/ + // Register all the APIs exposed by the services + handler := rpc.NewServer() + for _, api := range apis { + //if whitelist[api.Namespace] || (len(whitelist) == 0 && api.Public) { + if err := handler.RegisterName(api.Namespace, api.Service); err != nil { + return err + } + glog.V(logger.Debug).Infof("WebSocket registered %T under '%s'", api.Service, api.Namespace) + //} + } + // All APIs registered, start the HTTP listener + var ( + listener net.Listener + err error + ) + if listener, err = net.Listen("tcp", server.Endpoint); err != nil { + return err + } + rpc.NewWSServer(server.CorsString, handler).Serve(listener) + glog.V(logger.Info).Infof("WebSocket endpoint opened: ws://%s", server.Endpoint) + + // All listeners booted successfully + //n.wsEndpoint = endpoint + //n.wsListener = listener + //n.wsHandler = handler + + return nil +} diff --git a/swarm/api/ws/server_test.go b/swarm/api/ws/server_test.go new file mode 100644 index 0000000000..56f2aa9fe2 --- /dev/null +++ b/swarm/api/ws/server_test.go @@ -0,0 +1,62 @@ +package ws + +import ( + "context" + "testing" + "time" + + "github.com/ethereum/go-ethereum/logger" + "github.com/ethereum/go-ethereum/logger/glog" + "github.com/ethereum/go-ethereum/rpc" +) + +func init() { + glog.SetV(logger.Detail) + glog.SetToStderr(true) +} + +type TestResult struct { + Foo string `json:"foo"` +} + +func TestStartWSServer(t *testing.T) { + ep := "localhost:8099" + server := &Server{ + Endpoint: ep, + CorsString: "*", + } + apis := []rpc.API{ + { + Namespace: "pss", + Version: "0.1", + Service: makeFakeAPIHandler(), + Public: true, + }, + } + go func() { + err := StartWSServer(apis, server) + t.Logf("wsserver exited: %v", err) + }() + + time.Sleep(time.Second) + + client, err := rpc.DialWebsocket(context.Background(), "ws://" + ep, "ws://localhost") + if err != nil { + t.Fatalf("could not connect: %v", err) + } else { + t.Logf("client: %v", client) + client.Call(&TestResult{}, "pss_test") + } + +} + +func makeFakeAPIHandler() *FakeAPIHandler { + return &FakeAPIHandler{} +} + +type FakeAPIHandler struct { +} + +func (self *FakeAPIHandler) Test() { + glog.V(logger.Detail).Infof("in fakehandler Test()") +} diff --git a/swarm/network/discovery.go b/swarm/network/discovery.go index 9f628632be..cc032f60ff 100644 --- a/swarm/network/discovery.go +++ b/swarm/network/discovery.go @@ -53,7 +53,9 @@ func (self *discPeer) NotifyPeer(p Peer, po uint8) error { resp := &peersMsg{ Peers: []*peerAddr{&peerAddr{OAddr: p.OverlayAddr(), UAddr: p.UnderlayAddr()}}, // perhaps the PeerAddr interface is unnecessary generalization } - return self.Send(resp) + self.Send(resp) + //return err + return nil } // NotifyProx sends a subPeers Msg to the receiver notifying them about @@ -61,7 +63,9 @@ func (self *discPeer) NotifyPeer(p Peer, po uint8) error { // or first empty row) // callback for overlay driver func (self *discPeer) NotifyProx(po uint8) error { - return self.Send(&subPeersMsg{ProxLimit: po}) + self.Send(&subPeersMsg{ProxLimit: po}) + //return err + return nil } /* @@ -115,7 +119,7 @@ func (self *discPeer) handleSubPeersMsg(msg interface{}) error { self.proxLimit = spm.ProxLimit if !self.sentPeers { var peers []*peerAddr - self.overlay.EachLivePeer(self.OverlayAddr(), 255, func(p Peer, po int) bool { + self.overlay.EachLivePeer(self.OverlayAddr(), 255, func(p Peer, po int, isproxbin bool) bool { if uint8(po) < self.proxLimit { return false } @@ -161,7 +165,7 @@ func (self *discPeer) handleGetPeersMsg(msg interface{}) error { var peers []*peerAddr req := msg.(*getPeersMsg) i := 0 - self.overlay.EachLivePeer(self.OverlayAddr(), int(req.Order), func(n Peer, po int) bool { + self.overlay.EachLivePeer(self.OverlayAddr(), int(req.Order), func(n Peer, po int, isproxbin bool) bool { i++ // only send peers we have not sent before in this session if self.seen(n) { @@ -177,7 +181,8 @@ func (self *discPeer) handleGetPeersMsg(msg interface{}) error { resp := &peersMsg{ Peers: peers, } - return self.Send(resp) + self.Send(resp) + return nil } func RequestOrder(k Overlay, order, broadcastSize, maxPeers uint8) { @@ -186,10 +191,10 @@ func RequestOrder(k Overlay, order, broadcastSize, maxPeers uint8) { Max: maxPeers, } var i uint8 - var err error - k.EachLivePeer(nil, 255, func(n Peer, po int) bool { + //var err error + k.EachLivePeer(nil, 255, func(n Peer, po int, isproxbin bool) bool { log.Trace(fmt.Sprintf("%T sent to %v", req, n.ID())) - err = n.Send(req) + err := n.Send(req) if err == nil { i++ if i >= broadcastSize { diff --git a/swarm/network/hive.go b/swarm/network/hive.go index 8783da2c18..2e8a758ce7 100644 --- a/swarm/network/hive.go +++ b/swarm/network/hive.go @@ -45,12 +45,13 @@ type Overlay interface { On(Peer) Off(Peer) - EachLivePeer([]byte, int, func(Peer, int) bool) + EachLivePeer([]byte, int, func(Peer, int, bool) bool) EachPeer([]byte, int, func(PeerAddr, int) bool) SuggestPeer() (PeerAddr, int, bool) String() string + GetAddr() PeerAddr } // Hive implements the PeerPool interface @@ -126,7 +127,7 @@ func (self *Hive) Start(server p2p.Server, af func() <-chan time.Time) error { want = want && self.Discovery if want { - RequestOrder(self.Overlay, uint8(order), self.PeersBroadcastSetSize, self.MaxPeersPerRequest) + go RequestOrder(self.Overlay, uint8(order), self.PeersBroadcastSetSize, self.MaxPeersPerRequest) } select { diff --git a/swarm/network/hive_test.go b/swarm/network/hive_test.go index e3f690a702..741e5c4432 100644 --- a/swarm/network/hive_test.go +++ b/swarm/network/hive_test.go @@ -82,7 +82,8 @@ func TestRegisterAndConnect(t *testing.T) { }, ticker: make(chan time.Time), } - pp.Start(tc.connect, tc.ping) + //pp.Start(tc.connect, tc.ping) + pp.Start(s.TestNodeAdapter, tc.ping) defer pp.Stop() tc.ticker <- time.Now() diff --git a/swarm/network/kademlia.go b/swarm/network/kademlia.go index 12e62588b9..23f9b86560 100644 --- a/swarm/network/kademlia.go +++ b/swarm/network/kademlia.go @@ -196,7 +196,13 @@ func NewKadPeer(na PeerAddr) *KadPeer { } } -// Register enters each PeerAddr as kademlia peers into the +// retrieve the base address +// which is the overlayaddress used by peers to reach us +func (self *Kademlia) GetAddr() PeerAddr { + return self.addr +} + +// Register(nas) enters each PeerAddr as kademlia peers into the // database of known peers func (self *Kademlia) Register(nas ...PeerAddr) error { label := fmt.Sprintf("%x", RandomAddr().OverlayAddr()) @@ -288,7 +294,7 @@ type ByteAddr struct { // EachLivePeer(base, po, f) is an iterator applying f to each live peer // that has proximity order po or less as measured from the base // if base is nil, kademlia base address is used -func (self *Kademlia) EachLivePeer(base []byte, o int, f func(Peer, int) bool) { +func (self *Kademlia) EachLivePeer(base []byte, o int, f func(Peer, int, bool) bool) { var p pot.PotVal if base == nil { p = pot.PotVal(self.addr) @@ -299,7 +305,11 @@ func (self *Kademlia) EachLivePeer(base []byte, o int, f func(Peer, int) bool) { if po > o { return true } - return f(val.(*KadPeer).Peer, po) + isproxbin := false + if l, _ := p.PO(val, 0); l >= self.proxLimit() { + isproxbin = true + } + return f(val.(*KadPeer).Peer, po, isproxbin) }) } diff --git a/swarm/network/pss.go b/swarm/network/pss.go index ec1f5fed2e..dafcbf713b 100644 --- a/swarm/network/pss.go +++ b/swarm/network/pss.go @@ -2,50 +2,482 @@ package network import ( "bytes" + "encoding/binary" + "errors" "fmt" + "sync" + "time" + "github.com/ethereum/go-ethereum/common" "github.com/ethereum/go-ethereum/log" + "github.com/ethereum/go-ethereum/p2p" + "github.com/ethereum/go-ethereum/p2p/adapters" + "github.com/ethereum/go-ethereum/p2p/protocols" + "github.com/ethereum/go-ethereum/pot" + "github.com/ethereum/go-ethereum/rlp" + "github.com/ethereum/go-ethereum/swarm/storage" ) -type Pss struct { - Overlay - LocalAddr []byte - C chan []byte +const ( + DefaultTTL = 6000 + TopicLength = 32 + TopicResolverLength = 8 + PssPeerCapacity = 256 + PssPeerTopicDefaultCapacity = 8 + digestLength = 64 + digestCapacity = 256 + defaultDigestCacheTTL = time.Second +) + +var ( + errorNoForwarder = errors.New("no available forwarders in routing table") + errorForwardToSelf = errors.New("forward to self") + errorBlockByCache = errors.New("message found in blocking cache") +) + +// Defines params for Pss +type PssParams struct { + Cachettl time.Duration } -func NewPss(k Overlay, addr []byte) *Pss { - return &Pss{ - Overlay: k, - LocalAddr: addr, - C: make(chan []byte), +// Initializes default params for Pss +func NewPssParams() *PssParams { + return &PssParams{ + Cachettl: defaultDigestCacheTTL, } } +// Encapsulates the message transported over pss. +// +// Warning: do not access the To-member directly. Use *PssMsg.GetRecipient() and *PssMsg.SetRecipient() instead. type PssMsg struct { - To []byte - Data []byte + // (we need the To-member exported for type inference) + To []byte + Payload pssEnvelope } -func (pm *PssMsg) String() string { - return fmt.Sprintf("PssMsg: Recipient: %v", pm.To) +// Retrieve the remote peer receipient address of the message +func (self *PssMsg) GetRecipient() []byte { + return self.To } -func (ps *Pss) HandlePssMsg(msg interface{}) error { - pssmsg := msg.(*PssMsg) - to := pssmsg.To - if bytes.Equal(to, ps.LocalAddr) { - log.Trace(fmt.Sprintf("Pss to us, yay! %v", to)) - ps.C <- pssmsg.Data +// Set the remote peer recipient address of the message +func (self *PssMsg) SetRecipient(to []byte) { + self.To = to +} + +// String representation of PssMsg +func (self *PssMsg) String() string { + return fmt.Sprintf("PssMsg: Recipient: %x", common.ByteLabel(self.GetRecipient())) +} + +// Pre-Whisper placeholder +type pssEnvelope struct { + Topic PssTopic + TTL uint16 + Payload []byte + SenderOAddr []byte + SenderUAddr []byte +} + +// Pre-Whisper placeholder +type pssPayload struct { + Code uint64 + Size uint32 + Data []byte + ReceivedAt time.Time +} + +// Pre-Whisper placeholder +type pssCacheEntry struct { + expiresAt time.Time + receivedFrom []byte +} + +// Topic defines the context of a message being transported over pss +// It is used by pss to determine what action is to be taken on an incoming message +// Typically, one can map protocol handlers for the message payloads by mapping topic to them; see *Pss.Register() +type PssTopic [TopicLength]byte + +// Pre-Whisper placeholder +type pssDigest uint32 + +// pss provides sending messages to nodes without having to be directly connected to them. +// +// The messages are wrapped in a PssMsg structure and routed using the swarm kademlia routing. +// The structure is used by normal incoming message handlers on the nodes to determine which action to take, forward or process. +// Thus it is up to the implementer to write a handler, and link the PssMsg to this appropriate handler. +// +// The top-level Pss object provides: +// +// - access to the swarm overlay and routing (kademlia) +// - a collection of remote overlay addresses mapped to MsgReadWriters, representing the virtually connected peers +// - a collection of remote underlay address, mapped to the overlay addresses above +// - a method to send a message to specific overlayaddr +// - a dispatcher lookup, mapping protocols to topics +// - a message cache to spot messages that previously have been forwarded +type Pss struct { + Overlay // we can get the overlayaddress from this + //peerPool map[pot.Address]map[PssTopic]*PssReadWriter // keep track of all virtual p2p.Peers we are currently speaking to + peerPool map[pot.Address]map[PssTopic]p2p.MsgReadWriter // keep track of all virtual p2p.Peers we are currently speaking to + handlers map[PssTopic]func([]byte, *p2p.Peer, []byte) error // 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 + hasher func(string) storage.Hasher // hasher to digest message to cache + lock sync.Mutex +} + +func (self *Pss) hashMsg(msg *PssMsg) pssDigest { + hasher := self.hasher("SHA3")() + hasher.Reset() + hasher.Write(msg.GetRecipient()) + hasher.Write(msg.Payload.SenderUAddr) + hasher.Write(msg.Payload.SenderOAddr) + hasher.Write(msg.Payload.Topic[:]) + hasher.Write(msg.Payload.Payload) + b := hasher.Sum([]byte{}) + return pssDigest(binary.BigEndian.Uint32(b)) +} + +// Creates a new Pss instance. A node should only need one of these +// +// TODO error check overlay integrity +func NewPss(k Overlay, params *PssParams) *Pss { + return &Pss{ + Overlay: k, + //peerPool: make(map[pot.Address]map[PssTopic]*PssReadWriter, PssPeerCapacity), + peerPool: make(map[pot.Address]map[PssTopic]p2p.MsgReadWriter, PssPeerCapacity), + handlers: make(map[PssTopic]func([]byte, *p2p.Peer, []byte) error), + fwdcache: make(map[pssDigest]pssCacheEntry), + cachettl: params.Cachettl, + hasher: storage.MakeHashFunc, + } +} + +// enables to set address of node, to avoid backwards forwarding +// +// currently not in use as forwarder address is not known in the handler function hooked to the pss dispatcher. +// it is included as a courtesy to custom transport layers that may want to implement this +func (self *Pss) AddToCache(addr []byte, msg *PssMsg) error { + digest := self.hashMsg(msg) + return self.addFwdCacheSender(addr, digest) +} + +func (self *Pss) addFwdCacheSender(addr []byte, digest pssDigest) error { + self.lock.Lock() + defer self.lock.Unlock() + var entry pssCacheEntry + var ok bool + if entry, ok = self.fwdcache[digest]; !ok { + entry = pssCacheEntry{} + } + entry.receivedFrom = addr + self.fwdcache[digest] = entry + return nil +} + +func (self *Pss) addFwdCacheExpire(digest pssDigest) error { + self.lock.Lock() + defer self.lock.Unlock() + var entry pssCacheEntry + var ok bool + if entry, ok = self.fwdcache[digest]; !ok { + entry = pssCacheEntry{} + } + entry.expiresAt = time.Now().Add(self.cachettl) + self.fwdcache[digest] = entry + return nil +} + +func (self *Pss) checkFwdCache(addr []byte, digest pssDigest) bool { + self.lock.Lock() + defer self.lock.Unlock() + entry, ok := self.fwdcache[digest] + if ok { + if entry.expiresAt.After(time.Now()) { + log.Debug(fmt.Sprintf("unexpired cache for digest %x", digest)) + return true + } else if entry.expiresAt.IsZero() && bytes.Equal(addr, entry.receivedFrom) { + log.Debug(fmt.Sprintf("sendermatch %x for digest %x", common.ByteLabel(addr), digest)) + return true + } + } + return false +} + +// Takes the generated PssTopic of a protocol, and links a handler function to it +// This allows the implementer to retrieve the right handler function (invoke the right protocol) for an incoming message by inspecting the topic on it. +func (self *Pss) Register(topic PssTopic, handler func(msg []byte, p *p2p.Peer, from []byte) error) error { + self.lock.Lock() + defer self.lock.Unlock() + self.handlers[topic] = handler + return nil +} + +// Retrieves the handler function registered by *Pss.Register() +func (self *Pss) GetHandler(topic PssTopic) func([]byte, *p2p.Peer, []byte) error { + self.lock.Lock() + defer self.lock.Unlock() + return self.handlers[topic] +} + +// Links a pss peer address and topic to a dedicated p2p.MsgReadWriter in the pss peerpool, and runs the specificed protocol on this p2p.MsgReadWriter and the specified peer +// +// The effect is that now we have a "virtual" protocol running on an artificial p2p.Peer, which can be looked up and piped to through Pss using swarm overlay address and topic +func (self *Pss) AddPeer(p *p2p.Peer, addr pot.Address, protocall adapters.ProtoCall, topic PssTopic, rw p2p.MsgReadWriter) error { + self.lock.Lock() + defer self.lock.Unlock() + self.addPeerTopic(addr, topic, rw) + go func() { + err := protocall(p, rw) + log.Warn(fmt.Sprintf("pss vprotocol quit on addr %v topic %v: %v", addr, topic, err)) + }() + return nil +} + +// Removes a pss peer from the pss peerpool +func (self *Pss) RemovePeer(id pot.Address) { + self.lock.Lock() + defer self.lock.Unlock() + self.peerPool[id] = nil + return +} + +func (self *Pss) addPeerTopic(id pot.Address, topic PssTopic, rw p2p.MsgReadWriter) error { + if self.peerPool[id][topic] == nil { + self.peerPool[id] = make(map[PssTopic]p2p.MsgReadWriter, PssPeerTopicDefaultCapacity) + } + self.peerPool[id][topic] = rw + return nil +} + +func (self *Pss) removePeerTopic(id pot.Address, topic PssTopic) { + self.peerPool[id][topic] = nil + return +} + +func (self *Pss) isActive(id pot.Address, topic PssTopic) bool { + if self.peerPool[id][topic] == nil { + return false + } + return true +} + +// Sends a message using pss. The message could be anything at all, and will be handled by whichever handler function is mapped to PssTopic using *Pss.Register() +// +// The to address is a swarm overlay address +func (self *Pss) Send(to []byte, topic PssTopic, msg []byte) error { + + pssenv := pssEnvelope{ + SenderOAddr: self.Overlay.GetAddr().OverlayAddr(), + SenderUAddr: self.Overlay.GetAddr().UnderlayAddr(), + Topic: topic, + TTL: DefaultTTL, + Payload: msg, + } + + pssmsg := &PssMsg{ + Payload: pssenv, + } + pssmsg.SetRecipient(to) + + return self.Forward(pssmsg) +} + +// Forwards a pss message to the peer(s) closest to the to address +// +// Handlers that want to pass on a message should call this directly +func (self *Pss) Forward(msg *PssMsg) error { + + if self.isSelfRecipient(msg) { + return errorForwardToSelf + } + + digest := self.hashMsg(msg) + + if self.checkFwdCache(nil, digest) { + log.Trace(fmt.Sprintf("pss relay block-cache match: FROM %x TO %x", common.ByteLabel(self.Overlay.GetAddr().OverlayAddr()), common.ByteLabel(msg.GetRecipient()))) + //return errorBlockByCache return nil } - ps.EachLivePeer(to, 255, func(p Peer, po int) bool { - err := p.Send(pssmsg) - if err != nil { + // TODO:check integrity of message + + sent := 0 + + // send with kademlia + // find the closest peer to the recipient and attempt to send + self.Overlay.EachLivePeer(msg.GetRecipient(), 256, func(p Peer, po int, isproxbin bool) bool { + if self.checkFwdCache(p.OverlayAddr(), digest) { + log.Warn(fmt.Sprintf("BOUNCE DEFER PSS-relay FROM %x TO %x THRU %x:", common.ByteLabel(self.Overlay.GetAddr().OverlayAddr()), common.ByteLabel(msg.GetRecipient()), common.ByteLabel(p.OverlayAddr()))) return true } - return false + log.Warn(fmt.Sprintf("Attempting PSS-relay FROM %x TO %x THRU %x", common.ByteLabel(self.Overlay.GetAddr().OverlayAddr()), common.ByteLabel(msg.GetRecipient()), common.ByteLabel(p.OverlayAddr()))) + err := p.Send(msg) + if err != nil { + log.Warn(fmt.Sprintf("FAILED PSS-relay FROM %x TO %x THRU %x: %v", common.ByteLabel(self.Overlay.GetAddr().OverlayAddr()), common.ByteLabel(msg.GetRecipient()), common.ByteLabel(p.OverlayAddr()), err)) + return true + } + sent++ + if bytes.Equal(msg.GetRecipient(), p.OverlayAddr()) || !isproxbin { + return false + } + log.Trace(fmt.Sprintf("%x is in proxbin, so we continue sending", common.ByteLabel(p.OverlayAddr()))) + return true }) + if sent == 0 { + log.Warn("PSS Was not able to send to any peers") + } else { + self.addFwdCacheExpire(digest) + } return nil } + +// Convenience object that: +// +// - allows passing of the unwrapped PssMsg payload to the p2p level message handlers +// - interprets outgoing p2p.Msg from the p2p level to pass in to *Pss.Send() +// +// Implements p2p.MsgReadWriter +type PssReadWriter struct { + *Pss + RecipientOAddr pot.Address + LastActive time.Time + rw chan p2p.Msg + ct *protocols.CodeMap + topic *PssTopic +} + +// Implements p2p.MsgReader +func (prw PssReadWriter) ReadMsg() (p2p.Msg, error) { + msg := <-prw.rw + + log.Trace(fmt.Sprintf("pssrw readmsg: %v", msg)) + + return msg, nil +} + +// Implements p2p.MsgWriter +func (prw PssReadWriter) WriteMsg(msg p2p.Msg) error { + log.Trace(fmt.Sprintf("pssrw writemsg: %v", msg)) + ifc, found := prw.ct.GetInterface(msg.Code) + if !found { + return fmt.Errorf("Writemsg couldn't find matching interface for code %d", msg.Code) + } + msg.Decode(ifc) + + to := prw.RecipientOAddr.Bytes() + + pmsg, _ := makeMsg(msg.Code, ifc) + + return prw.Pss.Send(to, *prw.topic, pmsg) +} + +// Injects a p2p.Msg into the MsgReadWriter, so that it appears on the associated p2p.MsgReader +func (prw PssReadWriter) injectMsg(msg p2p.Msg) error { + log.Trace(fmt.Sprintf("pssrw injectmsg: %v", msg)) + prw.rw <- msg + return nil +} + +// Convenience object for passing messages in and out of the p2p layer +type PssProtocol struct { + *Pss + virtualProtocol *p2p.Protocol + topic *PssTopic + ct *protocols.CodeMap +} + +// Constructor +func NewPssProtocol(pss *Pss, topic *PssTopic, ct *protocols.CodeMap, targetprotocol *p2p.Protocol) *PssProtocol { + pp := &PssProtocol{ + Pss: pss, + virtualProtocol: targetprotocol, + topic: topic, + ct: ct, + } + return pp +} + +// Retrieves a convenience method for passing an incoming message into the p2p layer +// +// If the implementer wishes to use the p2p.Protocol (or p2p/protocols) message handling, this handler can be directly registered as a handler for the PssMsg structure +func (self *PssProtocol) GetHandler() func([]byte, *p2p.Peer, []byte) error { + return self.handle +} + +func (self *PssProtocol) handle(msg []byte, p *p2p.Peer, senderAddr []byte) error { + hashoaddr := pot.NewHashAddressFromBytes(senderAddr).Address + if !self.isActive(hashoaddr, *self.topic) { + rw := &PssReadWriter{ + Pss: self.Pss, + RecipientOAddr: hashoaddr, + rw: make(chan p2p.Msg), + ct: self.ct, + topic: self.topic, + } + self.Pss.AddPeer(p, hashoaddr, self.virtualProtocol.Run, *self.topic, rw) + } + + payload := &pssPayload{} + rlp.DecodeBytes(msg, payload) + + pmsg := p2p.Msg{ + Code: payload.Code, + Size: uint32(len(payload.Data)), + ReceivedAt: time.Now(), + Payload: bytes.NewBuffer(payload.Data), + } + + vrw := self.Pss.peerPool[hashoaddr][*self.topic].(*PssReadWriter) + vrw.injectMsg(pmsg) + + return nil +} + +func (ps *Pss) isSelfRecipient(msg *PssMsg) bool { + if bytes.Equal(msg.GetRecipient(), ps.Overlay.GetAddr().OverlayAddr()) { + return true + } + return false +} + +// Pre-Whisper placeholder +func makeMsg(code uint64, msg interface{}) ([]byte, error) { + + rlpdata, err := rlp.EncodeToBytes(msg) + if err != nil { + return nil, err + } + + // previous attempts corrupted nested structs in the payload iself upon deserializing + // therefore we use two separate []byte fields instead of peerAddr + // TODO verify that nested structs cannot be used in rlp + smsg := &pssPayload{ + Code: code, + Size: uint32(len(rlpdata)), + Data: rlpdata, + } + + rlpbundle, err := rlp.EncodeToBytes(smsg) + if err != nil { + return nil, err + } + + return rlpbundle, nil +} + +// Compiles a new PssTopic from a given name and version. +// +// Analogous to the name and version members of p2p.Protocol +func MakeTopic(s string, v int) (PssTopic, error) { + t := [TopicLength]byte{} + if len(s)+4 <= TopicLength { + copy(t[4:len(s)+4], s) + } else { + return t, fmt.Errorf("topic '%t' too long", s) + } + binary.PutVarint(t[:4], int64(v)) + return t, nil +} diff --git a/swarm/network/pss_test.go b/swarm/network/pss_test.go index 497e6e8f59..cebab91774 100644 --- a/swarm/network/pss_test.go +++ b/swarm/network/pss_test.go @@ -1,192 +1,927 @@ package network import ( - "bytes" + "context" + "encoding/hex" + "fmt" + "math/rand" + "os" + "strconv" "testing" "time" + "github.com/ethereum/go-ethereum/common" + "github.com/ethereum/go-ethereum/log" + "github.com/ethereum/go-ethereum/p2p" + "github.com/ethereum/go-ethereum/p2p/adapters" "github.com/ethereum/go-ethereum/p2p/protocols" - p2ptest "github.com/ethereum/go-ethereum/p2p/testing" + "github.com/ethereum/go-ethereum/p2p/simulations" + "github.com/ethereum/go-ethereum/rpc" ) -type pssTester struct { - *p2ptest.ProtocolTester - ct *protocols.CodeMap +const ( + protocolName = "foo" + protocolVersion = 42 +) + +func init() { + h := log.CallerFileHandler(log.StreamHandler(os.Stderr, log.TerminalFormat(true))) + log.Root().SetHandler(h) +} + +// example protocol implementation peer +// message handlers are methods of this +// goal is that we can use the same for "normal" p2p.protocols operations aswell as pss +type pssTestPeer struct { + *protocols.Peer + hasProtocol bool + successC chan bool + resultC chan int +} + +// example node simulation peer +// modeled from swarm/network/simulations/discovery/discovery_test.go - commit 08b1e42f +type pssTestNode struct { + *Hive *Pss + *adapters.SimNode + + id *adapters.NodeId + network *simulations.Network + trigger chan *adapters.NodeId + run adapters.ProtoCall + ct *protocols.CodeMap + expectC chan []int } -func TestPssTwoToSelf(t *testing.T) { - addr := RandomAddr() - pt := newPssTester(t, addr, 2) - defer pt.Stop() - payload := []byte("foo42") +func (n *pssTestNode) Add(peer Peer) error { + err := n.Hive.Add(peer) + time.Sleep(time.Millisecond * 250) + n.triggerCheck() + return err +} - subpeermsgcode, found := pt.ct.GetCode(&subPeersMsg{}) - if !found { - t.Fatalf("peerMsg not defined") - } - pssmsgcode, found := pt.ct.GetCode(&PssMsg{}) - if !found { - t.Fatalf("PssMsg not defined") - } +func (n *pssTestNode) Remove(peer Peer) { + n.Hive.Remove(peer) +} - hs_pivot := correctBzzHandshake(addr) +func (n *pssTestNode) Start() error { + return n.Hive.Start(n.SimNode, n.hiveKeepAlive) +} - for _, id := range pt.Ids { - hs_sim := correctBzzHandshake(NewPeerAddrFromNodeId(id)) - <-pt.GetPeer(id).Connc - err := pt.TestExchanges(bzzHandshakeExchange(hs_pivot, hs_sim, id)...) - if err != nil { - t.Fatalf("Handshake fail: %v", err) - } +func (n *pssTestNode) Stop() error { + n.Hive.Stop() + return nil +} - err = pt.TestExchanges( - p2ptest.Exchange{ - Expects: []p2ptest.Expect{ - p2ptest.Expect{ - Code: subpeermsgcode, - Msg: &subPeersMsg{}, - Peer: id, - }, - }, - }, - ) - if err != nil { - t.Fatalf("Subpeersmsg to peer %v fail: %v", id, err) - } - } +func (n *pssTestNode) connectPeer(s string) error { + return n.network.Connect(n.id, adapters.NewNodeIdFromHex(s)) +} - err := pt.TestExchanges( - p2ptest.Exchange{ - Triggers: []p2ptest.Trigger{ - p2ptest.Trigger{ - Code: pssmsgcode, - Msg: &PssMsg{ - To: addr.OverlayAddr(), - Data: payload, - }, - Peer: pt.Ids[0], - }, - }, +func (n *pssTestNode) hiveKeepAlive() <-chan time.Time { + return time.Tick(time.Second * 10) +} + +func (n *pssTestNode) triggerCheck() { + //go func() { n.trigger <- n.id }() + go func() { n.trigger <- adapters.NewNodeId(n.Addr()) }() +} +/* +func (n *pssTestNode) RunProtocol(id *adapters.NodeId, rw, rrw p2p.MsgReadWriter, peer *adapters.Peer) error { + return n.NodeAdapter.(adapters.ProtocolRunner).RunProtocol(id, rw, rrw, peer) +} +*/ +func (n *pssTestNode) ProtoCall() adapters.ProtoCall { + return n.run +} + +func (n *pssTestNode) OverlayAddr() []byte { + return n.Pss.Overlay.GetAddr().OverlayAddr() +} + +func (n *pssTestNode) UnderlayAddr() []byte { + return n.id.Bytes() + //return n.Addr() +} + +// the content of the msgs we're sending in the tests +type PssTestPayload struct { + Data string +} + +func (m *PssTestPayload) String() string { + return m.Data +} + +func TestPssCache(t *testing.T) { + var err error + to, _ := hex.DecodeString("08090a0b0c0d0e0f1011121314150001020304050607161718191a1b1c1d1e1f") + oaddr, _ := hex.DecodeString("000102030405060708090a0b0c0d0e0f101112131415161718191a1b1c1d1e1f") + uaddr, _ := hex.DecodeString("101112131415161718191a1b1c1d1e1f000102030405060708090a0b0c0d0e0f") + ps := makePss(oaddr) + pp := NewPssParams() + topic, _ := MakeTopic(protocolName, protocolVersion) + data := []byte("foo") + fwdaddr := RandomAddr() + msg := &PssMsg{ + Payload: pssEnvelope{ + TTL: 0, + SenderOAddr: oaddr, + SenderUAddr: uaddr, + Topic: topic, + Payload: data, }, - ) - if err != nil { - t.Fatalf("PssMsg sending %v to %v (pivot) fail: %v", pt.Ids[0], addr.OverlayAddr(), err) } + msg.SetRecipient(to) - alarm := time.NewTimer(1000 * time.Millisecond) - select { - case data := <-pt.C: - if !bytes.Equal(data, payload) { - t.Fatalf("Data transfer failed, expected: %v, got: %v", payload, data) - } - case <-alarm.C: - t.Fatalf("Pivot receive of PssMsg from %v timeout", pt.Ids[0]) - } -} - -func TestPssTwoRelaySelf(t *testing.T) { - // <-(chan bool)(nil) - addr := RandomAddr() - pt := newPssTester(t, addr, 2) - defer pt.Stop() - - subpeermsgcode, found := pt.ct.GetCode(&subPeersMsg{}) - if !found { - t.Fatalf("peerMsg not defined") - } - pssmsgcode, found := pt.ct.GetCode(&PssMsg{}) - if !found { - t.Fatalf("PssMsg not defined") - } - - hs_pivot := correctBzzHandshake(addr) - - for _, id := range pt.Ids { - hs_sim := correctBzzHandshake(NewPeerAddrFromNodeId(id)) - <-pt.GetPeer(id).Connc - err := pt.TestExchanges(bzzHandshakeExchange(hs_pivot, hs_sim, id)...) - if err != nil { - t.Fatalf("Handshake fail: %v", err) - } - - err = pt.TestExchanges( - p2ptest.Exchange{ - Expects: []p2ptest.Expect{ - p2ptest.Expect{ - Code: subpeermsgcode, - Msg: &subPeersMsg{}, - Peer: id, - }, - }, - }, - ) - if err != nil { - t.Fatalf("Subpeersmsg to peer %v fail: %v", id, err) - } - } - - err := pt.TestExchanges( - p2ptest.Exchange{ - Expects: []p2ptest.Expect{ - p2ptest.Expect{ - Code: pssmsgcode, - Msg: &PssMsg{ - To: pt.Ids[0].Bytes(), - Data: []byte("foo42"), - }, - Peer: pt.Ids[0], - }, - }, - Triggers: []p2ptest.Trigger{ - p2ptest.Trigger{ - Code: pssmsgcode, - Msg: &PssMsg{ - To: pt.Ids[0].Bytes(), - Data: []byte("foo42"), - }, - Peer: pt.Ids[1], - }, - }, + msgtwo := &PssMsg{ + Payload: pssEnvelope{ + TTL: 0, + SenderOAddr: uaddr, + SenderUAddr: oaddr, + Topic: topic, + Payload: data, }, - ) + } + msgtwo.SetRecipient(to) + + digest := ps.hashMsg(msg) + digesttwo := ps.hashMsg(msgtwo) + + if digest != 3595343914 { + t.Fatalf("digest - got: %d, expected: %d", digest, 3595343914) + } + + if digest == digesttwo { + t.Fatalf("different msgs return same crc: %d", digesttwo) + } + + // check the sender cache + err = ps.addFwdCacheSender(fwdaddr.OverlayAddr(), digest) if err != nil { - t.Fatalf("PssMsg routing from %v to %v fail: %v", pt.Ids[0], pt.Ids[1], err) + t.Fatalf("write to pss sender cache failed: %v", err) + } + + if !ps.checkFwdCache(fwdaddr.OverlayAddr(), digest) { + t.Fatalf("message %v should have SENDER record in cache but checkCache returned false", msg) + } + + if ps.checkFwdCache(fwdaddr.OverlayAddr(), digesttwo) { + t.Fatalf("message %v should NOT have SENDER record in cache but checkCache returned true", msgtwo) + } + + // check the expire cache + err = ps.addFwdCacheExpire(digest) + if err != nil { + t.Fatalf("write to pss expire cache failed: %v", err) + } + + if !ps.checkFwdCache(nil, digest) { + t.Fatalf("message %v should have EXPIRE record in cache but checkCache returned false", msg) + } + + if ps.checkFwdCache(nil, digesttwo) { + t.Fatalf("message %v should NOT have EXPIRE record in cache but checkCache returned true", msgtwo) + } + + time.Sleep(pp.Cachettl) + if ps.checkFwdCache(nil, digest) { + t.Fatalf("message %v should have expired from cache but checkCache returned true", msg) + } + + err = ps.AddToCache(fwdaddr.OverlayAddr(), msgtwo) + if err != nil { + t.Fatalf("public accessor cache write failed: %v", err) + } + + if !ps.checkFwdCache(fwdaddr.OverlayAddr(), digesttwo) { + t.Fatalf("message %v should have SENDER record in cache but checkCache returned false", msgtwo) } } -func newPssTester(t *testing.T, addr *peerAddr, n int) *pssTester { - return newPssBaseTester(t, addr, n) +func TestPssRegisterHandler(t *testing.T) { + var topic PssTopic + var err error + addr := RandomAddr() + ps := makePss(addr.UnderlayAddr()) + + topic, _ = MakeTopic(protocolName, protocolVersion) + err = ps.Register(topic, func(msg []byte, p *p2p.Peer, sender []byte) error { return nil }) + if err != nil { + t.Fatalf("couldnt register protocol 'foo' v 42: %v", err) + } + + topic, _ = MakeTopic(protocolName, protocolVersion) + err = ps.Register(topic, func(msg []byte, p *p2p.Peer, sender []byte) error { return nil }) + if err == nil { + t.Fatalf("register protocol 'abc..789' v 65536 should have failed") + } } -func newPssBaseTester(t *testing.T, addr *peerAddr, n int) *pssTester { +func TestPssFullRandom10_10_5(t *testing.T) { + testPssFullRandom(t, 10, 10, 5) +} + +func TestPssFullRandom50_50_5(t *testing.T) { + testPssFullRandom(t, 50, 50, 5) +} + +func TestPssFullRandom50_50_25(t *testing.T) { + testPssFullRandom(t, 50, 50, 25) +} + +func TestPssFullRandom10_100_50(t *testing.T) { + testPssFullRandom(t, 10, 100, 50) +} + +func TestPssFullRandom50_100_50(t *testing.T) { + testPssFullRandom(t, 50, 100, 50) +} + +func TestPssFullRandom100_100_5(t *testing.T) { + testPssFullRandom(t, 100, 100, 5) +} + +func TestPssFullRandom100_100_25(t *testing.T) { + testPssFullRandom(t, 100, 100, 25) +} + +func TestPssFullRandom100_100_50(t *testing.T) { + testPssFullRandom(t, 100, 100, 50) +} + +func testPssFullRandom(t *testing.T, numsends int, numnodes int, numfullnodes int) { + var action func(ctx context.Context) error + var i int + var check func(ctx context.Context, id *adapters.NodeId) (bool, error) + var ctx context.Context + var result *simulations.StepResult + var timeout time.Duration + var cancel context.CancelFunc + + fullnodes := []*adapters.NodeId{} + sends := []int{} // sender/receiver ids array indices pairs + expectnodes := make(map[*adapters.NodeId]int) // how many messages we're expecting on each respective node + expectnodesids := []*adapters.NodeId{} // the nodes to expect on (needed by checker) + expectnodesresults := make(map[*adapters.NodeId][]int) // which messages expect actually got + + vct := protocols.NewCodeMap(protocolName, protocolVersion, 65535, &PssTestPayload{}) + topic, _ := MakeTopic(protocolName, protocolVersion) + + trigger := make(chan *adapters.NodeId) + net := simulations.NewNetwork(&simulations.NetworkConfig{ + Id: "0", + Backend: true, + }) + testpeers := make(map[*adapters.NodeId]*pssTestPeer) + nodes := newPssSimulationTester(t, numnodes, numfullnodes, net, trigger, vct, protocolName, protocolVersion, testpeers) + + ids := []*adapters.NodeId{} + + // connect the peers + action = func(ctx context.Context) error { + for id, _ := range nodes { + ids = append(ids, id) + if _, ok := testpeers[id]; ok { + log.Trace(fmt.Sprintf("adding fullnode %x to testpeers %p", common.ByteLabel(id.Bytes()), testpeers)) + fullnodes = append(fullnodes, id) + } + } + for i, id := range ids { + var peerId *adapters.NodeId + if i != 0 { + peerId = ids[i-1] + if err := net.Connect(id, peerId); err != nil { + return err + } + } + } + return nil + } + check = func(ctx context.Context, id *adapters.NodeId) (bool, error) { + select { + case <-ctx.Done(): + return false, ctx.Err() + default: + } + + node, ok := nodes[id] + if !ok { + return false, fmt.Errorf("unknown node: %s (%v)", id, node) + } else { + log.Trace(fmt.Sprintf("sim check ok node %v", id)) + } + + return true, nil + } + + timeout = 10 * time.Second + ctx, cancel = context.WithTimeout(context.Background(), timeout) + + 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) + } + cancel() + + // ensure that we didn't get lost in concurrency issues + if len(fullnodes) != numfullnodes { + t.Fatalf("corrupt fullnodes array, expected %d, have %d", numfullnodes, len(fullnodes)) + } + + // ensure that the channel is clean + trigger = make(chan *adapters.NodeId) + + // randomly decide which nodes to send to and from + rand.Seed(time.Now().Unix()) + for i = 0; i < numsends; i++ { + s := rand.Int() % numfullnodes + r := s + for r == s { + r = rand.Int() % numfullnodes + } + log.Trace(fmt.Sprintf("rnd pss: idx %d->%d (%x -> %x)", s, r, common.ByteLabel(fullnodes[s].Bytes()), common.ByteLabel(fullnodes[r].Bytes()))) + expectnodes[fullnodes[r]]++ + sends = append(sends, s, r) + } + + // distinct array of nodes to expect on + for k, _ := range expectnodes { + expectnodesids = append(expectnodesids, k) + } + + // wait a bit for the kademlias to fill up + z, _ := time.ParseDuration(fmt.Sprintf("%dms", (numnodes * 25))) + if z.Seconds() < 1.0 { + z = time.Second + } + time.Sleep(z) + + // send and monitor receive of pss + action = func(ctx context.Context) error { + code, _ := vct.GetCode(&PssTestPayload{}) + + for i := 0; i < len(sends); i += 2 { + msgbytes, _ := makeMsg(code, &PssTestPayload{ + Data: fmt.Sprintf("%v", i+1), + }) + go func(i int, expectnodesresults map[*adapters.NodeId][]int) { + expectnode := fullnodes[sends[i+1]] // the receiving node + sendnode := fullnodes[sends[i]] // the sending node + oaddr := nodes[expectnode].OverlayAddr() + err := nodes[sendnode].Pss.Send(oaddr, topic, msgbytes) + if err != nil { + t.Fatalf("could not send pss: %v", err) + } + + select { + // if the pss is delivered + case <-testpeers[expectnode].successC: + log.Trace(fmt.Sprintf("got successC from node %x", common.ByteLabel(expectnode.Bytes()))) + expectnodesresults[expectnode] = append(expectnodesresults[expectnode], <-testpeers[expectnode].resultC) + // if not we time out, -1 means fail tick + case <-time.NewTimer(time.Second).C: + log.Trace(fmt.Sprintf("result timed out on node %x", common.ByteLabel(expectnode.Bytes()))) + expectnodesresults[expectnode] = append(expectnodesresults[expectnode], -1) + } + + // we can safely send to the check handler if we got feedback for all msgs we sent to a particular node + if len(expectnodesresults[expectnode]) == expectnodes[expectnode] { + trigger <- expectnode + nodes[expectnode].expectC <- expectnodesresults[expectnode] + } + }(i, expectnodesresults) + } + return nil + } + + // results + check = func(ctx context.Context, id *adapters.NodeId) (bool, error) { + select { + case <-ctx.Done(): + return false, ctx.Err() + default: + } + + receives := <-nodes[id].expectC + log.Trace(fmt.Sprintf("expect received %d msgs on from node %x: %v", len(receives), common.ByteLabel(id.Bytes()), receives)) + return true, nil + } + + timeout = 10 * time.Second + ctx, cancel = context.WithTimeout(context.Background(), timeout) + defer cancel() + + result = simulations.NewSimulation(net).Run(ctx, &simulations.Step{ + Action: action, + Trigger: trigger, + Expect: &simulations.Expectation{ + Nodes: expectnodesids, + Check: check, + }, + }) + if result.Error != nil { + t.Fatalf("simulation failed: %s", result.Error) + } + + t.Log("Simulation Passed:") + + for i := 0; i < len(sends); i += 2 { + t.Logf("Pss #%d: oaddr %x -> %x (uaddr %x -> %x)", i/2+1, + common.ByteLabel(nodes[fullnodes[sends[i]]].Pss.GetAddr().OverlayAddr()), + common.ByteLabel(nodes[fullnodes[sends[i+1]]].Pss.GetAddr().OverlayAddr()), + common.ByteLabel(fullnodes[sends[i]].Bytes()), + common.ByteLabel(fullnodes[sends[i+1]].Bytes())) + } + totalfails := 0 + for id, results := range expectnodesresults { + fails := 0 + for _, r := range results { + if r == -1 { + fails++ + } + } + t.Logf("Node oaddr %x (uaddr %x) was sent %d msgs, of which %d failed", common.ByteLabel(nodes[id].Pss.GetAddr().OverlayAddr()), common.ByteLabel(id.Bytes()), len(results), fails) + totalfails += fails + } + t.Logf("Total sent: %d, total fail: %d (%.2f%%)", len(sends)/2, totalfails, (float32(totalfails)/float32(len(sends)/2+1))*100) + + for _, node := range nodes { + logstring := fmt.Sprintf("Node oaddr %x kademlia: ", common.ByteLabel(node.Pss.Overlay.GetAddr().OverlayAddr())) + node.Pss.Overlay.EachLivePeer(nil, 256, func(p Peer, po int, isprox bool) bool { + logstring += fmt.Sprintf("%x ", common.ByteLabel(p.OverlayAddr())) + return true + }) + t.Log(logstring) + } +} + +// pss simulation test +// (simnodes running protocols) +func TestPssFullLinearEcho(t *testing.T) { + + var action func(ctx context.Context) error + var check func(ctx context.Context, id *adapters.NodeId) (bool, error) + var ctx context.Context + var result *simulations.StepResult + var timeout time.Duration + var cancel context.CancelFunc + + var firstpssnode *adapters.NodeId + var secondpssnode *adapters.NodeId + + vct := protocols.NewCodeMap(protocolName, protocolVersion, 65535, &PssTestPayload{}) + topic, _ := MakeTopic(protocolName, protocolVersion) + + fullnodes := []*adapters.NodeId{} + trigger := make(chan *adapters.NodeId) + net := simulations.NewNetwork(&simulations.NetworkConfig{ + Id: "0", + Backend: true, + }) + testpeers := make(map[*adapters.NodeId]*pssTestPeer) + nodes := newPssSimulationTester(t, 3, 2, net, trigger, vct, protocolName, protocolVersion, testpeers) + ids := []*adapters.NodeId{} // ohh risky! but the action for a specific id should come before the expect anyway + + action = func(ctx context.Context) error { + var thinnodeid *adapters.NodeId + for id, _ := range nodes { + ids = append(ids, id) + if _, ok := testpeers[id]; ok { + log.Trace(fmt.Sprintf("adding fullnode %x to testpeers %p", common.ByteLabel(id.Bytes()), testpeers)) + fullnodes = append(fullnodes, id) + } else { + thinnodeid = id + } + } + if err := net.Connect(fullnodes[0], thinnodeid); err != nil { + return err + } + if err := net.Connect(thinnodeid, fullnodes[1]); err != nil { + return err + } + + /*for i, id := range ids { + var peerId *adapters.NodeId + if i != 0 { + peerId = ids[i-1] + if err := net.Connect(id, peerId); err != nil { + return err + } + } + }*/ + return nil + } + check = func(ctx context.Context, id *adapters.NodeId) (bool, error) { + select { + case <-ctx.Done(): + return false, ctx.Err() + default: + } + + node, ok := nodes[id] + if !ok { + return false, fmt.Errorf("unknown node: %s (%v)", id, node) + } else { + log.Trace(fmt.Sprintf("sim check ok node %v", id)) + } + + return true, nil + } + + timeout = 10 * time.Second + ctx, cancel = context.WithTimeout(context.Background(), timeout) + + 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) + } + cancel() + + nonode := &adapters.NodeId{} + firstpssnode = nonode + secondpssnode = nonode + + // first find a node that we're connected to + for firstpssnode == nonode { + log.Debug(fmt.Sprintf("Waiting for pss relaypeer for %x close to %x ...", common.ByteLabel(nodes[fullnodes[0]].OverlayAddr()), common.ByteLabel(nodes[ids[1]].OverlayAddr()))) + nodes[fullnodes[0]].Pss.Overlay.EachLivePeer(nodes[fullnodes[1]].OverlayAddr(), 256, func(p Peer, po int, isprox bool) bool { + for _, id := range ids { + if id.NodeID == p.ID() { + firstpssnode = id + log.Debug(fmt.Sprintf("PSS relay found; relaynode %v kademlia %v", common.ByteLabel(id.Bytes()), common.ByteLabel(firstpssnode.Bytes()))) + } + } + if firstpssnode == nonode { + return true + } + return false + }) + if firstpssnode == nonode { + time.Sleep(time.Millisecond * 100) + } + } + + // then find the node it's connected to + for secondpssnode == nonode { + log.Debug(fmt.Sprintf("PSS kademlia: Waiting for recipientpeer for %x close to %x ...", common.ByteLabel(nodes[firstpssnode].OverlayAddr()), common.ByteLabel(nodes[fullnodes[1]].OverlayAddr()))) + nodes[firstpssnode].Pss.Overlay.EachLivePeer(nodes[fullnodes[1]].OverlayAddr(), 256, func(p Peer, po int, isprox bool) bool { + for _, id := range ids { + if id.NodeID == p.ID() && id.NodeID != fullnodes[0].NodeID { + secondpssnode = id + log.Debug(fmt.Sprintf("PSS recipient found; relaynode %v kademlia %v", common.ByteLabel(id.Bytes()), common.ByteLabel(secondpssnode.Bytes()))) + } + } + if secondpssnode == nonode { + return true + } + return false + }) + if secondpssnode == nonode { + time.Sleep(time.Millisecond * 100) + } + } + + action = func(ctx context.Context) error { + code, _ := vct.GetCode(&PssTestPayload{}) + msgbytes, _ := makeMsg(code, &PssTestPayload{ + Data: "ping", + }) + + go func() { + oaddr := nodes[secondpssnode].OverlayAddr() + err := nodes[ids[0]].Pss.Send(oaddr, topic, msgbytes) + if err != nil { + t.Fatalf("could not send pss: %v", err) + } + trigger <- ids[0] + }() + + return nil + } + check = func(ctx context.Context, id *adapters.NodeId) (bool, error) { + select { + case <-ctx.Done(): + return false, ctx.Err() + default: + } + + // also need to know if the protocolpeer is set up + time.Sleep(time.Millisecond * 100) + return <-testpeers[ids[0]].successC, nil + //return true, nil + } + + timeout = 10 * time.Second + ctx, cancel = context.WithTimeout(context.Background(), timeout) + defer cancel() + + result = simulations.NewSimulation(net).Run(ctx, &simulations.Step{ + Action: action, + Trigger: trigger, + Expect: &simulations.Expectation{ + Nodes: []*adapters.NodeId{ids[0]}, + Check: check, + }, + }) + if result.Error != nil { + t.Fatalf("simulation failed: %s", result.Error) + } + + t.Log("Simulation Passed:") +} + +// numnodes: how many nodes to create +// pssnodeidx: on which node indices to start the pss +// net: the simulated network +// trigger: hook needed for simulation event reporting +// vct: codemap for virtual protocol +// name: name for virtual protocol (and pss topic) +// version: name for virtual protocol (and pss topic) +// testpeers: pss-specific peers, with hook needed for simulation event reporting + +func newPssSimulationTester(t *testing.T, numnodes int, numfullnodes int, net *simulations.Network, trigger chan *adapters.NodeId, vct *protocols.CodeMap, name string, version int, testpeers map[*adapters.NodeId]*pssTestPeer) map[*adapters.NodeId]*pssTestNode { + topic, _ := MakeTopic(name, version) + nodes := make(map[*adapters.NodeId]*pssTestNode, numnodes) + //svcs := make(map[*adapters.NodeId]*pssTestService, numnodes) + psss := make(map[*adapters.NodeId]*Pss) + net.SetNaf(func(conf *simulations.NodeConfig) adapters.NodeAdapter { + node := &pssTestNode{ + Pss: psss[conf.Id], + Hive: nil, + SimNode: &adapters.SimNode{}, + id: conf.Id, + network: net, + trigger: trigger, + ct: vct, + expectC: make(chan []int), + } + var handlefunc func(interface{}) error + addr := NewPeerAddrFromNodeId(conf.Id) + if testpeers[conf.Id] != nil { + handlefunc = makePssHandleProtocol(psss[conf.Id]) + log.Trace(fmt.Sprintf("Making full protocol id %x addr %x (testpeers %p)", common.ByteLabel(conf.Id.Bytes()), common.ByteLabel(addr.OverlayAddr()), testpeers)) + } else { + handlefunc = makePssHandleForward(psss[conf.Id]) + } + //node = newPssTester(t, psss[conf.Id], addr, 0, handlefunc, net, trigger) + testservice := newPssTestService(t, handlefunc, node) + nodes[conf.Id] = testservice.node + svc := adapters.NewSimNode(conf.Id, testservice, net) + testservice.Start(svc) + //svcs[conf.Id] = testservice + return node.SimNode + }) + ids := adapters.RandomNodeIds(numnodes) + i := 0 + for _, id := range ids { + addr := NewPeerAddrFromNodeId(id) + psss[id] = makePss(addr.OverlayAddr()) + if i < numfullnodes { + tp := &pssTestPeer{ + Peer: &protocols.Peer{ + Peer: &p2p.Peer{}, + }, + successC: make(chan bool), + resultC: make(chan int), + } + testpeers[id] = tp + targetprotocol := makeCustomProtocol(name, version, vct, testpeers[id]) + pssprotocol := NewPssProtocol(psss[id], &topic, vct, targetprotocol) + psss[id].Register(topic, pssprotocol.GetHandler()) + } + + net.NewNode(&simulations.NodeConfig{Id: id}) + if err := net.Start(id); err != nil { + t.Fatalf("error starting node %s: %s", id.Label(), err) + } + i++ + } + + return nodes + //return svcs +} + +type pssTestService struct{ + node *pssTestNode // get addrs from this + msgFunc func(interface{}) error +} + +func newPssTestService(t *testing.T, handlefunc func(interface{}) error, testnode *pssTestNode) *pssTestService { + hp := NewHiveParams() + //hp.CallInterval = 250 + testnode.Hive = NewHive(hp, testnode.Pss.Overlay) + return &pssTestService{ + //nid := adapters.NewNodeId(addr.UnderlayAddr()) + msgFunc: handlefunc, + node: testnode, + } +} + +func (self *pssTestService) Start(server p2p.Server) error { + self.node.SimNode = server.(*adapters.SimNode) // server is adapter.SimnNode now + return nil +} + +func (self *pssTestService) Stop() error { + return nil +} + +func (self *pssTestService) Protocols() []p2p.Protocol { ct := BzzCodeMap() + for _, m := range DiscoveryMsgs { + ct.Register(m) + } ct.Register(&PssMsg{}) - ct.Register(&peersMsg{}) - ct.Register(&getPeersMsg{}) - ct.Register(&subPeersMsg{}) // why is this public? - - kp := NewKadParams() - kp.MinProxBinSize = 3 - to := NewKademlia(addr.OverlayAddr(), kp) - pp := NewHive(NewHiveParams(), to) - ps := NewPss(to, addr.OverlayAddr()) + //Bzz(addr.OverlayAddr(), addr.UnderlayAddr(), ct, srv, nil, nil).Run + /* + node := &pssTestNode{ + Hive: hive, + Pss: ps, + NodeAdapter: nil, + id: nid, + network: net, + trigger: trigger, + ct: ct, + expectC: make(chan []int), + }*/ + srv := func(p Peer) error { - p.Register(&PssMsg{}, ps.HandlePssMsg) - pp.Add(p) + p.Register(&PssMsg{}, self.msgFunc) + self.node.Add(p) p.DisconnectHook(func(err error) { - pp.Remove(p) + self.node.Remove(p) }) return nil } - protocall := Bzz(addr.OverlayAddr(), addr.UnderlayAddr(), ct, srv, nil, nil).Run + + proto := Bzz(self.node.OverlayAddr(), self.node.UnderlayAddr(), ct, srv, nil, nil) + + return []p2p.Protocol{*proto} +} - s := p2ptest.NewProtocolTester(t, NodeId(addr), n, protocall) +func (self *pssTestService) APIs() []rpc.API { + return nil +} - return &pssTester{ - ProtocolTester: s, - ct: ct, - Pss: ps, +/* +func newPssTester(t *testing.T, ps *Pss, addr *peerAddr, numsimnodes int, handlefunc func(interface{}) error, net *simulations.Network, trigger chan *adapters.NodeId) *pssTestNode { + + + + // set up the outer protocol + + + + srv := func(p Peer) error { + p.Register(&PssMsg{}, handlefunc) + node.Add(p) + p.DisconnectHook(func(err error) { + hive.Remove(p) + }) + return nil + } + + node.run = Bzz(addr.OverlayAddr(), addr.UnderlayAddr(), ct, srv, nil, nil).Run + nodeAdapter.Run = node.run + + node.NodeAdapter = adapters.NewSimNode(nid, net) + + + return node +} +*/ + +func makePss(addr []byte) *Pss { + kp := NewKadParams() + kp.MinProxBinSize = 3 + + pp := NewPssParams() + + overlay := NewKademlia(addr, kp) + ps := NewPss(overlay, pp) + overlay.Prune(time.Tick(time.Millisecond * 250)) + return ps +} + +func makeCustomProtocol(name string, version int, ct *protocols.CodeMap, testpeer *pssTestPeer) *p2p.Protocol { + run := func(p *protocols.Peer) error { + log.Trace(fmt.Sprintf("running pss vprotocol on peer %v", p)) + if testpeer == nil { + testpeer = &pssTestPeer{} + } + testpeer.Peer = p + p.Register(&PssTestPayload{}, testpeer.SimpleHandlePssPayload) + err := p.Run() + return err } + return protocols.NewProtocol(name, uint(version), run, ct, nil, nil) +} + +func makeFakeMsg(ps *Pss, ct *protocols.CodeMap, topic PssTopic, senderaddr PeerAddr, content string) PssMsg { + data := PssTestPayload{} + code, found := ct.GetCode(&data) + if !found { + return PssMsg{} + } + + data.Data = content + + rlpbundle, err := makeMsg(code, data) + if err != nil { + return PssMsg{} + } + + pssenv := pssEnvelope{ + SenderOAddr: senderaddr.OverlayAddr(), + SenderUAddr: senderaddr.UnderlayAddr(), + Topic: topic, + TTL: DefaultTTL, + Payload: rlpbundle, + } + pssmsg := PssMsg{ + Payload: pssenv, + } + pssmsg.SetRecipient(ps.Overlay.GetAddr().OverlayAddr()) + + return pssmsg +} + +func makePssHandleForward(ps *Pss) func(msg interface{}) error { + // for the simple check it passes on the message if it's not for us + return func(msg interface{}) error { + pssmsg := msg.(*PssMsg) + if ps.isSelfRecipient(pssmsg) { + log.Trace("pss for us .. yay!") + } else { + log.Trace("passing on pss") + return ps.Forward(pssmsg) + } + return nil + } +} + +func makePssHandleProtocol(ps *Pss) func(msg interface{}) error { + return func(msg interface{}) error { + pssmsg := msg.(*PssMsg) + + if ps.isSelfRecipient(pssmsg) { + log.Trace("pss for us ... let's process!") + env := pssmsg.Payload + umsg := env.Payload // this will be rlp encrypted + f := ps.GetHandler(env.Topic) + if f == nil { + return fmt.Errorf("No registered handler for topic '%s'", env.Topic) + } + nid := adapters.NewNodeId(env.SenderUAddr) + p := p2p.NewPeer(nid.NodeID, fmt.Sprintf("%x", common.ByteLabel(nid.Bytes())), []p2p.Cap{}) + return f(umsg, p, env.SenderOAddr) + } else { + log.Trace("pss was for someone else :'(") + return ps.Forward(pssmsg) + } + return nil + } +} + +// echoes an incoming message +// it comes in through +// Any pointer receiver that has protocols.Peer +func (ptp *pssTestPeer) SimpleHandlePssPayload(msg interface{}) error { + pmsg := msg.(*PssTestPayload) + log.Trace(fmt.Sprintf("PssTestPayloadhandler got message %v", pmsg)) + if pmsg.Data == "ping" { + pmsg.Data = "pong" + log.Trace(fmt.Sprintf("PssTestPayloadhandler reply %v", pmsg)) + ptp.Send(pmsg) + } else if pmsg.Data == "pong" { + ptp.successC <- true + } else { + res, err := strconv.Atoi(pmsg.Data) + if err != nil { + log.Trace(fmt.Sprintf("PssTestPayloadhandlererr %v", err)) + ptp.successC <- false + } else { + log.Trace(fmt.Sprintf("PssTestPayloadhandler sending %d on chan", pmsg)) + ptp.successC <- true + ptp.resultC <- res + } + } + + return nil } diff --git a/swarm/network/test_overlay.go b/swarm/network/test_overlay.go index 8d2b4ba98c..de839bcbab 100644 --- a/swarm/network/test_overlay.go +++ b/swarm/network/test_overlay.go @@ -28,6 +28,13 @@ func (self *testOverlay) Register(nas ...PeerAddr) error { return self.register(nas...) } +func (self *testOverlay) GetAddr() PeerAddr { + return &peerAddr{ + OAddr: self.addr, + UAddr: []byte{}, + } +} + func (self *testOverlay) register(nas ...PeerAddr) error { for _, na := range nas { tna := &testPeerAddr{PeerAddr: na} @@ -95,14 +102,14 @@ func (self *testOverlay) off(po []*testPeerAddr) (nas []PeerAddr) { return nas } -func (self *testOverlay) EachLivePeer(base []byte, o int, f func(Peer, int) bool) { +func (self *testOverlay) EachLivePeer(base []byte, o int, f func(Peer, int, bool) bool) { if base == nil { base = self.addr } for i := o; i < len(self.pos); i++ { for _, na := range self.pos[i] { if na.Peer != nil { - if !f(na.Peer, o) { + if !f(na.Peer, o, false) { return } } diff --git a/swarm/swarm.go b/swarm/swarm.go index 00d92ea450..8e77a37f85 100644 --- a/swarm/swarm.go +++ b/swarm/swarm.go @@ -21,6 +21,7 @@ import ( "context" "crypto/ecdsa" "fmt" + "time" "github.com/ethereum/go-ethereum/accounts/abi/bind" "github.com/ethereum/go-ethereum/common" @@ -31,6 +32,7 @@ import ( "github.com/ethereum/go-ethereum/log" "github.com/ethereum/go-ethereum/node" "github.com/ethereum/go-ethereum/p2p" + "github.com/ethereum/go-ethereum/p2p/adapters" "github.com/ethereum/go-ethereum/p2p/discover" "github.com/ethereum/go-ethereum/rpc" "github.com/ethereum/go-ethereum/swarm/api" @@ -45,18 +47,21 @@ type Swarm struct { config *api.Config // swarm configuration api *api.Api // high level api layer (fs/manifest) dns api.Resolver // DNS registrar - dbAccess *network.DbAccess // access to local chunk db iterator and storage counter + //dbAccess *network.DbAccess // access to local chunk db iterator and storage counter storage storage.ChunkStore // internal access to storage, common interface to cloud storage backends dpa *storage.DPA // distributed preimage archive, the local API to the storage with document level storage/retrieval support - depo network.StorageHandler // remote request handler, interface between bzz protocol and the storage + //depo network.StorageHandler // remote request handler, interface between bzz protocol and the storage cloud storage.CloudStore // procurement, cloud storage backend (can multi-cloud) hive *network.Hive // the logistic manager backend chequebook.Backend // simple blockchain Backend privateKey *ecdsa.PrivateKey corsString string swapEnabled bool + pssEnabled bool + pss *network.Pss lstore *storage.LocalStore // local store, needs to store for releasing resources after node stopped sfs *fuse.SwarmFS // need this to cleanup all the active mounts on node exit + na *adapters.NodeAdapter } type SwarmAPI struct { @@ -75,7 +80,7 @@ func (self *Swarm) API() *SwarmAPI { // creates a new swarm service instance // implements node.Service -func NewSwarm(ctx *node.ServiceContext, backend chequebook.Backend, config *api.Config, swapEnabled, syncEnabled bool, cors string) (self *Swarm, err error) { +func NewSwarm(ctx *node.ServiceContext, backend chequebook.Backend, config *api.Config, swapEnabled, syncEnabled bool, cors string, pssEnabled bool) (self *Swarm, err error) { if bytes.Equal(common.FromHex(config.PublicKey), storage.ZeroKey) { return nil, fmt.Errorf("empty public key") } @@ -89,6 +94,7 @@ func NewSwarm(ctx *node.ServiceContext, backend chequebook.Backend, config *api. backend: backend, privateKey: config.Swap.PrivateKey(), corsString: cors, + pssEnabled: pssEnabled, } log.Debug(fmt.Sprintf("Setting up Swarm service components")) @@ -99,31 +105,35 @@ func NewSwarm(ctx *node.ServiceContext, backend chequebook.Backend, config *api. } // setup local store - log.Debug(fmt.Sprintf("Set up local storage")) - - self.dbAccess = network.NewDbAccess(self.lstore) - log.Debug(fmt.Sprintf("Set up local db access (iterator/counter)")) + //glog.V(logger.Debug).Infof("Set up local storage") + //self.dbAccess = network.NewDbAccess(self.lstore) + glog.V(logger.Debug).Infof("Set up local db access (iterator/counter)") + + kp := network.NewKadParams() + + to := network.NewKademlia( + common.FromHex(config.BzzKey), + kp, + ) // set up the kademlia hive self.hive = network.NewHive( - common.HexToHash(self.config.BzzKey), // key to hive (kademlia base address) config.HiveParams, // configuration parameters - swapEnabled, // SWAP enabled - syncEnabled, // syncronisation enabled + to, ) log.Debug(fmt.Sprintf("Set up swarm network with Kademlia hive")) // setup cloud storage backend - cloud := network.NewForwarder(self.hive) - log.Debug(fmt.Sprintf("-> set swarm forwarder as cloud storage backend")) + //cloud := network.NewForwarder(self.hive) + //glog.V(logger.Debug).Infof("-> set swarm forwarder as cloud storage backend") + // setup cloud storage internal access layer - - self.storage = storage.NewNetStore(hash, self.lstore, cloud, config.StoreParams) - log.Debug(fmt.Sprintf("-> swarm net store shared access layer to Swarm Chunk Store")) + self.storage = storage.NewNetStore(hash, self.lstore, nil, config.StoreParams) + glog.V(logger.Debug).Infof("-> swarm net store shared access layer to Swarm Chunk Store") // set up Depo (storage handler = cloud storage access layer for incoming remote requests) - self.depo = network.NewDepo(hash, self.lstore, self.storage) - log.Debug(fmt.Sprintf("-> REmote Access to CHunks")) + // self.depo = network.NewDepo(hash, self.lstore, self.storage) + // glog.V(logger.Debug).Infof("-> REmote Access to CHunks") // set up DPA, the cloud storage local access layer dpaChunkStore := storage.NewDpaChunkStore(self.lstore, self.storage) @@ -188,13 +198,23 @@ func (self *Swarm) Start(net p2p.Server) error { } log.Warn(fmt.Sprintf("Starting Swarm service")) + + glog.V(logger.Warn).Infof("Starting Swarm service") self.hive.Start( - discover.PubkeyID(&net.PrivateKey.PublicKey), - func() string { return net.ListenAddr }, connectPeer, + func () <-chan time.Time{ + return time.NewTicker(time.Second).C + }, ) + log.Info(fmt.Sprintf("Swarm network started on bzz address: %v", self.hive.Addr())) + if self.pssEnabled { + pssparams := network.NewPssParams() + self.pss = network.NewPss(self.hive.Overlay, pssparams) + glog.V(logger.Info).Infof("Pss started: %v", self.pss) + } + self.dpa.Start() log.Debug(fmt.Sprintf("Swarm DPA started")) @@ -235,11 +255,37 @@ func (self *Swarm) Stop() error { // implements the node.Service interface func (self *Swarm) Protocols() []p2p.Protocol { - proto, err := network.Bzz(self.depo, self.backend, self.hive, self.dbAccess, self.config.Swap, self.config.SyncParams, self.config.NetworkId) - if err != nil { + //proto, err := network.Bzz(self.depo, self.backend, self.hive, self.dbAccess, self.config.Swap, self.config.SyncParams, self.config.NetworkId) + ct := network.BzzCodeMap() + for _, m := range network.DiscoveryMsgs { + ct.Register(m) + } + if self.pssEnabled { + ct.Register(&network.PssMsg{}) + } + + srv := func(p network.Peer) error { + if self.pssEnabled { + //p.Register(&PssMsg{}, self.pssFunc) + glog.V(logger.Warn).Infof("pss is enabled, but handler not yet implemented - it won't work yet, sorry") + } + self.hive.Add(p) + p.DisconnectHook(func(err error) { + self.hive.Remove(p) + }) return nil } - return []p2p.Protocol{proto} + + proto := network.Bzz( + self.hive.Overlay.GetAddr().OverlayAddr(), + self.hive.Overlay.GetAddr().UnderlayAddr(), + ct, + srv, + nil, + nil, + ) + + return []p2p.Protocol{*proto} } // implements node.Service @@ -302,7 +348,7 @@ func (self *Swarm) SetChequebook(ctx context.Context) error { } log.Info(fmt.Sprintf("new chequebook set (%v): saving config file, resetting all connections in the hive", self.config.Swap.Contract.Hex())) self.config.Save() - self.hive.DropAll() + //self.hive.DropAll() return nil } @@ -332,7 +378,7 @@ func NewLocalSwarm(datadir, port string) (self *Swarm, err error) { return } - + // serialisable info about swarm type Info struct { *api.Config