Testing with proper locking

Makefile

@@ -31,7 +31,12 @@ FLAKY_TESTS_MINOWS := (Test_session_Recv_SessionEnded)
 # test runs all tests in DELA without coverage
 # It first runs all the tests in "short" mode, so the flaky tests don't run.
 # Then the flaky tests get run separately for at most 3 times, and hopefully it all works out.
-test: tidy
+test: 
+	for count in $$(seq 10); do \
+		go test -count=1 ./mino/minows -run="${FLAKY_TESTS_MINOWS}"; \
+	done
+
+test_temp:
 	go test ./... -short -count=1 || exit 1
 	@for count in $$( seq 4 ); do \
 		if [ "$$count" -eq 4 ]; then \

mino/minows/rpc_test.go

@@ -6,6 +6,7 @@ import (
 	"go.dedis.ch/dela/mino"
 	"go.dedis.ch/dela/serde"
 	"go.dedis.ch/dela/testing/fake"
+	"sync"
 	"testing"
 )
 
@@ -244,8 +245,14 @@ func Test_rpc_Stream_ContextCancelled(t *testing.T) {
 // echos back the same message
 // - implements mino.Handler
 type echoHandler struct {
-	from     []mino.Address
-	messages []serde.Message
+	from       []mino.Address
+	messages   []serde.Message
+	mutex      sync.Mutex
+	msgCounter chan struct{}
+}
+
+func newEchoHandler() *echoHandler {
+	return &echoHandler{msgCounter: make(chan struct{}, 100)}
 }
 
 func (h *echoHandler) Process(req mino.Request) (resp serde.Message,
@@ -261,15 +268,24 @@ func (h *echoHandler) Stream(out mino.Sender, in mino.Receiver) error {
 		if err != nil {
 			return err
 		}
+		h.mutex.Lock()
 		h.from = append(h.from, from)
 		h.messages = append(h.messages, msg)
 		err = <-out.Send(msg, from)
+		h.msgCounter <- struct{}{}
+		h.mutex.Unlock()
 		if err != nil {
 			return err
 		}
 	}
 }
 
+func (h *echoHandler) wait(count int) {
+	for i := 0; i < count; i++ {
+		<-h.msgCounter
+	}
+}
+
 func mustCreateRPC(t *testing.T, m mino.Mino, name string,
 	h mino.Handler) mino.RPC {
 	r, err := m.CreateRPC(name, h, fake.MessageFactory{})

mino/minows/session_test.go

@@ -16,7 +16,7 @@ import (
 )
 
 func Test_session_Send(t *testing.T) {
-	handler := &echoHandler{}
+	handler := newEchoHandler()
 	const addrInitiator = "/ip4/127.0.0.1/tcp/6001/ws"
 	initiator, stop := mustCreateMinows(t, addrInitiator, addrInitiator)
 	defer stop()
@@ -45,15 +45,15 @@ func Test_session_Send(t *testing.T) {
 	require.NoError(t, err)
 	require.False(t, open)
 
-	wait()
+	handler.wait(3)
 	require.Equal(t, []mino.Address{s.(*messageHandler).myAddr,
 		s.(*messageHandler).myAddr, s.(*messageHandler).myAddr}, handler.from)
 	require.Equal(t, []serde.Message{fake.Message{},
 		fake.Message{}, fake.Message{}}, handler.messages)
 }
 
 func Test_session_Send_ToSelf(t *testing.T) {
-	handler := &echoHandler{}
+	handler := newEchoHandler()
 	const addrInitiator = "/ip4/127.0.0.1/tcp/6001/ws"
 	initiator, stop := mustCreateMinows(t, addrInitiator, addrInitiator)
 	defer stop()
@@ -77,7 +77,7 @@ func Test_session_Send_ToSelf(t *testing.T) {
 	require.NoError(t, err)
 	require.False(t, open)
 
-	wait()
+	handler.wait(3)
 	require.Equal(t, []mino.Address{s.(*messageHandler).myAddr,
 		s.(*messageHandler).myAddr, s.(*messageHandler).myAddr}, handler.from)
 	require.Equal(t, []serde.Message{fake.Message{},
@@ -86,7 +86,7 @@ func Test_session_Send_ToSelf(t *testing.T) {
 }
 
 func Test_session_Send_WrongAddressType(t *testing.T) {
-	handler := &echoHandler{}
+	handler := newEchoHandler()
 	const addrInitiator = "/ip4/127.0.0.1/tcp/6001/ws"
 	initiator, stop := mustCreateMinows(t, addrInitiator, addrInitiator)
 	defer stop()
@@ -105,7 +105,7 @@ func Test_session_Send_WrongAddressType(t *testing.T) {
 }
 
 func Test_session_Send_AddressNotPlayer(t *testing.T) {
-	handler := &echoHandler{}
+	handler := newEchoHandler()
 	const addrInitiator = "/ip4/127.0.0.1/tcp/6001/ws"
 	initiator, stop := mustCreateMinows(t, addrInitiator, addrInitiator)
 	defer stop()
@@ -126,7 +126,7 @@ func Test_session_Send_AddressNotPlayer(t *testing.T) {
 }
 
 func Test_session_Send_SessionEnded(t *testing.T) {
-	handler := &echoHandler{}
+	handler := newEchoHandler()
 	const addrInitiator = "/ip4/127.0.0.1/tcp/6001/ws"
 	initiator, stop := mustCreateMinows(t, addrInitiator, addrInitiator)
 	defer stop()
@@ -139,7 +139,6 @@ func Test_session_Send_SessionEnded(t *testing.T) {
 
 	s, _, stop := mustStream(t, rpc, initiator, player)
 	stop()
-	wait()
 
 	errs := s.Send(fake.Message{}, initiator.GetAddress(), player.GetAddress())
 	for i := 0; i < 2; i++ {
@@ -154,7 +153,7 @@ func Test_session_Send_SessionEnded(t *testing.T) {
 }
 
 func Test_session_Recv(t *testing.T) {
-	handler := &echoHandler{}
+	handler := newEchoHandler()
 	const addrInitiator = "/ip4/127.0.0.1/tcp/6001/ws"
 	initiator, stop := mustCreateMinows(t, addrInitiator, addrInitiator)
 	defer stop()
@@ -203,7 +202,7 @@ func Test_session_Recv(t *testing.T) {
 }
 
 func Test_session_Recv_FromSelf(t *testing.T) {
-	handler := &echoHandler{}
+	handler := newEchoHandler()
 	const addrInitiator = "/ip4/127.0.0.1/tcp/6001/ws"
 	initiator, stop := mustCreateMinows(t, addrInitiator, addrInitiator)
 	defer stop()
@@ -250,10 +249,10 @@ func Test_session_Recv_FromSelf(t *testing.T) {
 }
 
 func Test_session_Recv_SessionEnded(t *testing.T) {
-	if testing.Short() {
-		t.Skip("Will be tested in a loop")
-	}
-	handler := &echoHandler{}
+	//if testing.Short() {
+	//	t.Skip("Will be tested in a loop")
+	//}
+	handler := newEchoHandler()
 	const addrInitiator = "/ip4/127.0.0.1/tcp/6001/ws"
 	initiator, stop := mustCreateMinows(t, addrInitiator, addrInitiator)
 	defer stop()
@@ -279,7 +278,7 @@ func Test_session_Recv_SessionEnded(t *testing.T) {
 }
 
 func Test_session_Recv_ContextCancelled(t *testing.T) {
-	handler := &echoHandler{}
+	handler := newEchoHandler()
 	const addrInitiator = "/ip4/127.0.0.1/tcp/6001/ws"
 	initiator, stop := mustCreateMinows(t, addrInitiator, addrInitiator)
 	defer stop()
@@ -304,10 +303,6 @@ func setTimeout() (context.Context, context.CancelFunc) {
 	return ctx, cancel
 }
 
-func wait() {
-	time.Sleep(2 * time.Second)
-}
-
 func mustStream(t *testing.T, rpc mino.RPC,
 	minos ...mino.Mino) (mino.Sender, mino.Receiver, func()) {
 	ctx, cancel := context.WithCancel(context.Background())