pss: Improve pressure backstop queue handling - no mutex (ethersphere#1695)

reimplement outbox so failed messages could be reenqueued and process messages in parallel

Author: kortatu

pss/forwarding_test.go

@@ -259,8 +259,11 @@ func testForwardMsg(t *testing.T, ps *Pss, c *testCase) {
 	}
 
 	msg := newTestMsg(recipientAddr)
-	ps.forward(msg)
+	err := ps.forward(msg)
 
+	if err != nil {
+		t.Fatal(fmt.Sprintf("test [%s]\nmsg can't be forwarded. Expected no error but got %v", c.name, err.Error()))
+	}
 	// check test results
 	var fail bool
 	precision := len(recipientAddr)

pss/pss.go

@@ -117,6 +117,90 @@ func (params *Params) WithPrivateKey(privatekey *ecdsa.PrivateKey) *Params {
 	return params
 }
 
+type outbox struct {
+	queue   []*outboxMsg
+	slots   chan int
+	process chan int
+	quitC   chan struct{}
+	forward func(msg *PssMsg) error
+}
+
+func newOutbox(capacity int, quitC chan struct{}, forward func(msg *PssMsg) error) outbox {
+	outbox := outbox{
+		queue:   make([]*outboxMsg, capacity),
+		slots:   make(chan int, capacity),
+		process: make(chan int),
+		quitC:   quitC,
+		forward: forward,
+	}
+	// fill up outbox slots
+	for i := 0; i < cap(outbox.slots); i++ {
+		outbox.slots <- i
+	}
+	return outbox
+}
+
+func (o outbox) len() int {
+	return cap(o.slots) - len(o.slots)
+}
+
+// enqueue a new element in the outbox if there is any slot available.
+// Then send it to process. This method is blocking in the process channel!
+func (o *outbox) enqueue(outboxmsg *outboxMsg) error {
+	// first we try to obtain a slot in the outbox
+	select {
+	case slot := <-o.slots:
+		o.queue[slot] = outboxmsg
+		metrics.GetOrRegisterGauge("pss.outbox.len", nil).Update(int64(o.len()))
+		// we send this message slot to process
+		select {
+		case o.process <- slot:
+		case <-o.quitC:
+		}
+		return nil
+	default:
+		metrics.GetOrRegisterCounter("pss.enqueue.outbox.full", nil).Inc(1)
+		return errors.New("outbox full")
+	}
+}
+
+func (o *outbox) processOutbox() {
+	for slot := range o.process {
+		go func(slot int) {
+			msg := o.msg(slot)
+			metrics.GetOrRegisterResettingTimer("pss.handle.outbox", nil).UpdateSince(msg.startedAt)
+			if err := o.forward(msg.msg); err != nil {
+				metrics.GetOrRegisterCounter("pss.forward.err", nil).Inc(1)
+				// if we failed to forward, re-insert message in the queue
+				log.Debug(err.Error())
+				// reenqueue the message for processing
+				o.reenqueue(slot)
+				log.Debug("Message re-enqued", "slot", slot)
+				return
+			}
+			// free the outbox slot
+			o.free(slot)
+			metrics.GetOrRegisterGauge("pss.outbox.len", nil).Update(int64(o.len()))
+		}(slot)
+	}
+}
+
+func (o outbox) msg(slot int) *outboxMsg {
+	return o.queue[slot]
+}
+
+func (o outbox) free(slot int) {
+	o.slots <- slot
+}
+
+func (o outbox) reenqueue(slot int) {
+	select {
+	case o.process <- slot:
+	case <-o.quitC:
+	}
+
+}
+
 // Pss is the top-level struct, which takes care of message sending, receiving, decryption and encryption, message handler dispatchers
 // and message forwarding. Implements node.Service
 type Pss struct {
@@ -136,7 +220,7 @@ type Pss struct {
 	msgTTL          time.Duration
 	paddingByteSize int
 	capstring       string
-	outbox          chan *outboxMsg
+	outbox          outbox
 
 	// message handling
 	handlers           map[Topic]map[*handler]bool // topic and version based pss payload handlers. See pss.Handle()
@@ -178,7 +262,6 @@ func New(k *network.Kademlia, params *Params) (*Pss, error) {
 		msgTTL:          params.MsgTTL,
 		paddingByteSize: defaultPaddingByteSize,
 		capstring:       c.String(),
-		outbox:          make(chan *outboxMsg, defaultOutboxCapacity),
 
 		handlers:         make(map[Topic]map[*handler]bool),
 		topicHandlerCaps: make(map[Topic]*handlerCaps),
@@ -189,6 +272,7 @@ func New(k *network.Kademlia, params *Params) (*Pss, error) {
 			},
 		},
 	}
+	ps.outbox = newOutbox(defaultOutboxCapacity, ps.quitC, ps.forward)
 
 	for i := 0; i < hasherCount; i++ {
 		hashfunc := storage.MakeHashFunc(storage.DefaultHash)()
@@ -219,23 +303,10 @@ func (p *Pss) Start(srv *p2p.Server) error {
 			}
 		}
 	}()
-	go func() {
-		for {
-			select {
-			case msg := <-p.outbox:
-				metrics.GetOrRegisterGauge("pss.outbox.len", nil).Update(int64(len(p.outbox)))
-
-				err := p.forward(msg.msg)
-				if err != nil {
-					log.Error(err.Error())
-					metrics.GetOrRegisterCounter("pss.forward.err", nil).Inc(1)
-				}
-				metrics.GetOrRegisterResettingTimer("pss.handle.outbox", nil).UpdateSince(msg.startedAt)
-			case <-p.quitC:
-				return
-			}
-		}
-	}()
+
+	// Forward outbox messages
+	go p.outbox.processOutbox()
+
 	log.Info("Started Pss")
 	log.Info("Loaded EC keys", "pubkey", common.ToHex(crypto.FromECDSAPub(p.PublicKey())), "secp256", common.ToHex(crypto.CompressPubkey(p.PublicKey())))
 	return nil
@@ -557,16 +628,8 @@ func (p *Pss) enqueue(msg *PssMsg) error {
 	defer metrics.GetOrRegisterResettingTimer("pss.enqueue", nil).UpdateSince(time.Now())
 
 	outboxmsg := newOutboxMsg(msg)
-	select {
-	case p.outbox <- outboxmsg:
-		metrics.GetOrRegisterGauge("pss.outbox.len", nil).Update(int64(len(p.outbox)))
 
-		return nil
-	default:
-	}
-
-	metrics.GetOrRegisterCounter("pss.enqueue.outbox.full", nil).Inc(1)
-	return errors.New("outbox full")
+	return p.outbox.enqueue(outboxmsg)
 }
 
 // Send a raw message (any encryption is responsibility of calling client)
@@ -730,8 +793,8 @@ func sendMsg(p *Pss, sp *network.Peer, msg *PssMsg) bool {
 // node, then it will be forwarded to all the nearest neighbours of the forwarding node. In case of
 // partial address, it should be forwarded to all the peers matching the partial address, if there
 // are any; otherwise only to one peer, closest to the recipient address. In any case, if the message
-// forwarding fails, the node should try to forward it to the next best peer, until the message is
-// successfully forwarded to at least one peer.
+//// forwarding fails, the node should try to forward it to the next best peer, until the message is
+//// successfully forwarded to at least one peer.
 func (p *Pss) forward(msg *PssMsg) error {
 	metrics.GetOrRegisterCounter("pss.forward", nil).Inc(1)
 	sent := 0 // number of successful sends
@@ -779,19 +842,14 @@ func (p *Pss) forward(msg *PssMsg) error {
 		return true
 	})
 
-	// if we failed to send to anyone, re-insert message in the send-queue
-	if sent == 0 {
-		log.Debug("unable to forward to any peers")
-		if err := p.enqueue(msg); err != nil {
-			metrics.GetOrRegisterCounter("pss.forward.enqueue.error", nil).Inc(1)
-			log.Error(err.Error())
-			return err
-		}
-	}
-
 	// cache the message
 	p.addFwdCache(msg)
-	return nil
+
+	if sent == 0 {
+		return errors.New("unable to forward to any peers")
+	} else {
+		return nil
+	}
 }
 
 /////////////////////////////////////////////////////////////////////