pss: refactored outbox into own package, touched pss_tests and pss to accommodate changes

Author: aperonam

pss/outbox/message.go

@@ -0,0 +1,17 @@
+package outbox
+
+import (
+	"time"
+)
+
+type outboxMsg struct {
+	msg       interface{}
+	startedAt time.Time
+}
+
+func NewOutboxMessage(msg interface{}) *outboxMsg {
+	return &outboxMsg{
+		msg:       msg,
+		startedAt: time.Now(),
+	}
+}

pss/outbox/mock.go

@@ -0,0 +1,22 @@
+package outbox
+
+import (
+	"github.com/ethersphere/swarm/log"
+)
+
+func NewOutboxMock(config *Config) (outbox *outbox) {
+	if config.Forward == nil {
+		config.Forward = func(msg interface{}) error {
+			log.Debug("Forwarded message", "msg", msg)
+			return nil
+		}
+	}
+
+	if &config.NumberSlots == nil {
+		config.NumberSlots = defaultOutboxCapacity
+	}
+	if config.QuitC == nil {
+		config.QuitC = make(chan struct{})
+	}
+	return NewOutbox(config)
+}

pss/outbox/outbox.go

@@ -0,0 +1,108 @@
+package outbox
+
+import (
+	"errors"
+	"github.com/ethereum/go-ethereum/metrics"
+	"github.com/ethersphere/swarm/log"
+)
+
+const (
+	defaultOutboxCapacity = 100000
+)
+
+type Outbox interface {
+	Enqueue(outboxmsg *outboxMsg) error
+	ProcessOutbox()
+	SetForwardFunction(forwardFunc func(msg interface{}) error) //Eventually this will not be needed, when pss has a newMock that creates a mock outbox, this is used just for a test outside of outbox package
+}
+
+type Config struct {
+	NumberSlots int
+	QuitC       chan struct{}
+	Forward     func(msg interface{}) error
+}
+
+type outbox struct {
+	Config
+	queue   []*outboxMsg
+	slots   chan int
+	process chan int
+}
+
+func NewOutbox(config *Config) *outbox {
+	outbox := outbox{
+		Config:  *config,
+		process: make(chan int),
+		slots:   make(chan int, config.NumberSlots),
+		queue:   make([]*outboxMsg, config.NumberSlots),
+	}
+	// fill up outbox slots
+	for i := 0; i < cap(outbox.slots); i++ {
+		outbox.slots <- i
+	}
+	return &outbox
+}
+
+// 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) SetForwardFunction(forwardFunc func(msg interface{}) error) {
+	o.Forward = forwardFunc
+}
+
+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:
+	}
+}
+
+func (o *outbox) len() int {
+	return cap(o.slots) - len(o.slots)
+}

pss/outbox/outbox_test.go

@@ -0,0 +1,133 @@
+package outbox
+
+import (
+	"errors"
+	"github.com/ethersphere/swarm/log"
+	"testing"
+	"time"
+)
+
+type testMsg struct {
+}
+
+func TestMessageOutbox(t *testing.T) {
+
+	outboxCapacity := 2
+	successC := make(chan struct{})
+	forward := func(msg interface{}) error {
+		successC <- struct{}{}
+		return nil
+	}
+
+	outbox := NewOutboxMock(&Config{
+		NumberSlots: outboxCapacity,
+		QuitC:       nil,
+		Forward:     forward,
+	})
+
+	outboxMessage := NewOutboxMessage(testMsg{})
+
+	go outbox.ProcessOutbox()
+
+	err := outbox.Enqueue(outboxMessage)
+	if err != nil {
+		t.Fatalf("expected no error, got %v", err)
+	}
+
+	usedSlots := outbox.len()
+	if usedSlots != 1 {
+		t.Fatalf("incorrect outbox length. expected 1, got %v", usedSlots)
+	}
+	t.Log("Message enqueued", "Outbox len", outbox.len())
+
+	select {
+	case <-successC:
+	case <-time.After(2 * time.Second):
+		t.Fatal("timeout waiting for success forward")
+	}
+
+	failed := make([]interface{}, 0)
+	failedC := make(chan struct{})
+	continueC := make(chan struct{})
+	failedForward := func(msg interface{}) error {
+		failed = append(failed, msg)
+		failedC <- struct{}{}
+		<-continueC
+		return errors.New("forced test error forwarding message")
+	}
+
+	outbox.Forward = failedForward
+
+	err = outbox.Enqueue(outboxMessage)
+	if err != nil {
+		t.Fatalf("Expected no error enqueing, got %v", err.Error())
+	}
+
+	select {
+	case <-failedC:
+	case <-time.After(2 * time.Second):
+		t.Fatal("timeout waiting for failing forward")
+	}
+
+	if len(failed) == 0 {
+		t.Fatal("Incorrect number of failed messages, expected 1 got 0")
+	}
+	// The message will be retried once we send to continueC, so first, we change the forward function
+	outbox.Forward = forward
+	continueC <- struct{}{}
+	select {
+	case <-successC:
+	case <-time.After(2 * time.Second):
+		t.Fatal("Timeout waiting for second success forward")
+	}
+}
+
+func TestOutboxFull(t *testing.T) {
+
+	addr := make([]byte, 32)
+	addr[0] = 0x01
+	outboxCapacity := 2
+
+	procChan := make(chan struct{})
+	succesForward := func(msg interface{}) error {
+		<-procChan
+		log.Info("Message processed")
+		return nil
+	}
+
+	outbox := NewOutboxMock(&Config{
+		NumberSlots: outboxCapacity,
+		QuitC:       nil,
+		Forward:     succesForward,
+	})
+	go outbox.ProcessOutbox()
+
+	err := outbox.Enqueue(outboxTestMessage())
+	if err != nil {
+		t.Fatalf("expected no error enqueing first message, got %v", err)
+	}
+	err = outbox.Enqueue(outboxTestMessage())
+	if err != nil {
+		t.Fatalf("expected no error enqueing second message, got %v", err)
+	}
+	//As we haven't signaled procChan, the messages are still in the outbox
+
+	err = outbox.Enqueue(outboxTestMessage())
+	if err == nil {
+		t.Fatalf("expected error enqueing third message, instead got nil")
+	}
+	procChan <- struct{}{}
+	//There should be a slot again in the outbox
+	select {
+	case <-outbox.slots:
+	case <-time.After(2 * time.Second):
+		t.Fatalf("timeout waiting for a free slot")
+	}
+}
+
+func outboxTestMessage() *outboxMsg {
+	return &outboxMsg{
+		msg:       nil,
+		startedAt: time.Time{},
+	}
+}