diff --git a/cmd/geth/config.go b/cmd/geth/config.go index 27490c4048..7d53ce4044 100644 --- a/cmd/geth/config.go +++ b/cmd/geth/config.go @@ -104,7 +104,7 @@ func defaultNodeConfig() node.Config { cfg.Name = clientIdentifier cfg.Version = params.VersionWithCommit(gitCommit) cfg.HTTPModules = append(cfg.HTTPModules, "eth", "shh") - cfg.WSModules = append(cfg.WSModules, "eth", "shh") + cfg.WSModules = append(cfg.WSModules, "eth", "shh", "pss") cfg.IPCPath = "geth.ipc" return cfg } diff --git a/p2p/simulations/adapters/inproc.go b/p2p/simulations/adapters/inproc.go index 48d7c17301..0d22b4f56f 100644 --- a/p2p/simulations/adapters/inproc.go +++ b/p2p/simulations/adapters/inproc.go @@ -17,11 +17,14 @@ package adapters import ( + "crypto/rand" "errors" "fmt" "math" "net" + "os" "sync" + "syscall" "github.com/ethereum/go-ethereum/event" "github.com/ethereum/go-ethereum/log" @@ -31,9 +34,15 @@ import ( "github.com/ethereum/go-ethereum/rpc" ) +const ( + socketReadBuffer = 5000 * 1024 + socketWriteBuffer = 5000 * 1024 +) + // SimAdapter is a NodeAdapter which creates in-memory simulation nodes and -// connects them using in-memory net.Pipe connections +// connects them using net.Pipe or OS socket connections type SimAdapter struct { + pipe func() (net.Conn, net.Conn, error) mtx sync.RWMutex nodes map[discover.NodeID]*SimNode services map[string]ServiceFunc @@ -42,8 +51,30 @@ type SimAdapter struct { // NewSimAdapter creates a SimAdapter which is capable of running in-memory // simulation nodes running any of the given services (the services to run on a // particular node are passed to the NewNode function in the NodeConfig) +// the adapter uses a net.Pipe for in-memory simulated network connections func NewSimAdapter(services map[string]ServiceFunc) *SimAdapter { return &SimAdapter{ + pipe: netPipe, + nodes: make(map[discover.NodeID]*SimNode), + services: services, + } +} + +// NewSocketAdapter creates a SimAdapter which is capable of running in-memory +// simulation nodes running any of the given services (the services to run on a +// particular node are passed to the NewNode function in the NodeConfig) +// the adapter uses a OS socketpairs for in-memory simulated network connections +func NewSocketAdapter(services map[string]ServiceFunc) *SimAdapter { + return &SimAdapter{ + pipe: socketPipe, + nodes: make(map[discover.NodeID]*SimNode), + services: services, + } +} + +func NewTCPAdapter(services map[string]ServiceFunc) *SimAdapter { + return &SimAdapter{ + pipe: tcpPipe, nodes: make(map[discover.NodeID]*SimNode), services: services, } @@ -102,7 +133,7 @@ func (s *SimAdapter) NewNode(config *NodeConfig) (Node, error) { } // Dial implements the p2p.NodeDialer interface by connecting to the node using -// an in-memory net.Pipe connection +// an in-memory net.Pipe or OS socket connection func (s *SimAdapter) Dial(dest *discover.Node) (conn net.Conn, err error) { node, ok := s.GetNode(dest.ID) if !ok { @@ -112,7 +143,14 @@ func (s *SimAdapter) Dial(dest *discover.Node) (conn net.Conn, err error) { if srv == nil { return nil, fmt.Errorf("node not running: %s", dest.ID) } - pipe1, pipe2 := net.Pipe() + // SimAdapter.pipe is either net.Pipe (NewSimAdapter) or socketPipe (NewSocketAdapter) + pipe1, pipe2, err := s.pipe() + if err != nil { + return nil, err + } + // this is simulated 'listening' + // asynchronously call the dialed destintion node's p2p server + // to set up connection on the 'listening' side go srv.SetupConn(pipe1, 0, nil) return pipe2, nil } @@ -140,7 +178,7 @@ func (s *SimAdapter) GetNode(id discover.NodeID) (*SimNode, bool) { } // SimNode is an in-memory simulation node which connects to other nodes using -// an in-memory net.Pipe connection (see SimAdapter.Dial), running devp2p +// net.Pipe or OS socket connection (see SimAdapter.Dial), running devp2p // protocols directly over that pipe type SimNode struct { lock sync.RWMutex @@ -314,3 +352,117 @@ func (self *SimNode) NodeInfo() *p2p.NodeInfo { } return server.NodeInfo() } + +// socketPipe creates an in process full duplex pipe based on OS sockets +// credit to @lmars & Flynn +// https://github.com/flynn/flynn/blob/master/host/containerinit/init.go#L743-L749 +// using this in large simulations requires raising OS's max open file limit +func socketPipe() (net.Conn, net.Conn, error) { + pair, err := syscall.Socketpair(syscall.AF_UNIX, syscall.SOCK_STREAM, 0) + if err != nil { + return nil, nil, err + } + nameb := make([]byte, 8) + _, err = rand.Read(nameb) + if err != nil { + return nil, nil, err + } + f1 := os.NewFile(uintptr(pair[0]), string(nameb)+".out") + f2 := os.NewFile(uintptr(pair[1]), string(nameb)+".in") + pipe1, err := net.FileConn(f1) + if err != nil { + return nil, nil, err + } + pipe2, err := net.FileConn(f2) + if err != nil { + return nil, nil, err + } + + err = setSocketBuffer(pipe1) + if err != nil { + return nil, nil, err + } + + err = setSocketBuffer(pipe2) + if err != nil { + return nil, nil, err + } + + return pipe1, pipe2, nil +} + +func setSocketBuffer(conn net.Conn) error { + switch v := conn.(type) { + case *net.UnixConn: + err := v.SetReadBuffer(socketReadBuffer) + if err != nil { + return err + } + err = v.SetWriteBuffer(socketWriteBuffer) + if err != nil { + return err + } + } + return nil +} + +// netPipe wraps net.Pipe in a signature returning an error +func netPipe() (net.Conn, net.Conn, error) { + p1, p2 := net.Pipe() + return p1, p2, nil +} + +// tcpPipe creates an in process full duplex pipe based on a localhost TCP socket +func tcpPipe() (net.Conn, net.Conn, error) { + type result struct { + conn net.Conn + err error + } + + cl := make(chan result) + cd := make(chan result) + + start := make(chan net.Addr) + + go func(res chan result, start chan net.Addr) { + // resolve + addr, err := net.ResolveTCPAddr("tcp", "localhost:0") + if err != nil { + res <- result{err: err} + return + } + // listen + l, err := net.ListenTCP("tcp", addr) + if err != nil { + res <- result{err: err} + return + } + start <- l.Addr() + c, err := l.AcceptTCP() + if err != nil { + res <- result{err: err} + return + } + res <- result{conn: c} + }(cl, start) + + go func(res chan result, start chan net.Addr) { + addr := <-start + c, err := net.DialTCP("tcp", nil, addr.(*net.TCPAddr)) + if err != nil { + res <- result{err: err} + return + } + res <- result{conn: c} + }(cd, start) + + a := <-cl + if a.err != nil { + return nil, nil, a.err + } + b := <-cd + if b.err != nil { + return nil, nil, b.err + } + return a.conn, b.conn, nil +} diff --git a/p2p/simulations/adapters/inproc_test.go b/p2p/simulations/adapters/inproc_test.go new file mode 100644 index 0000000000..76be7228d1 --- /dev/null +++ b/p2p/simulations/adapters/inproc_test.go @@ -0,0 +1,344 @@ +// Copyright 2017 The go-ethereum Authors +// This file is part of the go-ethereum library. +// +// The go-ethereum library is free software: you can redistribute it and/or modify +// it under the terms of the GNU Lesser General Public License as published by +// the Free Software Foundation, either version 3 of the License, or +// (at your option) any later version. +// +// The go-ethereum library is distributed in the hope that it will be useful, +// but WITHOUT ANY WARRANTY; without even the implied warranty of +// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the +// GNU Lesser General Public License for more details. +// +// You should have received a copy of the GNU Lesser General Public License +// along with the go-ethereum library. If not, see . + +package adapters + +import ( + "bytes" + "encoding/binary" + "fmt" + "testing" + "time" +) + +func TestSocketPipe(t *testing.T) { + c1, c2, _ := socketPipe() + + done := make(chan struct{}) + + go func() { + msgs := 20 + size := 8 + for i := 0; i < msgs; i++ { + msg := make([]byte, size) + _ = binary.PutUvarint(msg, uint64(i)) + + _, err := c1.Write(msg) + if err != nil { + t.Fatal(err) + } + } + + for i := 0; i < msgs; i++ { + msg := make([]byte, size) + _ = binary.PutUvarint(msg, uint64(i)) + + out := make([]byte, size) + _, err := c2.Read(out) + if err != nil { + t.Fatal(err) + } + + if bytes.Compare(msg, out) != 0 { + t.Fatalf("expected %#v, got %#v", msg, out) + } + } + done <- struct{}{} + }() + + select { + case <-done: + case <-time.After(1 * time.Second): + t.Fatal("test timeout") + } +} + +func TestSocketPipeBidirections(t *testing.T) { + c1, c2, _ := socketPipe() + + done := make(chan struct{}) + + go func() { + msgs := 100 + size := 4 + for i := 0; i < msgs; i++ { + msg := []byte(`ping`) + + _, err := c1.Write(msg) + if err != nil { + t.Fatal(err) + } + } + + for i := 0; i < msgs; i++ { + out := make([]byte, size) + _, err := c2.Read(out) + if err != nil { + t.Fatal(err) + } + + if bytes.Compare(out, []byte(`ping`)) == 0 { + msg := []byte(`pong`) + _, err := c2.Write(msg) + if err != nil { + t.Fatal(err) + } + } + } + + for i := 0; i < msgs; i++ { + expected := []byte(`pong`) + + out := make([]byte, size) + _, err := c1.Read(out) + if err != nil { + t.Fatal(err) + } + + if bytes.Compare(out, expected) != 0 { + t.Fatalf("expected %#v, got %#v", expected, out) + } + } + + done <- struct{}{} + }() + + select { + case <-done: + case <-time.After(1 * time.Second): + t.Fatal("test timeout") + } +} + +func TestTcpPipe(t *testing.T) { + c1, c2, _ := tcpPipe() + + done := make(chan struct{}) + + go func() { + msgs := 50 + size := 1024 + for i := 0; i < msgs; i++ { + msg := make([]byte, size) + _ = binary.PutUvarint(msg, uint64(i)) + + _, err := c1.Write(msg) + if err != nil { + t.Fatal(err) + } + } + + for i := 0; i < msgs; i++ { + msg := make([]byte, size) + _ = binary.PutUvarint(msg, uint64(i)) + + out := make([]byte, size) + _, err := c2.Read(out) + if err != nil { + t.Fatal(err) + } + + if bytes.Compare(msg, out) != 0 { + t.Fatalf("expected %#v, got %#v", msg, out) + } + } + done <- struct{}{} + }() + + select { + case <-done: + case <-time.After(1 * time.Second): + t.Fatal("test timeout") + } +} + +func TestTcpPipeBidirections(t *testing.T) { + c1, c2, _ := tcpPipe() + + done := make(chan struct{}) + + go func() { + msgs := 50 + size := 7 + for i := 0; i < msgs; i++ { + msg := []byte(fmt.Sprintf("ping %02d", i)) + + _, err := c1.Write(msg) + if err != nil { + t.Fatal(err) + } + } + + for i := 0; i < msgs; i++ { + expected := []byte(fmt.Sprintf("ping %02d", i)) + + out := make([]byte, size) + _, err := c2.Read(out) + if err != nil { + t.Fatal(err) + } + + if bytes.Compare(expected, out) != 0 { + t.Fatalf("expected %#v, got %#v", out, expected) + } else { + msg := []byte(fmt.Sprintf("pong %02d", i)) + _, err := c2.Write(msg) + if err != nil { + t.Fatal(err) + } + } + } + + for i := 0; i < msgs; i++ { + expected := []byte(fmt.Sprintf("pong %02d", i)) + + out := make([]byte, size) + _, err := c1.Read(out) + if err != nil { + t.Fatal(err) + } + + if bytes.Compare(expected, out) != 0 { + t.Fatalf("expected %#v, got %#v", out, expected) + } + } + done <- struct{}{} + }() + + select { + case <-done: + case <-time.After(1 * time.Second): + t.Fatal("test timeout") + } +} + +func TestNetPipe(t *testing.T) { + c1, c2, _ := netPipe() + + done := make(chan struct{}) + + go func() { + msgs := 50 + size := 1024 + // netPipe is blocking, so writes are emitted asynchronously + go func() { + for i := 0; i < msgs; i++ { + msg := make([]byte, size) + _ = binary.PutUvarint(msg, uint64(i)) + + _, err := c1.Write(msg) + if err != nil { + t.Fatal(err) + } + } + }() + + for i := 0; i < msgs; i++ { + msg := make([]byte, size) + _ = binary.PutUvarint(msg, uint64(i)) + + out := make([]byte, size) + _, err := c2.Read(out) + if err != nil { + t.Fatal(err) + } + + if bytes.Compare(msg, out) != 0 { + t.Fatalf("expected %#v, got %#v", msg, out) + } + } + + done <- struct{}{} + }() + + select { + case <-done: + case <-time.After(1 * time.Second): + t.Fatal("test timeout") + } +} + +func TestNetPipeBidirections(t *testing.T) { + c1, c2, _ := netPipe() + + done := make(chan struct{}) + + go func() { + msgs := 1000 + size := 8 + pingTemplate := "ping %03d" + pongTemplate := "pong %03d" + + // netPipe is blocking, so writes are emitted asynchronously + go func() { + for i := 0; i < msgs; i++ { + msg := []byte(fmt.Sprintf(pingTemplate, i)) + + _, err := c1.Write(msg) + if err != nil { + t.Fatal(err) + } + } + }() + + // netPipe is blocking, so reads for pong are emitted asynchronously + go func() { + for i := 0; i < msgs; i++ { + expected := []byte(fmt.Sprintf(pongTemplate, i)) + + out := make([]byte, size) + _, err := c1.Read(out) + if err != nil { + t.Fatal(err) + } + + if bytes.Compare(expected, out) != 0 { + t.Fatalf("expected %#v, got %#v", expected, out) + } + } + + done <- struct{}{} + }() + + // expect to read pings, and respond with pongs to the alternate connection + for i := 0; i < msgs; i++ { + expected := []byte(fmt.Sprintf(pingTemplate, i)) + + out := make([]byte, size) + _, err := c2.Read(out) + if err != nil { + t.Fatal(err) + } + + if bytes.Compare(expected, out) != 0 { + t.Fatalf("expected %#v, got %#v", expected, out) + } else { + msg := []byte(fmt.Sprintf(pongTemplate, i)) + + _, err := c2.Write(msg) + if err != nil { + t.Fatal(err) + } + } + } + }() + + select { + case <-done: + case <-time.After(1 * time.Second): + t.Fatal("test timeout") + } +} diff --git a/rpc/client.go b/rpc/client.go index 8aa84ec982..2a66bb3b0e 100644 --- a/rpc/client.go +++ b/rpc/client.go @@ -60,7 +60,7 @@ const ( // The approach taken here is to maintain a per-subscription linked list buffer // shrinks on demand. If the buffer reaches the size below, the subscription is // dropped. - maxClientSubscriptionBuffer = 8000 + maxClientSubscriptionBuffer = 20000 ) // BatchElem is an element in a batch request. diff --git a/swarm/api/testapi.go b/swarm/api/testapi.go index 6631196c17..7d59a50fd2 100644 --- a/swarm/api/testapi.go +++ b/swarm/api/testapi.go @@ -29,18 +29,18 @@ 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) 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/network/hive.go b/swarm/network/hive.go index 2a180d61f1..d72d7c5e7c 100644 --- a/swarm/network/hive.go +++ b/swarm/network/hive.go @@ -79,7 +79,7 @@ func NewHiveParams() *HiveParams { type Hive struct { *HiveParams // settings Overlay // the overlay connectiviy driver - store StateStore // storage interface to save peers across sessions + Store StateStore // storage interface to save peers across sessions addPeer func(*discover.Node) // server callback to connect to a peer // bookkeeping lock sync.Mutex @@ -94,7 +94,7 @@ func NewHive(params *HiveParams, overlay Overlay, store StateStore) *Hive { return &Hive{ HiveParams: params, Overlay: overlay, - store: store, + Store: store, } } @@ -104,7 +104,7 @@ func NewHive(params *HiveParams, overlay Overlay, store StateStore) *Hive { func (h *Hive) Start(server *p2p.Server) error { log.Trace(fmt.Sprintf("%08x hive starting", h.BaseAddr()[:4])) // if state store is specified, load peers to prepopulate the overlay address book - if h.store != nil { + if h.Store != nil { if err := h.loadPeers(); err != nil { return err } @@ -122,7 +122,7 @@ func (h *Hive) Start(server *p2p.Server) error { func (h *Hive) Stop() error { log.Info(fmt.Sprintf("%08x hive stopping, saving peers", h.BaseAddr()[:4])) h.ticker.Stop() - if h.store != nil { + if h.Store != nil { return h.savePeers() } log.Info(fmt.Sprintf("%08x hive stopped, dropping peers", h.BaseAddr()[:4])) @@ -196,7 +196,7 @@ func ToAddr(pa OverlayPeer) *BzzAddr { // loadPeers, savePeer implement persistence callback/ func (h *Hive) loadPeers() error { - data, err := h.store.Load("peers") + data, err := h.Store.Load("peers") if err != nil { return err } @@ -233,7 +233,7 @@ func (h *Hive) savePeers() error { if err != nil { return fmt.Errorf("could not encode peers: %v", err) } - if err := h.store.Save("peers", data); err != nil { + if err := h.Store.Save("peers", data); err != nil { return fmt.Errorf("could not save peers: %v", err) } return nil diff --git a/swarm/network/simulations/discovery/discovery_test.go b/swarm/network/simulations/discovery/discovery_test.go index e90a80b0be..fc8d6f70ca 100644 --- a/swarm/network/simulations/discovery/discovery_test.go +++ b/swarm/network/simulations/discovery/discovery_test.go @@ -100,7 +100,8 @@ func TestDiscoverySimulationSimAdapter(t *testing.T) { } func testDiscoverySimulationSimAdapter(t *testing.T, nodes, conns int) { - testDiscoverySimulation(t, nodes, conns, adapters.NewSimAdapter(services)) + testDiscoverySimulation(t, nodes, conns, adapters.NewSocketAdapter(services)) + // testDiscoverySimulation(t, nodes, conns, adapters.NewSimAdapter(services)) } func testDiscoverySimulation(t *testing.T, nodes, conns int, adapter adapters.NodeAdapter) { @@ -310,7 +311,7 @@ func newService(ctx *adapters.ServiceContext) (node.Service, error) { kad := network.NewKademlia(addr.Over(), kp) hp := network.NewHiveParams() - hp.KeepAliveInterval = 500 * time.Millisecond + hp.KeepAliveInterval = 200 * time.Millisecond config := &network.BzzConfig{ OverlayAddr: addr.Over(), diff --git a/swarm/pss/protocol.go b/swarm/pss/protocol.go index ee2dcfea43..6c5c289559 100644 --- a/swarm/pss/protocol.go +++ b/swarm/pss/protocol.go @@ -5,11 +5,12 @@ package pss import ( "bytes" "fmt" + "time" + "github.com/ethereum/go-ethereum/log" "github.com/ethereum/go-ethereum/p2p" "github.com/ethereum/go-ethereum/p2p/protocols" "github.com/ethereum/go-ethereum/rlp" - "time" ) const ( @@ -197,8 +198,6 @@ func ToP2pMsg(msg []byte) (p2p.Msg, error) { // to link the peer to. // The key must exist in the pss store prior to adding the peer. func (self *Protocol) AddPeer(p *p2p.Peer, run func(*p2p.Peer, p2p.MsgReadWriter) error, topic Topic, asymmetric bool, key string) (p2p.MsgReadWriter, error) { - self.Pss.lock.Lock() - defer self.Pss.lock.Unlock() rw := &PssReadWriter{ Pss: self.Pss, rw: make(chan p2p.Msg), @@ -212,14 +211,18 @@ func (self *Protocol) AddPeer(p *p2p.Peer, run func(*p2p.Peer, p2p.MsgReadWriter rw.sendFunc = self.Pss.SendSym } if asymmetric { + self.Pss.pubKeyPoolMu.Lock() if _, ok := self.Pss.pubKeyPool[key]; !ok { return nil, fmt.Errorf("asym key does not exist: %s", key) } + self.Pss.pubKeyPoolMu.Unlock() self.pubKeyRWPool[key] = rw } else { + self.Pss.symKeyPoolMu.Lock() if _, ok := self.Pss.symKeyPool[key]; !ok { return nil, fmt.Errorf("symkey does not exist: %s", key) } + self.Pss.symKeyPoolMu.Unlock() self.symKeyRWPool[key] = rw } go func() { diff --git a/swarm/pss/pss.go b/swarm/pss/pss.go index 4148e9a9e4..507b4ab655 100644 --- a/swarm/pss/pss.go +++ b/swarm/pss/pss.go @@ -94,25 +94,29 @@ type Pss struct { auxAPIs []rpc.API // builtins (handshake, test) can add APIs // sending and forwarding - fwdPool map[string]*protocols.Peer // keep track of all peers sitting on the pssmsg routing layer + fwdPool map[string]*protocols.Peer // keep track of all peers sitting on the pssmsg routing layer + fwdPoolMu sync.Mutex 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 (not implemented) + fwdCacheMu sync.Mutex + cacheTTL time.Duration // how long to keep messages in fwdCache (not implemented) msgTTL time.Duration paddingByteSize int capstring string // keys and peers pubKeyPool map[string]map[Topic]*pssPeer // mapping of hex public keys to peer address by topic. + pubKeyPoolMu sync.Mutex symKeyPool map[string]map[Topic]*pssPeer // mapping of symkeyids to peer address by topic. - symKeyDecryptCache []*string // fast lookup of symkeys recently used for decryption; last used is on top of stack - symKeyDecryptCacheCursor int // modular cursor pointing to last used, wraps on symKeyDecryptCache array - symKeyDecryptCacheCapacity int // max amount of symkeys to keep. + symKeyPoolMu sync.Mutex + symKeyDecryptCache []*string // fast lookup of symkeys recently used for decryption; last used is on top of stack + symKeyDecryptCacheCursor int // modular cursor pointing to last used, wraps on symKeyDecryptCache array + symKeyDecryptCacheCapacity int // max amount of symkeys to keep. // message handling - handlers map[Topic]map[*Handler]bool // topic and version based pss payload handlers. See pss.Handle() + handlers map[Topic]map[*Handler]bool // topic and version based pss payload handlers. See pss.Handle() + handlersMu sync.Mutex // process - lock sync.Mutex quitC chan struct{} } @@ -197,7 +201,9 @@ func (self *Pss) Protocols() []p2p.Protocol { func (self *Pss) Run(p *p2p.Peer, rw p2p.MsgReadWriter) error { pp := protocols.NewPeer(p, rw, pssSpec) + self.fwdPoolMu.Lock() self.fwdPool[p.Info().ID] = pp + self.fwdPoolMu.Unlock() return pp.Run(self.handlePssMsg) } @@ -246,8 +252,8 @@ func (self *Pss) PublicKey() *ecdsa.PublicKey { // Returns a deregister function which needs to be called to // deregister the handler, func (self *Pss) Register(topic *Topic, handler Handler) func() { - self.lock.Lock() - defer self.lock.Unlock() + self.handlersMu.Lock() + defer self.handlersMu.Unlock() handlers := self.handlers[*topic] if handlers == nil { handlers = make(map[*Handler]bool) @@ -257,8 +263,8 @@ func (self *Pss) Register(topic *Topic, handler Handler) func() { return func() { self.deregister(topic, &handler) } } func (self *Pss) deregister(topic *Topic, h *Handler) { - self.lock.Lock() - defer self.lock.Unlock() + self.handlersMu.Lock() + defer self.handlersMu.Unlock() handlers := self.handlers[*topic] if len(handlers) == 1 { delete(self.handlers, *topic) @@ -269,8 +275,8 @@ func (self *Pss) deregister(topic *Topic, h *Handler) { // get all registered handlers for respective topics func (self *Pss) getHandlers(topic Topic) map[*Handler]bool { - self.lock.Lock() - defer self.lock.Unlock() + self.handlersMu.Lock() + defer self.handlersMu.Unlock() return self.handlers[topic] } @@ -374,8 +380,6 @@ func (self *Pss) isSelfPossibleRecipient(msg *PssMsg) bool { // The value in `address` will be used as a routing hint for the // public key / topic association func (self *Pss) SetPeerPublicKey(pubkey *ecdsa.PublicKey, topic Topic, address *PssAddress) error { - self.lock.Lock() - defer self.lock.Unlock() pubkeybytes := crypto.FromECDSAPub(pubkey) if len(pubkeybytes) == 0 { return fmt.Errorf("invalid public key: %v", pubkey) @@ -384,10 +388,12 @@ func (self *Pss) SetPeerPublicKey(pubkey *ecdsa.PublicKey, topic Topic, address psp := &pssPeer{ address: address, } + self.pubKeyPoolMu.Lock() if _, ok := self.pubKeyPool[pubkeyid]; ok == false { self.pubKeyPool[pubkeyid] = make(map[Topic]*pssPeer) } self.pubKeyPool[pubkeyid][topic] = psp + self.pubKeyPoolMu.Unlock() log.Trace("added pubkey", "pubkeyid", pubkeyid, "topic", topic, "address", common.ToHex(*address)) return nil } @@ -427,15 +433,15 @@ func (self *Pss) SetSymmetricKey(key []byte, topic Topic, address *PssAddress, a // to the collection of keys used to attempt symmetric decryption of // incoming messages func (self *Pss) addSymmetricKeyToPool(keyid string, topic Topic, address *PssAddress, addtocache bool) { - self.lock.Lock() - defer self.lock.Unlock() psp := &pssPeer{ address: address, } + self.symKeyPoolMu.Lock() if _, ok := self.symKeyPool[keyid]; !ok { self.symKeyPool[keyid] = make(map[Topic]*pssPeer) } self.symKeyPool[keyid][topic] = psp + self.symKeyPoolMu.Unlock() if addtocache { self.symKeyDecryptCacheCursor++ self.symKeyDecryptCache[self.symKeyDecryptCacheCursor%cap(self.symKeyDecryptCache)] = &keyid @@ -476,7 +482,9 @@ func (self *Pss) processSym(envelope *whisper.Envelope) (*whisper.ReceivedMessag if !recvmsg.Validate() { return nil, "", nil, fmt.Errorf("symmetrically encrypted message has invalid signature or is corrupt") } + self.symKeyPoolMu.Lock() from := self.symKeyPool[*symkeyid][Topic(envelope.Topic)].address + self.symKeyPoolMu.Unlock() self.symKeyDecryptCacheCursor++ self.symKeyDecryptCache[self.symKeyDecryptCacheCursor%cap(self.symKeyDecryptCache)] = symkeyid return recvmsg, *symkeyid, from, nil @@ -501,9 +509,11 @@ func (self *Pss) processAsym(envelope *whisper.Envelope) (*whisper.ReceivedMessa } pubkeyid := common.ToHex(crypto.FromECDSAPub(recvmsg.Src)) var from *PssAddress + self.pubKeyPoolMu.Lock() if self.pubKeyPool[pubkeyid][Topic(envelope.Topic)] != nil { from = self.pubKeyPool[pubkeyid][Topic(envelope.Topic)].address } + self.pubKeyPoolMu.Unlock() return recvmsg, pubkeyid, from, nil } @@ -533,8 +543,10 @@ func (self *Pss) cleanKeys() (count int) { } } for _, topic := range expiredtopics { + self.symKeyPoolMu.Lock() delete(self.symKeyPool[keyid], topic) log.Trace("symkey cleanup deletion", "symkeyid", keyid, "topic", topic, "val", self.symKeyPool[keyid]) + self.symKeyPoolMu.Unlock() count++ } } @@ -553,7 +565,9 @@ func (self *Pss) SendSym(symkeyid string, topic Topic, msg []byte) error { if err != nil { return fmt.Errorf("missing valid send symkey %s: %v", symkeyid, err) } + self.symKeyPoolMu.Lock() psp, ok := self.symKeyPool[symkeyid][topic] + self.symKeyPoolMu.Unlock() if !ok { return fmt.Errorf("invalid topic '%s' for symkey '%s'", topic, symkeyid) } else if psp.address == nil { @@ -567,18 +581,21 @@ func (self *Pss) SendSym(symkeyid string, topic Topic, msg []byte) error { // // Fails if the key id does not match any in of the stored public keys func (self *Pss) SendAsym(pubkeyid string, topic Topic, msg []byte) error { - //pubkey := self.pubKeyIndex[pubkeyid] pubkey := crypto.ToECDSAPub(common.FromHex(pubkeyid)) if pubkey == nil { return fmt.Errorf("Invalid public key id %x", pubkey) } + self.pubKeyPoolMu.Lock() psp, ok := self.pubKeyPool[pubkeyid][topic] + self.pubKeyPoolMu.Unlock() if !ok { return fmt.Errorf("invalid topic '%s' for pubkey '%s'", topic, pubkeyid) } else if psp.address == nil { return fmt.Errorf("no address hint for topic '%s' pubkey '%s'", topic, pubkeyid) } - self.send(*psp.address, topic, msg, true, common.FromHex(pubkeyid)) + go func() { + self.send(*psp.address, topic, msg, true, common.FromHex(pubkeyid)) + }() return nil } @@ -688,11 +705,12 @@ func (self *Pss) forward(msg *PssMsg) error { return true } // attempt to send the message - err := pp.Send(msg) - if err != nil { - log.Debug(fmt.Sprintf("%v: failed forwarding: %v", sendMsg, err)) - return true - } + go func() { + err := pp.Send(msg) + if err != nil { + log.Debug(fmt.Sprintf("%v: failed forwarding: %v", sendMsg, err)) + } + }() log.Trace(fmt.Sprintf("%v: successfully forwarded", sendMsg)) sent++ // continue forwarding if: @@ -728,8 +746,8 @@ func (self *Pss) forward(msg *PssMsg) error { // add a message to the cache func (self *Pss) addFwdCache(digest pssDigest) error { - self.lock.Lock() - defer self.lock.Unlock() + self.fwdCacheMu.Lock() + defer self.fwdCacheMu.Unlock() var entry pssCacheEntry var ok bool if entry, ok = self.fwdCache[digest]; !ok { @@ -742,8 +760,8 @@ func (self *Pss) addFwdCache(digest pssDigest) error { // check if message is in the cache func (self *Pss) checkFwdCache(addr []byte, digest pssDigest) bool { - self.lock.Lock() - defer self.lock.Unlock() + self.fwdCacheMu.Lock() + defer self.fwdCacheMu.Unlock() entry, ok := self.fwdCache[digest] if ok { if entry.expiresAt.After(time.Now()) { diff --git a/swarm/pss/pss_test.go b/swarm/pss/pss_test.go index 1e61c03490..57d4a79170 100644 --- a/swarm/pss/pss_test.go +++ b/swarm/pss/pss_test.go @@ -59,12 +59,14 @@ var ( var services = newServices() func init() { - flag.Parse() rand.Seed(time.Now().Unix()) adapters.RegisterServices(services) + initTest() +} +func initTest() { initOnce.Do( func() { loglevel := log.LvlInfo @@ -609,94 +611,84 @@ func testAsymSend(t *testing.T) { } } -type networkParams struct { - snapshotFile string - numMessages int - addressSize int - adapterType string - messageDelay int +type Job struct { + Msg []byte + SendNode discover.NodeID + RecvNode discover.NodeID } -func (n *networkParams) String() string { - return fmt.Sprintf(":%s:%d:%d:%d:%s", n.snapshotFile, n.numMessages, n.addressSize, n.messageDelay, n.adapterType) +func worker(id int, jobs <-chan Job, rpcs map[discover.NodeID]*rpc.Client, pubkeys map[discover.NodeID]string, topic string) { + for j := range jobs { + rpcs[j.SendNode].Call(nil, "pss_sendAsym", pubkeys[j.RecvNode], topic, hexutil.Encode(j.Msg)) + } } -// Tests random message sending in network snapshots -// // params in run name: -// #nodes/#msgs/#addrbytes/adaptertype +// nodes/msgs/addrbytes/adaptertype // if adaptertype is exec uses execadapter, simadapter otherwise func TestNetwork(t *testing.T) { - var tests []*networkParams - if *snapshotflag != "" { - if *addresssizeflag < 0 || *addresssizeflag > 32 { - t.Fatal("invalid address size") - } - _, err := os.Stat(*snapshotflag) - if err != nil { - t.Fatal(err) - } - tests = append(tests, &networkParams{ - snapshotFile: *snapshotflag, - numMessages: *messagesflag, - addressSize: *addresssizeflag, - adapterType: *adaptertypeflag, - messageDelay: *messagedelayflag, - }) - } else { - tests = append(tests, &networkParams{ - snapshotFile: "testdata/snapshot_8.json", - numMessages: 2, - addressSize: 2, - adapterType: "sim", - messageDelay: 1000, - }) - } - for _, p := range tests { - t.Run(p.String(), testNetwork) - } + t.Run("3/2000/4/sock", testNetwork) + t.Run("4/2000/4/sock", testNetwork) + t.Run("8/2000/4/sock", testNetwork) + t.Run("16/2000/4/sock", testNetwork) + t.Run("32/2000/4/sock", testNetwork) + t.Run("64/2000/4/sim", testNetwork) } func testNetwork(t *testing.T) { - - lock := &sync.Mutex{} type msgnotifyC struct { id discover.NodeID msgIdx int } - paramstring := strings.Split(t.Name(), ":") + paramstring := strings.Split(t.Name(), "/") + nodecount, _ := strconv.ParseInt(paramstring[1], 10, 0) msgcount, _ := strconv.ParseInt(paramstring[2], 10, 0) addrsize, _ := strconv.ParseInt(paramstring[3], 10, 0) - messagedelaymax, _ := strconv.ParseInt(paramstring[4], 10, 0) - log.Info("network test", "snapshot", paramstring[1], "msgcount", msgcount, "addrhintsize", addrsize, "messagedelay", messagedelaymax) + adapter := paramstring[4] + + log.Info("network test", "nodecount", nodecount, "msgcount", msgcount, "addrhintsize", addrsize) + + nodes := make([]discover.NodeID, nodecount) + bzzaddrs := make(map[discover.NodeID]string, nodecount) + rpcs := make(map[discover.NodeID]*rpc.Client, nodecount) + pubkeys := make(map[discover.NodeID]string, nodecount) sentmsgs := make([][]byte, msgcount) recvmsgs := make([]bool, msgcount) + nodemsgcount := make(map[discover.NodeID]int, nodecount) + trigger := make(chan discover.NodeID) - var adapter adapters.NodeAdapter - if paramstring[5] == "exec" { + var a adapters.NodeAdapter + if adapter == "exec" { dirname, err := ioutil.TempDir(".", "") - defer os.RemoveAll(dirname) if err != nil { t.Fatal(err) } - adapter = adapters.NewExecAdapter(dirname) - } else { - adapter = adapters.NewSimAdapter(services) + a = adapters.NewExecAdapter(dirname) + } else if adapter == "sock" { + a = adapters.NewSocketAdapter(services) + } else if adapter == "tcp" { + a = adapters.NewTCPAdapter(services) + } else if adapter == "sim" { + a = adapters.NewSimAdapter(services) } - net := simulations.NewNetwork(adapter, &simulations.NetworkConfig{ + net := simulations.NewNetwork(a, &simulations.NetworkConfig{ ID: "0", }) defer net.Shutdown() - f, err := os.Open(paramstring[1]) + f, err := os.Open(fmt.Sprintf("testdata/snapshot_%d.json", nodecount)) + if err != nil { + t.Fatal(err) + } + jsonbyte, err := ioutil.ReadAll(f) if err != nil { t.Fatal(err) } var snap simulations.Snapshot - err = json.NewDecoder(f).Decode(&snap) + err = json.Unmarshal(jsonbyte, &snap) if err != nil { t.Fatal(err) } @@ -705,12 +697,6 @@ func testNetwork(t *testing.T) { t.Fatal(err) } - nodes := make([]discover.NodeID, len(snap.Nodes)) - bzzaddrs := make(map[discover.NodeID]string, len(snap.Nodes)) - rpcs := make(map[discover.NodeID]*rpc.Client, len(snap.Nodes)) - pubkeys := make(map[discover.NodeID]string, len(snap.Nodes)) - nodemsgcount := make(map[discover.NodeID]int, len(snap.Nodes)) - triggerChecks := func(trigger chan discover.NodeID, id discover.NodeID, rpcclient *rpc.Client, topic string) error { msgC := make(chan APIMsg) ctx, cancel := context.WithTimeout(context.Background(), time.Second) @@ -725,13 +711,11 @@ func testNetwork(t *testing.T) { select { case recvmsg := <-msgC: idx, _ := binary.Uvarint(recvmsg.Msg) - lock.Lock() if recvmsgs[idx] == false { log.Debug("msg recv", "idx", idx, "id", id) recvmsgs[idx] = true trigger <- id } - lock.Unlock() case <-sub.Err(): return } @@ -771,10 +755,15 @@ func testNetwork(t *testing.T) { } } - messagedelayhigh := 0 + // setup workers + jobs := make(chan Job, 10) + for w := 1; w <= 10; w++ { + go worker(w, jobs, rpcs, pubkeys, topic) + } + for i := 0; i < int(msgcount); i++ { - sendnodeidx := rand.Intn(int(len(snap.Nodes))) - recvnodeidx := rand.Intn(int(len(snap.Nodes) - 1)) + sendnodeidx := rand.Intn(int(nodecount)) + recvnodeidx := rand.Intn(int(nodecount - 1)) if recvnodeidx >= sendnodeidx { recvnodeidx++ } @@ -784,38 +773,23 @@ func testNetwork(t *testing.T) { if c == 0 { t.Fatal("0 byte message") } + if err != nil { + t.Fatal(err) + } err = rpcs[nodes[sendnodeidx]].Call(nil, "pss_setPeerPublicKey", pubkeys[nodes[recvnodeidx]], topic, bzzaddrs[nodes[recvnodeidx]]) if err != nil { t.Fatal(err) } - err = rpcs[nodes[recvnodeidx]].Call(nil, "pss_setPeerPublicKey", pubkeys[nodes[sendnodeidx]], topic, bzzaddrs[nodes[sendnodeidx]]) - if err != nil { - t.Fatal(err) - } - messagedelay := rand.Intn(int(messagedelaymax)) - if messagedelay > messagedelayhigh { - messagedelayhigh = messagedelay - } - messagedelayduration, err := time.ParseDuration(fmt.Sprintf("%dus", messagedelay)) - if err != nil { - t.Fatal(err) - } - go func(rpcclient *rpc.Client, pubkey string, msg []byte) { - time.Sleep(messagedelayduration) - err = rpcclient.Call(nil, "pss_sendAsym", pubkey, topic, hexutil.Encode(msg)) - if err != nil { - log.Error("Send asym rpc fail", "pubkey", pubkey, "topic", topic, "err", err) - } - }(rpcs[nodes[sendnodeidx]], pubkeys[nodes[recvnodeidx]], sentmsgs[i]) - } - timeout, err := time.ParseDuration(fmt.Sprintf("%dus", 60000000+messagedelayhigh)) - if err != nil { - t.Fatal(err) + jobs <- Job{ + Msg: sentmsgs[i], + SendNode: nodes[sendnodeidx], + RecvNode: nodes[recvnodeidx], + } } finalmsgcount := 0 - ctx, cancel := context.WithTimeout(context.Background(), timeout) + ctx, cancel := context.WithTimeout(context.Background(), 60*time.Second) defer cancel() outer: for i := 0; i < int(msgcount); i++ { @@ -996,7 +970,7 @@ func benchmarkSymkeyBruteforceChangeaddr(b *testing.B) { b.ResetTimer() for i := 0; i < b.N; i++ { if !ps.process(pssmsgs[len(pssmsgs)-(i%len(pssmsgs))-1]) { - b.Fatalf("pss processing failed") + b.Fatalf("pss processing failed: %v", err) } } } @@ -1154,6 +1128,9 @@ func newServices() adapters.Services { return nil, fmt.Errorf("local dpa creation failed", "error", err) } + // execadapter does not exec init() + initTest() + ctxlocal, _ := context.WithTimeout(context.Background(), time.Second) keys, err := wapi.NewKeyPair(ctxlocal) privkey, err := w.GetPrivateKey(keys) diff --git a/swarm/pss/testdata/addpsstodiscoverytestsnapshot.pl b/swarm/pss/testdata/addpsstodiscoverytestsnapshot.pl new file mode 100644 index 0000000000..b75cc9894a --- /dev/null +++ b/swarm/pss/testdata/addpsstodiscoverytestsnapshot.pl @@ -0,0 +1,28 @@ +#!/usr/bin/perl + +use JSON; + +my $f; +my $jsontext; +my $nodelist; +my $network; + +open($f, "<", $ARGV[0]) || die "cant open " . $ARGV[0]; +while (<$f>) { + $jsontext .= $_; +} +close($f); + +$network = decode_json($jsontext); +$nodelist = $network->{'nodes'}; + +for ($i = 0; $i < 0+@$nodelist; $i++) { + #my $protocollist = $$nodelist[$i]{'node'}{'info'}{'protocols'}; + #$$protocollist{'pss'} = "pss"; + my $svc = $$nodelist[$i]{'node'}{'config'}{'services'}; + pop(@$svc); + push(@$svc, "pss"); + push(@$svc, "bzz"); +} + +print encode_json($network); diff --git a/swarm/pss/testdata/snapshot_2.json b/swarm/pss/testdata/snapshot_2.json new file mode 100644 index 0000000000..5704bcf6af --- /dev/null +++ b/swarm/pss/testdata/snapshot_2.json @@ -0,0 +1,67 @@ +{ + "conns":[ + { + "other":"0eec333dd211c2ea81db614fe58bf0300c15e50e1b044e47ef93067a6cdbc3bc666b40bdcc515bbf580355dbef9370294ef1ee92ee0525e78a8beed00c2b99f5", + "one":"7b12f55c7c012104e006775d03b89722b403fb0e1ecb79af8cadfa6947425aedb323fb9416c84b782d35f3216acb5d94a1dd31d60a3eba45f9051bf503de1b66", + "up":true + } + ], + "nodes":[ + { + "node":{ + "config":{ + "private_key":"e567b7d9c554e5102cdc99b6523bace02dbb8951415c8816d82ba2d2e97fa23b", + "name":"node01", + "id":"7b12f55c7c012104e006775d03b89722b403fb0e1ecb79af8cadfa6947425aedb323fb9416c84b782d35f3216acb5d94a1dd31d60a3eba45f9051bf503de1b66", + "services":[ + "pss","bzz" + ] + }, + "info":{ + "ip":"0.0.0.0", + "listenAddr":"", + "protocols":{ + "hive":"\n=========================================================================\nFri Sep 29 21:22:53 UTC 2017 KΛÐΞMLIΛ hive: queen's address: 73d6ad\npopulation: 5 (7), MinProxBinSize: 2, MinBinSize: 1, MaxBinSize: 3\n000 1 dfd4 | 3 8a1e (0) d776 (0) dfd4 (0)\n============ DEPTH: 1 ==========================================\n001 3 05da 159c 3451 | 3 05da (0) 159c (0) 3451 (0)\n002 0 | 0\n003 1 6e8d | 1 6e8d (0)\n004 0 | 0\n005 0 | 0\n006 0 | 0\n007 0 | 0\n008 0 | 0\n009 0 | 0\n010 0 | 0\n011 0 | 0\n012 0 | 0\n013 0 | 0\n014 0 | 0\n015 0 | 0\n=========================================================================", + "bzz":"c9atSnUGnc7WYPpMuYFD7lVz33yxXZopWs8WVeloM4Q=" + }, + "ports":{ + "listener":0, + "discovery":0 + }, + "name":"node01", + "id":"7b12f55c7c012104e006775d03b89722b403fb0e1ecb79af8cadfa6947425aedb323fb9416c84b782d35f3216acb5d94a1dd31d60a3eba45f9051bf503de1b66", + "enode":"enode://7b12f55c7c012104e006775d03b89722b403fb0e1ecb79af8cadfa6947425aedb323fb9416c84b782d35f3216acb5d94a1dd31d60a3eba45f9051bf503de1b66@0.0.0.0:0" + }, + "up":true + } + }, + { + "node":{ + "info":{ + "listenAddr":"", + "ip":"0.0.0.0", + "ports":{ + "discovery":0, + "listener":0 + }, + "protocols":{ + "hive":"\n=========================================================================\nFri Sep 29 21:22:53 UTC 2017 KΛÐΞMLIΛ hive: queen's address: 6e8da8\npopulation: 5 (7), MinProxBinSize: 2, MinBinSize: 1, MaxBinSize: 3\n000 1 8a1e | 3 8a1e (0) d776 (0) dfd4 (0)\n============ DEPTH: 1 ==========================================\n001 3 3451 159c 05da | 3 05da (0) 159c (0) 3451 (0)\n002 0 | 0\n003 1 73d6 | 1 73d6 (0)\n004 0 | 0\n005 0 | 0\n006 0 | 0\n007 0 | 0\n008 0 | 0\n009 0 | 0\n010 0 | 0\n011 0 | 0\n012 0 | 0\n013 0 | 0\n014 0 | 0\n015 0 | 0\n=========================================================================", + "bzz":"bo2oaruJSrNQRMjEVRRyJd+WyrSY2gZ6EY8fuaQX+eM=" + }, + "id":"0eec333dd211c2ea81db614fe58bf0300c15e50e1b044e47ef93067a6cdbc3bc666b40bdcc515bbf580355dbef9370294ef1ee92ee0525e78a8beed00c2b99f5", + "name":"node02", + "enode":"enode://0eec333dd211c2ea81db614fe58bf0300c15e50e1b044e47ef93067a6cdbc3bc666b40bdcc515bbf580355dbef9370294ef1ee92ee0525e78a8beed00c2b99f5@0.0.0.0:0" + }, + "config":{ + "name":"node02", + "id":"0eec333dd211c2ea81db614fe58bf0300c15e50e1b044e47ef93067a6cdbc3bc666b40bdcc515bbf580355dbef9370294ef1ee92ee0525e78a8beed00c2b99f5", + "services":[ + "pss","bzz" + ], + "private_key":"c7526db70acd02f36d3b201ef3e1d85e38c52bee6931453213dbc5edec4d0976" + }, + "up":true + } + } + ] +} diff --git a/swarm/pss/testdata/snapshot_3.json b/swarm/pss/testdata/snapshot_3.json new file mode 100644 index 0000000000..2d815eef40 --- /dev/null +++ b/swarm/pss/testdata/snapshot_3.json @@ -0,0 +1,100 @@ +{ + "conns":[ + { + "one":"0eec333dd211c2ea81db614fe58bf0300c15e50e1b044e47ef93067a6cdbc3bc666b40bdcc515bbf580355dbef9370294ef1ee92ee0525e78a8beed00c2b99f5", + "other":"6f6ee658538ea66a68c9cb914d09f228f6ee9942c337a8d5a2cb3a0f021e83dd0fab481ca8ebf56ed913f6ddf69caa3249459d43e61e5e5b162ded7e1c918c9c", + "up":true + }, + { + "other":"0eec333dd211c2ea81db614fe58bf0300c15e50e1b044e47ef93067a6cdbc3bc666b40bdcc515bbf580355dbef9370294ef1ee92ee0525e78a8beed00c2b99f5", + "one":"7b12f55c7c012104e006775d03b89722b403fb0e1ecb79af8cadfa6947425aedb323fb9416c84b782d35f3216acb5d94a1dd31d60a3eba45f9051bf503de1b66", + "up":true + } + ], + "nodes":[ + { + "node":{ + "config":{ + "private_key":"e567b7d9c554e5102cdc99b6523bace02dbb8951415c8816d82ba2d2e97fa23b", + "name":"node01", + "id":"7b12f55c7c012104e006775d03b89722b403fb0e1ecb79af8cadfa6947425aedb323fb9416c84b782d35f3216acb5d94a1dd31d60a3eba45f9051bf503de1b66", + "services":[ + "bzz","pss" + ] + }, + "info":{ + "ip":"0.0.0.0", + "listenAddr":"", + "protocols":{ + "hive":"\n=========================================================================\nFri Sep 29 21:22:53 UTC 2017 KΛÐΞMLIΛ hive: queen's address: 73d6ad\npopulation: 5 (7), MinProxBinSize: 2, MinBinSize: 1, MaxBinSize: 3\n000 1 dfd4 | 3 8a1e (0) d776 (0) dfd4 (0)\n============ DEPTH: 1 ==========================================\n001 3 05da 159c 3451 | 3 05da (0) 159c (0) 3451 (0)\n002 0 | 0\n003 1 6e8d | 1 6e8d (0)\n004 0 | 0\n005 0 | 0\n006 0 | 0\n007 0 | 0\n008 0 | 0\n009 0 | 0\n010 0 | 0\n011 0 | 0\n012 0 | 0\n013 0 | 0\n014 0 | 0\n015 0 | 0\n=========================================================================", + "bzz":"c9atSnUGnc7WYPpMuYFD7lVz33yxXZopWs8WVeloM4Q=" + }, + "ports":{ + "listener":0, + "discovery":0 + }, + "name":"node01", + "id":"7b12f55c7c012104e006775d03b89722b403fb0e1ecb79af8cadfa6947425aedb323fb9416c84b782d35f3216acb5d94a1dd31d60a3eba45f9051bf503de1b66", + "enode":"enode://7b12f55c7c012104e006775d03b89722b403fb0e1ecb79af8cadfa6947425aedb323fb9416c84b782d35f3216acb5d94a1dd31d60a3eba45f9051bf503de1b66@0.0.0.0:0" + }, + "up":true + } + }, + { + "node":{ + "info":{ + "listenAddr":"", + "ip":"0.0.0.0", + "ports":{ + "discovery":0, + "listener":0 + }, + "protocols":{ + "hive":"\n=========================================================================\nFri Sep 29 21:22:53 UTC 2017 KΛÐΞMLIΛ hive: queen's address: 6e8da8\npopulation: 5 (7), MinProxBinSize: 2, MinBinSize: 1, MaxBinSize: 3\n000 1 8a1e | 3 8a1e (0) d776 (0) dfd4 (0)\n============ DEPTH: 1 ==========================================\n001 3 3451 159c 05da | 3 05da (0) 159c (0) 3451 (0)\n002 0 | 0\n003 1 73d6 | 1 73d6 (0)\n004 0 | 0\n005 0 | 0\n006 0 | 0\n007 0 | 0\n008 0 | 0\n009 0 | 0\n010 0 | 0\n011 0 | 0\n012 0 | 0\n013 0 | 0\n014 0 | 0\n015 0 | 0\n=========================================================================", + "bzz":"bo2oaruJSrNQRMjEVRRyJd+WyrSY2gZ6EY8fuaQX+eM=" + }, + "id":"0eec333dd211c2ea81db614fe58bf0300c15e50e1b044e47ef93067a6cdbc3bc666b40bdcc515bbf580355dbef9370294ef1ee92ee0525e78a8beed00c2b99f5", + "name":"node02", + "enode":"enode://0eec333dd211c2ea81db614fe58bf0300c15e50e1b044e47ef93067a6cdbc3bc666b40bdcc515bbf580355dbef9370294ef1ee92ee0525e78a8beed00c2b99f5@0.0.0.0:0" + }, + "config":{ + "name":"node02", + "id":"0eec333dd211c2ea81db614fe58bf0300c15e50e1b044e47ef93067a6cdbc3bc666b40bdcc515bbf580355dbef9370294ef1ee92ee0525e78a8beed00c2b99f5", + "services":[ + "bzz","pss" + ], + "private_key":"c7526db70acd02f36d3b201ef3e1d85e38c52bee6931453213dbc5edec4d0976" + }, + "up":true + } + }, + { + "node":{ + "config":{ + "private_key":"61b5728f59bc43080c3b8eb0458fb30d7723e2747355b6dc980f35f3ed431199", + "id":"6f6ee658538ea66a68c9cb914d09f228f6ee9942c337a8d5a2cb3a0f021e83dd0fab481ca8ebf56ed913f6ddf69caa3249459d43e61e5e5b162ded7e1c918c9c", + "name":"node03", + "services":[ + "bzz","pss" + ] + }, + "info":{ + "ip":"0.0.0.0", + "listenAddr":"", + "protocols":{ + "hive":"\n=========================================================================\nFri Sep 29 21:22:53 UTC 2017 KΛÐΞMLIΛ hive: queen's address: 8a1eb7\npopulation: 3 (7), MinProxBinSize: 2, MinBinSize: 1, MaxBinSize: 3\n000 1 6e8d | 5 05da (0) 159c (0) 3451 (0) 73d6 (0)\n============ DEPTH: 1 ==========================================\n001 2 dfd4 d776 | 2 dfd4 (0) d776 (0)\n002 0 | 0\n003 0 | 0\n004 0 | 0\n005 0 | 0\n006 0 | 0\n007 0 | 0\n008 0 | 0\n009 0 | 0\n010 0 | 0\n011 0 | 0\n012 0 | 0\n013 0 | 0\n014 0 | 0\n015 0 | 0\n=========================================================================", + "bzz":"ih63j/E98xjn+BFt/+6YzX2ZBWUPpT8Wdmt1SmPzh6w=" + }, + "ports":{ + "discovery":0, + "listener":0 + }, + "name":"node03", + "id":"6f6ee658538ea66a68c9cb914d09f228f6ee9942c337a8d5a2cb3a0f021e83dd0fab481ca8ebf56ed913f6ddf69caa3249459d43e61e5e5b162ded7e1c918c9c", + "enode":"enode://6f6ee658538ea66a68c9cb914d09f228f6ee9942c337a8d5a2cb3a0f021e83dd0fab481ca8ebf56ed913f6ddf69caa3249459d43e61e5e5b162ded7e1c918c9c@0.0.0.0:0" + }, + "up":true + } + } + ] +} diff --git a/swarm/pss/testdata/snapshot_4.json b/swarm/pss/testdata/snapshot_4.json new file mode 100644 index 0000000000..36058bfe33 --- /dev/null +++ b/swarm/pss/testdata/snapshot_4.json @@ -0,0 +1,133 @@ +{ + "conns":[ + { + "one":"0eec333dd211c2ea81db614fe58bf0300c15e50e1b044e47ef93067a6cdbc3bc666b40bdcc515bbf580355dbef9370294ef1ee92ee0525e78a8beed00c2b99f5", + "other":"6f6ee658538ea66a68c9cb914d09f228f6ee9942c337a8d5a2cb3a0f021e83dd0fab481ca8ebf56ed913f6ddf69caa3249459d43e61e5e5b162ded7e1c918c9c", + "up":true + }, + { + "other":"0eec333dd211c2ea81db614fe58bf0300c15e50e1b044e47ef93067a6cdbc3bc666b40bdcc515bbf580355dbef9370294ef1ee92ee0525e78a8beed00c2b99f5", + "one":"7b12f55c7c012104e006775d03b89722b403fb0e1ecb79af8cadfa6947425aedb323fb9416c84b782d35f3216acb5d94a1dd31d60a3eba45f9051bf503de1b66", + "up":true + }, + { + "up":true, + "other":"83388147883592bab4fdaddb4e56f8cb1c56dc5c2e910fc6a7277ac89b77cc7ce24892ed6984f3414589cb3b8c4b69356ff9aab7ca52fdd58f12dee2a2152523", + "one":"6f6ee658538ea66a68c9cb914d09f228f6ee9942c337a8d5a2cb3a0f021e83dd0fab481ca8ebf56ed913f6ddf69caa3249459d43e61e5e5b162ded7e1c918c9c" + } + ], + "nodes":[ + { + "node":{ + "config":{ + "private_key":"e567b7d9c554e5102cdc99b6523bace02dbb8951415c8816d82ba2d2e97fa23b", + "name":"node01", + "id":"7b12f55c7c012104e006775d03b89722b403fb0e1ecb79af8cadfa6947425aedb323fb9416c84b782d35f3216acb5d94a1dd31d60a3eba45f9051bf503de1b66", + "services":[ + "bzz","pss" + ] + }, + "info":{ + "ip":"0.0.0.0", + "listenAddr":"", + "protocols":{ + "hive":"\n=========================================================================\nFri Sep 29 21:22:53 UTC 2017 KΛÐΞMLIΛ hive: queen's address: 73d6ad\npopulation: 5 (7), MinProxBinSize: 2, MinBinSize: 1, MaxBinSize: 3\n000 1 dfd4 | 3 8a1e (0) d776 (0) dfd4 (0)\n============ DEPTH: 1 ==========================================\n001 3 05da 159c 3451 | 3 05da (0) 159c (0) 3451 (0)\n002 0 | 0\n003 1 6e8d | 1 6e8d (0)\n004 0 | 0\n005 0 | 0\n006 0 | 0\n007 0 | 0\n008 0 | 0\n009 0 | 0\n010 0 | 0\n011 0 | 0\n012 0 | 0\n013 0 | 0\n014 0 | 0\n015 0 | 0\n=========================================================================", + "bzz":"c9atSnUGnc7WYPpMuYFD7lVz33yxXZopWs8WVeloM4Q=" + }, + "ports":{ + "listener":0, + "discovery":0 + }, + "name":"node01", + "id":"7b12f55c7c012104e006775d03b89722b403fb0e1ecb79af8cadfa6947425aedb323fb9416c84b782d35f3216acb5d94a1dd31d60a3eba45f9051bf503de1b66", + "enode":"enode://7b12f55c7c012104e006775d03b89722b403fb0e1ecb79af8cadfa6947425aedb323fb9416c84b782d35f3216acb5d94a1dd31d60a3eba45f9051bf503de1b66@0.0.0.0:0" + }, + "up":true + } + }, + { + "node":{ + "info":{ + "listenAddr":"", + "ip":"0.0.0.0", + "ports":{ + "discovery":0, + "listener":0 + }, + "protocols":{ + "hive":"\n=========================================================================\nFri Sep 29 21:22:53 UTC 2017 KΛÐΞMLIΛ hive: queen's address: 6e8da8\npopulation: 5 (7), MinProxBinSize: 2, MinBinSize: 1, MaxBinSize: 3\n000 1 8a1e | 3 8a1e (0) d776 (0) dfd4 (0)\n============ DEPTH: 1 ==========================================\n001 3 3451 159c 05da | 3 05da (0) 159c (0) 3451 (0)\n002 0 | 0\n003 1 73d6 | 1 73d6 (0)\n004 0 | 0\n005 0 | 0\n006 0 | 0\n007 0 | 0\n008 0 | 0\n009 0 | 0\n010 0 | 0\n011 0 | 0\n012 0 | 0\n013 0 | 0\n014 0 | 0\n015 0 | 0\n=========================================================================", + "bzz":"bo2oaruJSrNQRMjEVRRyJd+WyrSY2gZ6EY8fuaQX+eM=" + }, + "id":"0eec333dd211c2ea81db614fe58bf0300c15e50e1b044e47ef93067a6cdbc3bc666b40bdcc515bbf580355dbef9370294ef1ee92ee0525e78a8beed00c2b99f5", + "name":"node02", + "enode":"enode://0eec333dd211c2ea81db614fe58bf0300c15e50e1b044e47ef93067a6cdbc3bc666b40bdcc515bbf580355dbef9370294ef1ee92ee0525e78a8beed00c2b99f5@0.0.0.0:0" + }, + "config":{ + "name":"node02", + "id":"0eec333dd211c2ea81db614fe58bf0300c15e50e1b044e47ef93067a6cdbc3bc666b40bdcc515bbf580355dbef9370294ef1ee92ee0525e78a8beed00c2b99f5", + "services":[ + "bzz","pss" + ], + "private_key":"c7526db70acd02f36d3b201ef3e1d85e38c52bee6931453213dbc5edec4d0976" + }, + "up":true + } + }, + { + "node":{ + "config":{ + "private_key":"61b5728f59bc43080c3b8eb0458fb30d7723e2747355b6dc980f35f3ed431199", + "id":"6f6ee658538ea66a68c9cb914d09f228f6ee9942c337a8d5a2cb3a0f021e83dd0fab481ca8ebf56ed913f6ddf69caa3249459d43e61e5e5b162ded7e1c918c9c", + "name":"node03", + "services":[ + "bzz","pss" + ] + }, + "info":{ + "ip":"0.0.0.0", + "listenAddr":"", + "protocols":{ + "hive":"\n=========================================================================\nFri Sep 29 21:22:53 UTC 2017 KΛÐΞMLIΛ hive: queen's address: 8a1eb7\npopulation: 3 (7), MinProxBinSize: 2, MinBinSize: 1, MaxBinSize: 3\n000 1 6e8d | 5 05da (0) 159c (0) 3451 (0) 73d6 (0)\n============ DEPTH: 1 ==========================================\n001 2 dfd4 d776 | 2 dfd4 (0) d776 (0)\n002 0 | 0\n003 0 | 0\n004 0 | 0\n005 0 | 0\n006 0 | 0\n007 0 | 0\n008 0 | 0\n009 0 | 0\n010 0 | 0\n011 0 | 0\n012 0 | 0\n013 0 | 0\n014 0 | 0\n015 0 | 0\n=========================================================================", + "bzz":"ih63j/E98xjn+BFt/+6YzX2ZBWUPpT8Wdmt1SmPzh6w=" + }, + "ports":{ + "discovery":0, + "listener":0 + }, + "name":"node03", + "id":"6f6ee658538ea66a68c9cb914d09f228f6ee9942c337a8d5a2cb3a0f021e83dd0fab481ca8ebf56ed913f6ddf69caa3249459d43e61e5e5b162ded7e1c918c9c", + "enode":"enode://6f6ee658538ea66a68c9cb914d09f228f6ee9942c337a8d5a2cb3a0f021e83dd0fab481ca8ebf56ed913f6ddf69caa3249459d43e61e5e5b162ded7e1c918c9c@0.0.0.0:0" + }, + "up":true + } + }, + { + "node":{ + "info":{ + "name":"node04", + "id":"83388147883592bab4fdaddb4e56f8cb1c56dc5c2e910fc6a7277ac89b77cc7ce24892ed6984f3414589cb3b8c4b69356ff9aab7ca52fdd58f12dee2a2152523", + "enode":"enode://83388147883592bab4fdaddb4e56f8cb1c56dc5c2e910fc6a7277ac89b77cc7ce24892ed6984f3414589cb3b8c4b69356ff9aab7ca52fdd58f12dee2a2152523@0.0.0.0:0", + "ip":"0.0.0.0", + "listenAddr":"", + "protocols":{ + "bzz":"13aDNPedYmrbQz9EtwOoGFVeMzEFYDbvP40Sglhr8EQ=", + "hive":"\n=========================================================================\nFri Sep 29 21:22:53 UTC 2017 KΛÐΞMLIΛ hive: queen's address: d77683\npopulation: 5 (7), MinProxBinSize: 2, MinBinSize: 1, MaxBinSize: 3\n000 3 3451 159c 05da | 5 6e8d (0) 73d6 (0) 3451 (0) 159c (0)\n============ DEPTH: 1 ==========================================\n001 1 8a1e | 1 8a1e (0)\n002 0 | 0\n003 0 | 0\n004 1 dfd4 | 1 dfd4 (0)\n005 0 | 0\n006 0 | 0\n007 0 | 0\n008 0 | 0\n009 0 | 0\n010 0 | 0\n011 0 | 0\n012 0 | 0\n013 0 | 0\n014 0 | 0\n015 0 | 0\n=========================================================================" + }, + "ports":{ + "listener":0, + "discovery":0 + } + }, + "config":{ + "services":[ + "bzz","pss" + ], + "id":"83388147883592bab4fdaddb4e56f8cb1c56dc5c2e910fc6a7277ac89b77cc7ce24892ed6984f3414589cb3b8c4b69356ff9aab7ca52fdd58f12dee2a2152523", + "name":"node04", + "private_key":"075b07c29ceac4ffa2a114afd67b21dfc438126bc169bf7c154be6d81d86ed38" + }, + "up":true + } + } + ] +} diff --git a/swarm/pss/writeup.md b/swarm/pss/writeup.md new file mode 100644 index 0000000000..a0506ffa43 --- /dev/null +++ b/swarm/pss/writeup.md @@ -0,0 +1,125 @@ +## PSS tests failures explanation + +This document aims to explain the changes in https://github.com/ethersphere/go-ethereum/pull/126 and how those changes affect the pss_test.go TestNetwork tests. + +### Problem + +When running the TestNetwork test, execution sometimes: + +* deadlocks +* panics +* failures with wrong result, such as: + +``` +$ go test -v ./swarm/pss -cpu 4 -run TestNetwork +``` + +``` +--- FAIL: TestNetwork (68.13s) + --- FAIL: TestNetwork/3/10/4/sim (68.13s) + pss_test.go:697: 7 of 10 messages received + pss_test.go:700: 3 messages were not received +FAIL +``` + +Moreover execution almost always deadlocks with `sim` adapter, and `sock` adapter (when buffer is low), but is mostly stable with `exec` and `tcp` adapters. + +### Findings and Fixes + +#### 1. Addressing panics + +Panics were caused due to concurrent map read/writes and unsynchronised access to shared memory by multiple goroutines. This is visible when running the test with the `-race` flag. + +``` +go test -race -v ./swarm/pss -cpu 4 -run TestNetwork + + 1 ================== + 2 WARNING: DATA RACE + 3 Read at 0x00c424d456a0 by goroutine 1089: + 4 github.com/ethereum/go-ethereum/swarm/pss.(*Pss).forward.func1() + 5 /Users/nonsense/code/src/github.com/ethereum/go-ethereum/swarm/pss/pss.go:654 +0x44f + 6 github.com/ethereum/go-ethereum/swarm/network.(*Kademlia).eachConn.func1() + 7 /Users/nonsense/code/src/github.com/ethereum/go-ethereum/swarm/network/kademlia.go:350 +0xc9 + 8 github.com/ethereum/go-ethereum/pot.(*Pot).eachNeighbour.func1() + 9 /Users/nonsense/code/src/github.com/ethereum/go-ethereum/pot/pot.go:599 +0x59 + ... + + 28 + 29 Previous write at 0x00c424d456a0 by goroutine 829: + 30 github.com/ethereum/go-ethereum/swarm/pss.(*Pss).Run() + 31 /Users/nonsense/code/src/github.com/ethereum/go-ethereum/swarm/pss/pss.go:192 +0x16a + 32 github.com/ethereum/go-ethereum/swarm/pss.(*Pss).Run-fm() + 33 /Users/nonsense/code/src/github.com/ethereum/go-ethereum/swarm/pss/pss.go:185 +0x63 + 34 github.com/ethereum/go-ethereum/p2p.(*Peer).startProtocols.func1() + 35 /Users/nonsense/code/src/github.com/ethereum/go-ethereum/p2p/peer.go:347 +0x8b + ... +``` + +##### Current solution + +Adding a mutex around all shared data. + +#### 2. Failures with wrong result + +The validation phase of the TestNetwork test is done using an RPC subscription: + +``` + ... + triggerChecks := func(trigger chan discover.NodeID, id discover.NodeID, rpcclient *rpc.Client) error { + msgC := make(chan APIMsg) + ctx, cancel := context.WithTimeout(context.Background(), time.Second) + defer cancel() + sub, err := rpcclient.Subscribe(ctx, "pss", msgC, "receive", hextopic) + ... +``` + +By design the RPC uses a subscription buffer with a max length. When this length is reached, the subscription is dropped. The current config value is not suitable for stress tests. + +##### Current solution + +Increase the max length of the RPC subscription buffer. + +``` +const ( + // Subscriptions are removed when the subscriber cannot keep up. + // + // This can be worked around by supplying a channel with sufficiently sized buffer, + // but this can be inconvenient and hard to explain in the docs. Another issue with + // buffered channels is that the buffer is static even though it might not be needed + // most of the time. + // + // The approach taken here is to maintain a per-subscription linked list buffer + // shrinks on demand. If the buffer reaches the size below, the subscription is + // dropped. + maxClientSubscriptionBuffer = 20000 +) +``` + +#### 3. Deadlocks + +Deadlocks are triggered when using: +* `sim` adapter - synchronous, unbuffered channel +* `sock` adapter - asynchronous, buffered channel (when using a 1K buffer) + +No deadlocks were triggered when using: +* `tcp` adapter - asynchronous, buffered channel +* `exec` adapter - asynchronous, buffered channel + +Ultimately the deadlocks happen due to blocking `pp.Send()` call at: + + // attempt to send the message + err := pp.Send(msg) + if err != nil { + log.Debug(fmt.Sprintf("%v: failed forwarding: %v", sendMsg, err)) + return true + } + + `p2p` request handling is synchronous (as discussed at https://github.com/ethersphere/go-ethereum/issues/130), `pss` is also synchronous, therefore if two nodes happen to be processing a request, while at the same time waiting for response on `pp.Send(msg)`, deadlock occurs. + + `pp.Send(msg)` is only blocking when the underlying adapter is blocking (read `sim` or `sock`) or the buffer of the connection is full. + +##### Current solution + +Make no assumption on the undelying connection, and call `pp.Send` asynchronously in a go-routine. + +Alternatively, get rid of the `sim` and `sock` adapters, and use `tcp` adapter for testing. diff --git a/swarm/state.go b/swarm/state.go new file mode 100644 index 0000000000..6d85219516 --- /dev/null +++ b/swarm/state.go @@ -0,0 +1,12 @@ +package swarm + +type Voidstore struct { +} + +func (self Voidstore) Load(string) ([]byte, error) { + return nil, nil +} + +func (self Voidstore) Save(string, []byte) error { + return nil +}