swarm/pss: Create outbox queue for pss

Implements a queue manager to enable resending when forwarding fails.
Messages are not forwarded right away, but put in a queue which is in
turn fetched by a loop started when the service starts.

swarm/pss: WIP outbox

swarm/pss: Add read mutexes

swarm/pss: Implement queue as channel

swarm/pss: Remove commented code

swarm/pss/handshake.go

@@ -307,7 +307,7 @@ func (self *HandshakeController) handleKeys(pubkeyid string, keymsg *handshakeMs
 			copy(sendsymkey, key)
 			var address PssAddress
 			copy(address[:], keymsg.From)
-			sendsymkeyid, err := self.pss.SetSymmetricKey(sendsymkey, keymsg.Topic, &address, false)
+			sendsymkeyid, err := self.pss.setSymmetricKey(sendsymkey, keymsg.Topic, &address, false, false)
 			if err != nil {
 				return err
 			}

swarm/pss/pss.go

@@ -4,7 +4,6 @@ import (
 	"bytes"
 	"crypto/ecdsa"
 	"crypto/rand"
-	"errors"
 	"fmt"
 	"sync"
 	"time"
@@ -31,7 +30,9 @@ const (
 	defaultWhisperWorkTime     = 3
 	defaultWhisperPoW          = 0.0000000001
 	defaultMaxMsgSize          = 1024 * 1024
-	defaultCleanInterval       = 1000 * 60 * 10
+	defaultCleanInterval       = time.Second * 60 * 10
+	defaultDequeueInterval     = time.Millisecond * 10
+	defaultOutboxQueueSize     = 10000
 	pssProtocolName            = "pss"
 	pssVersion                 = 1
 )
@@ -94,13 +95,14 @@ type Pss struct {
 
 	// sending and forwarding
 	fwdPool         map[string]*protocols.Peer // keep track of all peers sitting on the pssmsg routing layer
-	fwdPoolMu       sync.Mutex
+	fwdPoolMu       sync.RWMutex
 	fwdCache        map[pssDigest]pssCacheEntry // checksum of unique fields from pssmsg mapped to expiry, cache to determine whether to drop msg
-	fwdCacheMu      sync.Mutex
+	fwdCacheMu      sync.RWMutex
 	cacheTTL        time.Duration // how long to keep messages in fwdCache (not implemented)
 	msgTTL          time.Duration
 	paddingByteSize int
 	capstring       string
+	outbox          chan *PssMsg
 
 	// keys and peers
 	pubKeyPool                 map[string]map[Topic]*pssPeer // mapping of hex public keys to peer address by topic.
@@ -113,7 +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.Mutex
+	handlersMu sync.RWMutex
 
 	// process
 	quitC chan struct{}
@@ -145,6 +147,7 @@ func NewPss(k network.Overlay, dpa *storage.DPA, params *PssParams) *Pss {
 		msgTTL:          params.MsgTTL,
 		paddingByteSize: defaultPaddingByteSize,
 		capstring:       cap.String(),
+		outbox:          make(chan *PssMsg, defaultOutboxQueueSize),
 
 		pubKeyPool:                 make(map[string]map[Topic]*pssPeer),
 		symKeyPool:                 make(map[string]map[Topic]*pssPeer),
@@ -161,12 +164,24 @@ func NewPss(k network.Overlay, dpa *storage.DPA, params *PssParams) *Pss {
 
 func (self *Pss) Start(srv *p2p.Server) error {
 	go func() {
-		tickC := time.Tick(defaultCleanInterval)
+		for {
-		select {
+			tickC := time.Tick(defaultCleanInterval)
-		case <-tickC:
+			select {
-			self.cleanKeys()
+			case <-tickC:
-		case <-self.quitC:
+				self.cleanKeys()
-			log.Info("pss shutting down")
+			case <-self.quitC:
+				log.Info("pss shutting down")
+			}
+		}
+	}()
+	go func() {
+		for {
+			select {
+			case msg := <-self.outbox:
+				self.forward(msg)
+			case <-self.quitC:
+				log.Info("pss shutting down")
+			}
 		}
 	}()
 	log.Debug("Started pss", "public key", common.ToHex(crypto.FromECDSAPub(self.PublicKey())))
@@ -272,8 +287,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.handlersMu.Lock()
+	self.handlersMu.RLock()
-	defer self.handlersMu.Unlock()
+	defer self.handlersMu.RUnlock()
 	return self.handlers[topic]
 }
 
@@ -286,20 +301,13 @@ func (self *Pss) handlePssMsg(msg interface{}) error {
 	if ok {
 		var err error
 		if !self.isSelfPossibleRecipient(pssmsg) {
-			msgexp := time.Unix(int64(pssmsg.Expire), 0)
-			if msgexp.Before(time.Now()) {
-				log.Trace("pss expired :/ ... dropping")
-				return nil
-			} else if msgexp.After(time.Now().Add(self.msgTTL)) {
-				return errors.New("Invalid TTL")
-			}
 			log.Trace("pss was for someone else :'( ... forwarding", "pss", common.ToHex(self.BaseAddr()))
-			return self.forward(pssmsg)
+			self.outbox <- pssmsg
 		}
 		log.Trace("pss for us, yay! ... let's process!", "pss", common.ToHex(self.BaseAddr()))
 
 		if !self.process(pssmsg) {
-			err = self.forward(pssmsg)
+			self.outbox <- pssmsg
 		}
 		return err
 	}
@@ -335,10 +343,7 @@ func (self *Pss) process(pssmsg *PssMsg) bool {
 
 	if len(pssmsg.To) < addressLength {
 		go func() {
-			err := self.forward(pssmsg)
+			self.outbox <- pssmsg
-			if err != nil {
-				log.Warn("Redundant forward fail: %v", err)
-			}
 		}()
 	}
 	handlers := self.getHandlers(psstopic)
@@ -401,7 +406,7 @@ func (self *Pss) generateSymmetricKey(topic Topic, address *PssAddress, addToCac
 	if err != nil {
 		return "", err
 	}
-	self.addSymmetricKeyToPool(keyid, topic, address, addToCache)
+	self.addSymmetricKeyToPool(keyid, topic, address, addToCache, false)
 	return keyid, nil
 }
 
@@ -418,20 +423,25 @@ func (self *Pss) generateSymmetricKey(topic Topic, address *PssAddress, addToCac
 // Returns a string id that can be used to retrieve the key bytes
 // from the whisper backend (see pss.GetSymmetricKey())
 func (self *Pss) SetSymmetricKey(key []byte, topic Topic, address *PssAddress, addtocache bool) (string, error) {
+	return self.setSymmetricKey(key, topic, address, addtocache, true)
+}
+
+func (self *Pss) setSymmetricKey(key []byte, topic Topic, address *PssAddress, addtocache bool, protected bool) (string, error) {
 	keyid, err := self.w.AddSymKeyDirect(key)
 	if err != nil {
 		return "", err
 	}
-	self.addSymmetricKeyToPool(keyid, topic, address, addtocache)
+	self.addSymmetricKeyToPool(keyid, topic, address, addtocache, protected)
 	return keyid, nil
 }
 
 // adds a symmetric key to the pss key pool, and optionally adds the key
 // 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) {
+func (self *Pss) addSymmetricKeyToPool(keyid string, topic Topic, address *PssAddress, addtocache bool, protected bool) {
 	psp := &pssPeer{
-		address: address,
+		address:   address,
+		protected: protected,
 	}
 	self.symKeyPoolMu.Lock()
 	if _, ok := self.symKeyPool[keyid]; !ok {
@@ -644,29 +654,23 @@ func (self *Pss) send(to []byte, topic Topic, msg []byte, asymmetric bool, key [
 		Expire:  uint32(time.Now().Add(self.msgTTL).Unix()),
 		Payload: envelope,
 	}
-	return self.forward(pssmsg)
+	self.outbox <- pssmsg
+	return nil
 }
 
 // Forwards a pss message to the peer(s) closest to the to recipient address in the PssMsg struct
 // The recipient address can be of any length, and the byte slice will be matched to the MSB slice
 // of the peer address of the equivalent length.
-func (self *Pss) forward(msg *PssMsg) error {
+func (self *Pss) forward(msg *PssMsg) {
 	to := make([]byte, addressLength)
 	copy(to[:len(msg.To)], msg.To)
 
-	// cache the message
+	// message hash
 	digest, err := self.storeMsg(msg)
 	if err != nil {
 		log.Warn(fmt.Sprintf("could not store message %v to cache: %v", msg, err))
 	}
 
-	// flood guard:
-	// don't allow identical messages we saw shortly before
-	if self.checkFwdCache(nil, digest) {
-		log.Trace(fmt.Sprintf("pss relay block-cache match: FROM %x TO %x", self.Overlay.BaseAddr(), common.ToHex(msg.To)))
-		return nil
-	}
-
 	// send with kademlia
 	// find the closest peer to the recipient and attempt to send
 	sent := 0
@@ -696,7 +700,9 @@ func (self *Pss) forward(msg *PssMsg) error {
 
 		// 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())
+		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
@@ -730,11 +736,12 @@ func (self *Pss) forward(msg *PssMsg) error {
 
 	if sent == 0 {
 		log.Debug("unable to forward to any peers")
-		return nil
+		time.Sleep(time.Millisecond)
+		self.outbox <- msg
 	}
 
+	// cache the message
 	self.addFwdCache(digest)
-	return nil
 }
 
 /////////////////////////////////////////////////////////////////////
@@ -757,8 +764,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.fwdCacheMu.Lock()
+	self.fwdCacheMu.RLock()
-	defer self.fwdCacheMu.Unlock()
+	defer self.fwdCacheMu.RUnlock()
 	entry, ok := self.fwdCache[digest]
 	if ok {
 		if entry.expiresAt.After(time.Now()) {
@@ -785,3 +792,11 @@ func (self *Pss) storeMsg(msg *PssMsg) (pssDigest, error) {
 	copy(digest[:], key[:digestLength])
 	return digest, nil
 }
+
+func (self *Pss) isMsgExpired(msg *PssMsg) bool {
+	msgexp := time.Unix(int64(msg.Expire), 0)
+	if msgexp.Before(time.Now()) || msgexp.After(time.Now().Add(self.msgTTL)) {
+		return true
+	}
+	return false
+}

swarm/pss/pss_test.go

@@ -477,7 +477,7 @@ func testSymSend(t *testing.T) {
 	if err != nil {
 		t.Fatal(err)
 	}
-	err = clients[1].Call(&rkeyids, "psstest_setSymKeys", rpubkeyhex, rrecvkey, lrecvkey, defaultSymKeySendLimit, topic, loaddrhex)
+	err = clients[1].Call(&rkeyids, "psstest_setSymKeys", lpubkeyhex, rrecvkey, lrecvkey, defaultSymKeySendLimit, topic, loaddrhex)
 	if err != nil {
 		t.Fatal(err)
 	}
@@ -504,7 +504,7 @@ func testSymSend(t *testing.T) {
 	select {
 	case recvmsg := <-rmsgC:
 		if !bytes.Equal(recvmsg.Msg, rmsg) {
-			t.Fatalf("node 2 received payload mismatch: expected %v, got %v", rmsg, recvmsg.Msg)
+			t.Fatalf("node 2 received payload mismatch: expected %x, got %v", rmsg, recvmsg.Msg)
 		}
 	case cerr := <-rctx.Done():
 		t.Fatalf("test message timed out: %v", cerr)