swarm/pss: eliminate handlerChannelListener() goroutine

Complexity got reduced. There is no need for the complex
synchronization pattern.

swarm/pss/prox_test.go

@@ -5,7 +5,6 @@ import (
 	"crypto/ecdsa"
 	"encoding/binary"
 	"fmt"
-	"strings"
 	"sync"
 	"testing"
 	"time"
@@ -37,14 +36,13 @@ type handlerNotification struct {
 }
 
 type testData struct {
-	sim       *simulation.Simulation
+	sim                *simulation.Simulation
-	kademlias map[enode.ID]*network.Kademlia
+	kademlias          map[enode.ID]*network.Kademlia
-	nodeAddrs map[enode.ID][]byte // make predictable overlay addresses from the generated random enode ids
+	nodeAddresses      map[enode.ID][]byte // make predictable overlay addresses from the generated random enode ids
-	senders   map[int]enode.ID    // originating nodes of the messages (intention is to choose as far as possible from the receiving neighborhood)
+	senders            map[int]enode.ID    // originating nodes of the messages (intention is to choose as far as possible from the receiving neighborhood)
-	msgs      [][]byte            // recipient addresses of messages
+	recipientAddresses [][]byte
 
 	requiredMsgCount int
-	allowedMsgCount  int
 	requiredMsgs     map[enode.ID][]uint64 // message serials we expect respective nodes to receive
 	allowedMsgs      map[enode.ID][]uint64 // message serials we expect respective nodes to receive
 
@@ -52,10 +50,6 @@ type testData struct {
 	totalMsgCount int
 	handlerDone   bool // set to true on termination of the simulation run
 	mu            sync.Mutex
-
-	doneC chan struct{}            // terminates the handler channel listener
-	errC  chan error               // error to pass to main sim thread
-	msgC  chan handlerNotification // message receipt notification to main sim thread
 }
 
 var (
@@ -63,63 +57,60 @@ var (
 	topic = BytesToTopic([]byte{0xf3, 0x9e, 0x06, 0x82})
 )
 
-func (d *testData) pushNotification(val handlerNotification) {
+func (td *testData) pushNotification(val handlerNotification) {
-	d.mu.Lock()
+	td.mu.Lock()
-	d.notifications = append(d.notifications, val)
+	td.notifications = append(td.notifications, val)
-	d.mu.Unlock()
+	td.mu.Unlock()
 }
 
-func (d *testData) popNotification() (ret handlerNotification, found bool) {
+func (td *testData) popNotification() (first handlerNotification, ok bool) {
-	d.mu.Lock()
+	td.mu.Lock()
-	if len(d.notifications) > 0 {
+	if len(td.notifications) > 0 {
-		found = true
+		ok = true
-		ret = d.notifications[0]
+		first = td.notifications[0]
-		d.notifications = d.notifications[1:]
+		td.notifications = td.notifications[1:]
 	}
-	d.mu.Unlock()
+	td.mu.Unlock()
-	return ret, found
+	return first, ok
 }
 
-func (d *testData) getMsgCount() int {
+func (td *testData) getMsgCount() int {
-	d.mu.Lock()
+	td.mu.Lock()
-	defer d.mu.Unlock()
+	defer td.mu.Unlock()
-	return d.totalMsgCount
+	return td.totalMsgCount
 }
 
-func (d *testData) incrementMsgCount() int {
+func (td *testData) incrementMsgCount() int {
-	d.mu.Lock()
+	td.mu.Lock()
-	defer d.mu.Unlock()
+	defer td.mu.Unlock()
-	d.totalMsgCount++
+	td.totalMsgCount++
-	return d.totalMsgCount
+	return td.totalMsgCount
 }
 
-func (d *testData) isDone() bool {
+func (td *testData) isDone() bool {
-	d.mu.Lock()
+	td.mu.Lock()
-	defer d.mu.Unlock()
+	defer td.mu.Unlock()
-	return d.handlerDone
+	return td.handlerDone
 }
 
-func (d *testData) setDone() {
+func (td *testData) setDone() {
-	d.mu.Lock()
+	td.mu.Lock()
-	defer d.mu.Unlock()
+	defer td.mu.Unlock()
-	d.handlerDone = true
+	td.handlerDone = true
 }
 
 func newTestData() *testData {
 	return &testData{
-		kademlias:    make(map[enode.ID]*network.Kademlia),
+		kademlias:     make(map[enode.ID]*network.Kademlia),
-		nodeAddrs:    make(map[enode.ID][]byte),
+		nodeAddresses: make(map[enode.ID][]byte),
-		requiredMsgs: make(map[enode.ID][]uint64),
+		requiredMsgs:  make(map[enode.ID][]uint64),
-		allowedMsgs:  make(map[enode.ID][]uint64),
+		allowedMsgs:   make(map[enode.ID][]uint64),
-		senders:      make(map[int]enode.ID),
+		senders:       make(map[int]enode.ID),
-		doneC:        make(chan struct{}),
-		errC:         make(chan error),
-		msgC:         make(chan handlerNotification),
 	}
 }
 
-func (d *testData) getKademlia(nodeId *enode.ID) (*network.Kademlia, error) {
+func (td *testData) getKademlia(nodeId *enode.ID) (*network.Kademlia, error) {
-	kadif, ok := d.sim.NodeItem(*nodeId, simulation.BucketKeyKademlia)
+	kadif, ok := td.sim.NodeItem(*nodeId, simulation.BucketKeyKademlia)
 	if !ok {
 		return nil, fmt.Errorf("no kademlia entry for %v", nodeId)
 	}
@@ -130,29 +121,29 @@ func (d *testData) getKademlia(nodeId *enode.ID) (*network.Kademlia, error) {
 	return kad, nil
 }
 
-func (d *testData) init(msgCount int) error {
+func (td *testData) init(msgCount int) error {
 	log.Debug("TestProxNetwork start")
 
-	for _, nodeId := range d.sim.NodeIDs() {
+	for _, nodeId := range td.sim.NodeIDs() {
-		kad, err := d.getKademlia(&nodeId)
+		kad, err := td.getKademlia(&nodeId)
 		if err != nil {
 			return err
 		}
-		d.nodeAddrs[nodeId] = kad.BaseAddr()
+		td.nodeAddresses[nodeId] = kad.BaseAddr()
 	}
 
 	for i := 0; i < int(msgCount); i++ {
 		msgAddr := pot.RandomAddress() // we choose message addresses randomly
-		d.msgs = append(d.msgs, msgAddr.Bytes())
+		td.recipientAddresses = append(td.recipientAddresses, msgAddr.Bytes())
 		smallestPo := 256
 		var targets []enode.ID
 		var closestPO int
 
 		// loop through all nodes and find the required and allowed recipients of each message
 		// (for more information, please see the comment to the main test function)
-		for _, nod := range d.sim.Net.GetNodes() {
+		for _, nod := range td.sim.Net.GetNodes() {
-			po, _ := pof(d.msgs[i], d.nodeAddrs[nod.ID()], 0)
+			po, _ := pof(td.recipientAddresses[i], td.nodeAddresses[nod.ID()], 0)
-			depth := d.kademlias[nod.ID()].NeighbourhoodDepth()
+			depth := td.kademlias[nod.ID()].NeighbourhoodDepth()
 
 			// only nodes with closest IDs (wrt the msg address) will be required recipients
 			if po > closestPO {
@@ -164,26 +155,25 @@ func (d *testData) init(msgCount int) error {
 			}
 
 			if po >= depth {
-				d.allowedMsgCount++
+				td.allowedMsgs[nod.ID()] = append(td.allowedMsgs[nod.ID()], uint64(i))
-				d.allowedMsgs[nod.ID()] = append(d.allowedMsgs[nod.ID()], uint64(i))
 			}
 
 			// a node with the smallest PO (wrt msg) will be the sender,
 			// in order to increase the distance the msg must travel
 			if po < smallestPo {
 				smallestPo = po
-				d.senders[i] = nod.ID()
+				td.senders[i] = nod.ID()
 			}
 		}
 
-		d.requiredMsgCount += len(targets)
+		td.requiredMsgCount += len(targets)
 		for _, id := range targets {
-			d.requiredMsgs[id] = append(d.requiredMsgs[id], uint64(i))
+			td.requiredMsgs[id] = append(td.requiredMsgs[id], uint64(i))
 		}
 
-		log.Debug("nn for msg", "targets", len(targets), "msgidx", i, "msg", common.Bytes2Hex(msgAddr[:8]), "sender", d.senders[i], "senderpo", smallestPo)
+		log.Debug("nn for msg", "targets", len(targets), "msgidx", i, "msg", common.Bytes2Hex(msgAddr[:8]), "sender", td.senders[i], "senderpo", smallestPo)
 	}
-	log.Debug("msgs to receive", "count", d.requiredMsgCount)
+	log.Debug("recipientAddresses to receive", "count", td.requiredMsgCount)
 	return nil
 }
 
@@ -207,11 +197,10 @@ func (d *testData) init(msgCount int) error {
 // whereas nodes X, Y and Z will be allowed recipients.
 func TestProxNetwork(t *testing.T) {
 	t.Run("16_nodes,_16_messages,_16_seconds", func(t *testing.T) {
-		testProxNetwork(t, 16, 16, 3*time.Second)
+		testProxNetwork(t, 16, 16, 16*time.Second)
 	})
 }
 
-// params in run name: nodes/msgs
 func TestProxNetworkLong(t *testing.T) {
 	if !*longrunning {
 		t.Skip("run with --longrunning flag to run extensive network tests")
@@ -256,7 +245,7 @@ func testProxNetwork(t *testing.T, nodeCount int, msgCount int, timeout time.Dur
 	}
 	result := td.sim.Run(ctx, wrapper) // call the main test function
 	if result.Error != nil {
-		timedOut := strings.Compare(result.Error.Error(), "context deadline exceeded") == 0
+		timedOut := result.Error == context.DeadlineExceeded
 		if !timedOut || td.getMsgCount() < td.requiredMsgCount {
 			t.Fatal(result.Error)
 		}
@@ -265,7 +254,7 @@ func testProxNetwork(t *testing.T, nodeCount int, msgCount int, timeout time.Dur
 
 func (td *testData) sendAllMsgs() error {
 	nodes := make(map[int]*rpc.Client)
-	for i := range td.msgs {
+	for i := range td.recipientAddresses {
 		nodeClient, err := td.sim.Net.GetNode(td.senders[i]).Client()
 		if err != nil {
 			return err
@@ -273,7 +262,7 @@ func (td *testData) sendAllMsgs() error {
 		nodes[i] = nodeClient
 	}
 
-	for i, msg := range td.msgs {
+	for i, msg := range td.recipientAddresses {
 		log.Debug("sending msg", "idx", i, "from", td.senders[i])
 		nodeClient := nodes[i]
 		var uvarByte [8]byte
@@ -285,60 +274,64 @@ func (td *testData) sendAllMsgs() error {
 
 // testRoutine is the main test function, called by Simulation.Run()
 func testRoutine(td *testData, ctx context.Context) error {
-	go handlerChannelListener(td)
 
-	res := td.sendAllMsgs()
+	if err := td.sendAllMsgs(); err != nil {
-	if res != nil {
+		return err
-		return res
 	}
 
-	received := 0
+	isMoreTimeLeft := func() bool {
-
-	for {
 		select {
-		case <-ctx.Done(): // timeout or cancel
+		case <-ctx.Done():
-			td.setDone()
+			return false
-			if td.getMsgCount() < td.requiredMsgCount {
+		default:
-				res = ctx.Err()
+			return true
-			}
-		case err := <-td.errC:
-			// only the first error matters
-			if res == nil && err != nil {
-				res = err
-				td.setDone()
-			}
-		case hn := <-td.msgC:
-			received++
-			log.Debug("msg received", "msgs_received", received, "total_expected", td.requiredMsgCount, "id", hn.id, "serial", hn.serial)
-		case <-td.doneC:
-			return res
 		}
 	}
-}
 
-func handlerChannelListener(d *testData) {
+	hasMoreRound := func(err error, hadMessage bool) bool {
-	for !d.isDone() {
+		return err == nil && (hadMessage || isMoreTimeLeft())
-		notification, exist := d.popNotification()
+	}
-		if exist {
+
-			if d.isAllowedMessage(notification) {
+	var err error
-				d.msgC <- notification //notify the main sim thread
+	received := 0
+	hadMessage := false
+	for oneMoreRound := true; oneMoreRound; oneMoreRound = hasMoreRound(err, hadMessage) {
+		message, hadMessage := td.popNotification()
+
+		if !isMoreTimeLeft() {
+			// Stop handlers from sending more messages.
+			// Note: only best effort, race is possible.
+			td.setDone()
+		}
+
+		if hadMessage {
+			if td.isAllowedMessage(message) {
+				received++
+				log.Debug("msg received", "msgs_received", received, "total_expected", td.requiredMsgCount, "id", message.id, "serial", message.serial)
 			} else {
-				log.Error("message received by wrong recipient", "num", notification.serial)
+				err = fmt.Errorf("message %d received by wrong recipient %v", message.serial, message.id)
-				d.errC <- fmt.Errorf("message %d received by wrong recipient %v", notification.serial, notification.id)
-				break
 			}
 		} else {
-			time.Sleep(time.Millisecond * 32)
+			time.Sleep(32 * time.Millisecond)
 		}
+
+		oneMoreRound = err == nil && (hadMessage || isMoreTimeLeft())
 	}
-	close(d.doneC)
+
-	close(d.errC)
+	if err != nil {
-	close(d.msgC)
+		return err
+	}
+
+	if td.getMsgCount() < td.requiredMsgCount {
+		return ctx.Err()
+	}
+	return nil
+
 }
 
-func (d *testData) isAllowedMessage(n handlerNotification) bool {
+func (td *testData) isAllowedMessage(n handlerNotification) bool {
 	// check if message serial is in expected messages for this recipient
-	for _, s := range d.allowedMsgs[n.id] {
+	for _, s := range td.allowedMsgs[n.id] {
 		if n.serial == s {
 			return true
 		}
@@ -346,10 +339,10 @@ func (d *testData) isAllowedMessage(n handlerNotification) bool {
 	return false
 }
 
-func (d *testData) removeAllowedMessage(id enode.ID, index int) {
+func (td *testData) removeAllowedMessage(id enode.ID, index int) {
-	last := len(d.allowedMsgs[id]) - 1
+	last := len(td.allowedMsgs[id]) - 1
-	d.allowedMsgs[id][index] = d.allowedMsgs[id][last]
+	td.allowedMsgs[id][index] = td.allowedMsgs[id][last]
-	d.allowedMsgs[id] = d.allowedMsgs[id][:last]
+	td.allowedMsgs[id] = td.allowedMsgs[id][:last]
 }
 
 func nodeMsgHandler(td *testData, config *adapters.NodeConfig) *handler {

swarm/pss/pss_test.go

@@ -1364,7 +1364,7 @@ func TestNetwork(t *testing.T) {
 }
 
 // params in run name:
-// nodes/msgs/addrbytes/adaptertype
+// nodes/recipientAddresses/addrbytes/adaptertype
 // if adaptertype is exec uses execadapter, simadapter otherwise
 func TestNetwork2000(t *testing.T) {
 	if !*longrunning {