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

@@ -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)

swarm/pss/pss.go

@@ -35,6 +35,7 @@ const (
 	defaultOutboxQueueSize     = 10000
 	pssProtocolName            = "pss"
 	pssVersion                 = 1
+	hasherCount                = 8
 )
 
 var (
@@ -46,7 +47,6 @@ var (
 // and mailbox implementation
 type pssCacheEntry struct {
 	expiresAt time.Time
-	receivedFrom []byte
 }
 
 // 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))
-			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))
+			return true
 		}
-		}()
-		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 {
+func (self *Pss) addFwdCache(msg *PssMsg) error {
-	self.fwdCacheMu.Lock()
-	defer self.fwdCacheMu.Unlock()
 	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 {
+func (self *Pss) checkFwdCache(msg *PssMsg) bool {
-	self.fwdCacheMu.RLock()
+	self.fwdCacheMu.Lock()
-	defer self.fwdCacheMu.RUnlock()
+	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
+// Digest of message
-func (self *Pss) storeMsg(msg *PssMsg) (pssDigest, error) {
+func (self *Pss) digest(msg *PssMsg) pssDigest {
-	buf := bytes.NewReader(msg.serialize())
+	hasher := self.hashPool.Get().(storage.SwarmHash)
-	key, _, err := self.dpa.Store(buf, int64(buf.Len()), false)
+	defer self.hashPool.Put(hasher)
-	if err != nil {
+	hasher.Reset()
-		log.Warn("Could not store in swarm", "err", err)
+	hasher.Write(msg.serialize())
-		return pssDigest{}, err
-	}
-	log.Trace("Stored msg in swarm", "key", key)
 	digest := pssDigest{}
+	key := hasher.Sum(nil)
 	copy(digest[:], key[:digestLength])
-	return digest, nil
+	return digest
 }
 
 func (self *Pss) isMsgExpired(msg *PssMsg) bool {

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
 }

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
 }
 

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())
 		}