From 8156737b57b83dd8405f210991bc98534a8454f7 Mon Sep 17 00:00:00 2001 From: lash Date: Thu, 29 Mar 2018 20:55:41 +0200 Subject: [PATCH] swarm/pss: Add message deduplication swarm/pss: WIP outbox swarm/pss: WIP deduplication swarm/pss: Remove pss DPA + leaner cache swarm/pss: Cleanup after rebase swarm/pss: Rebase on swarm-network-rewrite --- swarm/pss/client/client_test.go | 12 +-- swarm/pss/pss.go | 104 ++++++++++++++------------ swarm/pss/pss_test.go | 128 ++++++++++++++++++++++++-------- swarm/pss/types.go | 8 +- swarm/swarm.go | 2 +- 5 files changed, 162 insertions(+), 92 deletions(-) diff --git a/swarm/pss/client/client_test.go b/swarm/pss/client/client_test.go index 10f0171197..cc33b47231 100644 --- a/swarm/pss/client/client_test.go +++ b/swarm/pss/client/client_test.go @@ -4,7 +4,6 @@ import ( "context" "flag" "fmt" - "io/ioutil" "math/rand" "os" "sync" @@ -22,7 +21,6 @@ import ( "github.com/ethereum/go-ethereum/swarm/network" "github.com/ethereum/go-ethereum/swarm/pss" "github.com/ethereum/go-ethereum/swarm/state" - "github.com/ethereum/go-ethereum/swarm/storage" whisper "github.com/ethereum/go-ethereum/whisper/whisperv5" ) @@ -231,21 +229,13 @@ func newServices() adapters.Services { } return adapters.Services{ "pss": func(ctx *adapters.ServiceContext) (node.Service, error) { - cachedir, err := ioutil.TempDir("", "pss-cache") - if err != nil { - return nil, fmt.Errorf("create pss cache tmpdir failed: %s", err) - } - dpa, err := storage.NewLocalDPA(cachedir, make([]byte, 32)) - if err != nil { - return nil, fmt.Errorf("local dpa creation failed: %s", err) - } ctxlocal, cancel := context.WithTimeout(context.Background(), time.Second) defer cancel() keys, err := wapi.NewKeyPair(ctxlocal) privkey, err := w.GetPrivateKey(keys) psparams := pss.NewPssParams(privkey) pskad := kademlia(ctx.Config.ID) - ps := pss.NewPss(pskad, dpa, psparams) + ps := pss.NewPss(pskad, psparams) pshparams := pss.NewHandshakeParams() pshparams.SymKeySendLimit = sendLimit err = pss.SetHandshakeController(ps, pshparams) diff --git a/swarm/pss/pss.go b/swarm/pss/pss.go index f4e2f5e942..1b88c95db2 100644 --- a/swarm/pss/pss.go +++ b/swarm/pss/pss.go @@ -35,6 +35,7 @@ const ( defaultOutboxQueueSize = 10000 pssProtocolName = "pss" pssVersion = 1 + hasherCount = 8 ) var ( @@ -45,8 +46,7 @@ var ( // will also be instrumental in flood guard mechanism // and mailbox implementation type pssCacheEntry struct { - expiresAt time.Time - receivedFrom []byte + expiresAt time.Time } // abstraction to enable access to p2p.protocols.Peer.Send @@ -89,7 +89,6 @@ func NewPssParams(privatekey *ecdsa.PrivateKey) *PssParams { type Pss struct { network.Overlay // we can get the overlayaddress from this privateKey *ecdsa.PrivateKey // pss can have it's own independent key - dpa *storage.DPA // we use swarm to store the cache w *whisper.Whisper // key and encryption backend auxAPIs []rpc.API // builtins (handshake, test) can add APIs @@ -116,6 +115,7 @@ type Pss struct { // message handling handlers map[Topic]map[*Handler]bool // topic and version based pss payload handlers. See pss.Handle() handlersMu sync.RWMutex + hashPool sync.Pool // process quitC chan struct{} @@ -129,15 +129,14 @@ func (self *Pss) String() string { // // In addition to params, it takes a swarm network overlay // and a DPA storage for message cache storage. -func NewPss(k network.Overlay, dpa *storage.DPA, params *PssParams) *Pss { +func NewPss(k network.Overlay, params *PssParams) *Pss { cap := p2p.Cap{ Name: pssProtocolName, Version: pssVersion, } - return &Pss{ + ps := &Pss{ Overlay: k, privateKey: params.privateKey, - dpa: dpa, w: whisper.New(&whisper.DefaultConfig), quitC: make(chan struct{}), @@ -155,7 +154,19 @@ func NewPss(k network.Overlay, dpa *storage.DPA, params *PssParams) *Pss { symKeyDecryptCacheCapacity: params.SymKeyCacheCapacity, handlers: make(map[Topic]map[*Handler]bool), + hashPool: sync.Pool{ + New: func() interface{} { + return storage.MakeHashFunc(storage.SHA3Hash)() + }, + }, } + + for i := 0; i < hasherCount; i++ { + hashfunc := storage.MakeHashFunc(storage.SHA3Hash)() + ps.hashPool.Put(hashfunc) + } + + return ps } ///////////////////////////////////////////////////////////////////// @@ -170,7 +181,7 @@ func (self *Pss) Start(srv *p2p.Server) error { case <-tickC: self.cleanKeys() case <-self.quitC: - log.Info("pss shutting down") + return } } }() @@ -180,7 +191,7 @@ func (self *Pss) Start(srv *p2p.Server) error { case msg := <-self.outbox: self.forward(msg) case <-self.quitC: - log.Info("pss shutting down") + return } } }() @@ -189,6 +200,7 @@ func (self *Pss) Start(srv *p2p.Server) error { } func (self *Pss) Stop() error { + log.Info("pss shutting down") close(self.quitC) return nil } @@ -298,7 +310,14 @@ func (self *Pss) getHandlers(topic Topic) map[*Handler]bool { // Passes error to pss protocol handler if payload is not valid pssmsg func (self *Pss) handlePssMsg(msg interface{}) error { pssmsg, ok := msg.(*PssMsg) + if ok { + if self.checkFwdCache(pssmsg) { + log.Trace(fmt.Sprintf("pss relay block-cache match (process): FROM %x TO %x", self.Overlay.BaseAddr(), common.ToHex(pssmsg.To))) + return nil + } + self.addFwdCache(pssmsg) + var err error if !self.isSelfPossibleRecipient(pssmsg) { log.Trace("pss was for someone else :'( ... forwarding", "pss", common.ToHex(self.BaseAddr())) @@ -532,7 +551,7 @@ func (self *Pss) cleanKeys() (count int) { for keyid, peertopics := range self.symKeyPool { var expiredtopics []Topic for topic, psp := range peertopics { - log.Trace("check topic", "topic", topic, "id", keyid, "protect", psp.protected, "p", fmt.Sprintf("%p", self.symKeyPool[keyid][topic])) + //log.Trace("check topic", "topic", topic, "id", keyid, "protect", psp.protected, "p", fmt.Sprintf("%p", self.symKeyPool[keyid][topic])) if psp.protected { continue } @@ -540,7 +559,7 @@ func (self *Pss) cleanKeys() (count int) { var match bool for i := self.symKeyDecryptCacheCursor; i > self.symKeyDecryptCacheCursor-cap(self.symKeyDecryptCache) && i > 0; i-- { cacheid := self.symKeyDecryptCache[i%cap(self.symKeyDecryptCache)] - log.Trace("check cache", "idx", i, "id", *cacheid) + //log.Trace("check cache", "idx", i, "id", *cacheid) if *cacheid == keyid { match = true } @@ -665,16 +684,9 @@ func (self *Pss) forward(msg *PssMsg) { to := make([]byte, addressLength) copy(to[:len(msg.To)], msg.To) - // message hash - digest, err := self.storeMsg(msg) - if err != nil { - log.Warn(fmt.Sprintf("could not store message %v to cache: %v", msg, err)) - } - // send with kademlia // find the closest peer to the recipient and attempt to send sent := 0 - self.Overlay.EachConn(to, 256, func(op network.OverlayConn, po int, isproxbin bool) bool { // we need p2p.protocols.Peer.Send // cast and resolve @@ -699,23 +711,19 @@ func (self *Pss) forward(msg *PssMsg) { } // get the protocol peer from the forwarding peer cache - sendMsg := fmt.Sprintf("MSG %x TO %x FROM %x VIA %x", digest, to, self.BaseAddr(), op.Address()) + sendMsg := fmt.Sprintf("MSG TO %x FROM %x VIA %x", to, self.BaseAddr(), op.Address()) self.fwdPoolMu.RLock() pp := self.fwdPool[sp.Info().ID] self.fwdPoolMu.RUnlock() - if self.checkFwdCache(op.Address(), digest) { - log.Trace(fmt.Sprintf("%v: peer already forwarded to", sendMsg)) + + // attempt to send the message + err := pp.Send(msg) + if err != nil { return true } - // attempt to send the message - 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++ + log.Trace(fmt.Sprintf("%v: successfully forwarded", sendMsg)) + // continue forwarding if: // - if the peer is end recipient but the full address has not been disclosed // - if the peer address matches the partial address fully @@ -741,7 +749,7 @@ func (self *Pss) forward(msg *PssMsg) { } // cache the message - self.addFwdCache(digest) + self.addFwdCache(msg) } ///////////////////////////////////////////////////////////////////// @@ -749,11 +757,14 @@ func (self *Pss) forward(msg *PssMsg) { ///////////////////////////////////////////////////////////////////// // add a message to the cache -func (self *Pss) addFwdCache(digest pssDigest) error { - self.fwdCacheMu.Lock() - defer self.fwdCacheMu.Unlock() +func (self *Pss) addFwdCache(msg *PssMsg) error { var entry pssCacheEntry var ok bool + + self.fwdCacheMu.Lock() + defer self.fwdCacheMu.Unlock() + + digest := self.digest(msg) if entry, ok = self.fwdCache[digest]; !ok { entry = pssCacheEntry{} } @@ -763,34 +774,31 @@ func (self *Pss) addFwdCache(digest pssDigest) error { } // check if message is in the cache -func (self *Pss) checkFwdCache(addr []byte, digest pssDigest) bool { - self.fwdCacheMu.RLock() - defer self.fwdCacheMu.RUnlock() +func (self *Pss) checkFwdCache(msg *PssMsg) bool { + self.fwdCacheMu.Lock() + defer self.fwdCacheMu.Unlock() + + digest := self.digest(msg) entry, ok := self.fwdCache[digest] if ok { if entry.expiresAt.After(time.Now()) { log.Trace(fmt.Sprintf("unexpired cache for digest %x", digest)) return true - } else if entry.expiresAt.IsZero() && bytes.Equal(addr, entry.receivedFrom) { - log.Trace(fmt.Sprintf("sendermatch %x for digest %x", common.ToHex(addr), digest)) - return true } } return false } -// DPA storage handler for message cache -func (self *Pss) storeMsg(msg *PssMsg) (pssDigest, error) { - buf := bytes.NewReader(msg.serialize()) - key, _, err := self.dpa.Store(buf, int64(buf.Len()), false) - if err != nil { - log.Warn("Could not store in swarm", "err", err) - return pssDigest{}, err - } - log.Trace("Stored msg in swarm", "key", key) +// Digest of message +func (self *Pss) digest(msg *PssMsg) pssDigest { + hasher := self.hashPool.Get().(storage.SwarmHash) + defer self.hashPool.Put(hasher) + hasher.Reset() + hasher.Write(msg.serialize()) digest := pssDigest{} + key := hasher.Sum(nil) copy(digest[:], key[:digestLength]) - return digest, nil + return digest } func (self *Pss) isMsgExpired(msg *PssMsg) bool { diff --git a/swarm/pss/pss_test.go b/swarm/pss/pss_test.go index bbfb074dd3..4ea0d843df 100644 --- a/swarm/pss/pss_test.go +++ b/swarm/pss/pss_test.go @@ -32,7 +32,6 @@ import ( "github.com/ethereum/go-ethereum/rpc" "github.com/ethereum/go-ethereum/swarm/network" "github.com/ethereum/go-ethereum/swarm/state" - "github.com/ethereum/go-ethereum/swarm/storage" whisper "github.com/ethereum/go-ethereum/whisper/whisperv5" ) @@ -171,11 +170,11 @@ func TestCache(t *testing.T) { To: to, } - digest, err := ps.storeMsg(msg) + digest := ps.digest(msg) if err != nil { t.Fatalf("could not store cache msgone: %v", err) } - digesttwo, err := ps.storeMsg(msgtwo) + digesttwo := ps.digest(msgtwo) if err != nil { t.Fatalf("could not store cache msgtwo: %v", err) } @@ -185,21 +184,21 @@ func TestCache(t *testing.T) { } // check the cache - err = ps.addFwdCache(digest) + err = ps.addFwdCache(msg) if err != nil { t.Fatalf("write to pss expire cache failed: %v", err) } - if !ps.checkFwdCache(nil, digest) { + if !ps.checkFwdCache(msg) { t.Fatalf("message %v should have EXPIRE record in cache but checkCache returned false", msg) } - if ps.checkFwdCache(nil, digesttwo) { + if ps.checkFwdCache(msgtwo) { 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) { + if ps.checkFwdCache(msg) { t.Fatalf("message %v should have expired from cache but checkCache returned true", msg) } } @@ -220,7 +219,7 @@ func TestAddressMatch(t *testing.T) { } privkey, err := w.GetPrivateKey(keys) pssp := NewPssParams(privkey) - ps := NewPss(kad, nil, pssp) + ps := NewPss(kad, pssp) pssmsg := &PssMsg{ To: remoteaddr, @@ -839,6 +838,94 @@ outer: } +func TestDeduplication(t *testing.T) { + var err error + + clients, err := setupNetwork(3) + if err != nil { + t.Fatal(err) + } + + var addrsize = 32 + var loaddrhex string + err = clients[0].Call(&loaddrhex, "pss_baseAddr") + if err != nil { + t.Fatalf("rpc get node 1 baseaddr fail: %v", err) + } + loaddrhex = loaddrhex[:2+(addrsize*2)] + var roaddrhex string + err = clients[1].Call(&roaddrhex, "pss_baseAddr") + if err != nil { + t.Fatalf("rpc get node 2 baseaddr fail: %v", err) + } + roaddrhex = roaddrhex[:2+(addrsize*2)] + var xoaddrhex string + err = clients[2].Call(&xoaddrhex, "pss_baseAddr") + if err != nil { + t.Fatalf("rpc get node 3 baseaddr fail: %v", err) + } + xoaddrhex = xoaddrhex[:2+(addrsize*2)] + + log.Info("peer", "l", loaddrhex, "r", roaddrhex, "x", xoaddrhex) + + var topic string + err = clients[0].Call(&topic, "pss_stringToTopic", "foo:42") + if err != nil { + t.Fatal(err) + } + + time.Sleep(time.Millisecond * 250) + + // retrieve public key from pss instance + // set this public key reciprocally + var rpubkey string + err = clients[1].Call(&rpubkey, "pss_getPublicKey") + if err != nil { + t.Fatalf("rpc get receivenode pubkey fail: %v", err) + } + + time.Sleep(time.Millisecond * 500) // replace with hive healthy code + + rmsgC := make(chan APIMsg) + rctx, cancel := context.WithTimeout(context.Background(), time.Second*1) + defer cancel() + rsub, err := clients[1].Subscribe(rctx, "pss", rmsgC, "receive", topic) + log.Trace("rsub", "id", rsub) + defer rsub.Unsubscribe() + + // store public key for recipient + // zero-length address means forward to all + // we have just two peers, they will be in proxbin, and will both receive + err = clients[0].Call(nil, "pss_setPeerPublicKey", rpubkey, topic, "0x") + if err != nil { + t.Fatal(err) + } + + // send and verify delivery + rmsg := []byte("xyzzy") + err = clients[0].Call(nil, "pss_sendAsym", rpubkey, topic, hexutil.Encode(rmsg)) + if err != nil { + t.Fatal(err) + } + + var receivedok bool +OUTER: + for { + select { + case <-rmsgC: + if receivedok { + t.Fatalf("duplicate message received") + } + receivedok = true + case <-rctx.Done(): + break OUTER + } + } + if !receivedok { + t.Fatalf("message did not arrive") + } +} + // symmetric send performance with varying message sizes func BenchmarkSymkeySend(b *testing.B) { b.Run(fmt.Sprintf("%d", 256), benchmarkSymKeySend) @@ -1146,15 +1233,6 @@ func newServices() adapters.Services { } return adapters.Services{ pssProtocolName: func(ctx *adapters.ServiceContext) (node.Service, error) { - cachedir, err := ioutil.TempDir("", "pss-cache") - if err != nil { - return nil, fmt.Errorf("create pss cache tmpdir failed: %s", err) - } - dpa, err := storage.NewLocalDPA(cachedir, network.NewAddrFromNodeID(ctx.Config.ID).Over()) - if err != nil { - return nil, fmt.Errorf("local dpa creation failed: %s", err) - } - // execadapter does not exec init() initTest() @@ -1165,7 +1243,7 @@ func newServices() adapters.Services { pssp := NewPssParams(privkey) pssp.MsgTTL = time.Second * 30 pskad := kademlia(ctx.Config.ID) - ps := NewPss(pskad, dpa, pssp) + ps := NewPss(pskad, pssp) ping := &Ping{ OutC: make(chan bool), @@ -1217,18 +1295,6 @@ func newTestPss(privkey *ecdsa.PrivateKey, overlay network.Overlay, ppextra *Pss copy(nid[:], crypto.FromECDSAPub(&privkey.PublicKey)) addr := network.NewAddrFromNodeID(nid) - // set up storage - cachedir, err := ioutil.TempDir("", "pss-cache") - if err != nil { - log.Error("create pss cache tmpdir failed", "error", err) - os.Exit(1) - } - dpa, err := storage.NewLocalDPA(cachedir, addr.Over()) - if err != nil { - log.Error("local dpa creation failed", "error", err) - os.Exit(1) - } - // set up routing if kademlia is not passed to us if overlay == nil { kp := network.NewKadParams() @@ -1241,7 +1307,7 @@ func newTestPss(privkey *ecdsa.PrivateKey, overlay network.Overlay, ppextra *Pss if ppextra != nil { pp.SymKeyCacheCapacity = ppextra.SymKeyCacheCapacity } - ps := NewPss(overlay, dpa, pp) + ps := NewPss(overlay, pp) return ps } diff --git a/swarm/pss/types.go b/swarm/pss/types.go index 95cd22b068..9c6e49bb8f 100644 --- a/swarm/pss/types.go +++ b/swarm/pss/types.go @@ -75,7 +75,13 @@ type PssMsg struct { // serializes the message for use in cache func (msg *PssMsg) serialize() []byte { - rlpdata, _ := rlp.EncodeToBytes(msg) + rlpdata, _ := rlp.EncodeToBytes(struct { + To []byte + Payload *whisper.Envelope + }{ + To: msg.To, + Payload: msg.Payload, + }) return rlpdata } diff --git a/swarm/swarm.go b/swarm/swarm.go index ab4b3650e5..0406c560ed 100644 --- a/swarm/swarm.go +++ b/swarm/swarm.go @@ -171,7 +171,7 @@ func NewSwarm(ctx *node.ServiceContext, backend chequebook.Backend, config *api. // Pss = postal service over swarm (devp2p over bzz) if self.config.PssEnabled { pssparams := pss.NewPssParams(self.privateKey) - self.ps = pss.NewPss(to, self.dpa, pssparams) + self.ps = pss.NewPss(to, pssparams) if pss.IsActiveHandshake { pss.SetHandshakeController(self.ps, pss.NewHandshakeParams()) }