Merge pull request #386 from ayuryshev/bug/remove-multiple-manageTran…

…sport

Further improvements in ManagedTransport concurrency

pkg/router/router.go

@@ -85,6 +85,8 @@ func (r *Router) Serve(ctx context.Context) error {
 			isAccepted, isSetup := tp.Accepted, r.IsSetupTransport(tp)
 			r.mu.Unlock()
 
+			r.Logger.Infof("New transport: isAccepted: %v, isSetup: %v", isAccepted, isSetup)
+
 			var serve func(io.ReadWriter) error
 			switch {
 			case isAccepted && isSetup:

pkg/transport/discovery_test.go

@@ -1,19 +1,20 @@
-package transport
+package transport_test
 
 import (
 	"context"
 	"fmt"
 
 	"github.com/skycoin/skywire/pkg/cipher"
+	"github.com/skycoin/skywire/pkg/transport"
 )
 
 func ExampleNewDiscoveryMock() {
-	dc := NewDiscoveryMock()
+	dc := transport.NewDiscoveryMock()
 	pk1, _ := cipher.GenerateKeyPair()
 	pk2, _ := cipher.GenerateKeyPair()
-	entry := &Entry{Type: "mock", EdgeKeys: SortPubKeys(pk1, pk2)}
+	entry := &transport.Entry{Type: "mock", EdgeKeys: transport.SortPubKeys(pk1, pk2)}
 
-	sEntry := &SignedEntry{Entry: entry}
+	sEntry := &transport.SignedEntry{Entry: entry}
 
 	if err := dc.RegisterTransports(context.TODO(), sEntry); err == nil {
 		fmt.Println("RegisterTransport success")
@@ -33,7 +34,7 @@ func ExampleNewDiscoveryMock() {
 		fmt.Printf("entriesWS[0].Entry.Edges()[0] == entry.Edges()[0] is %v\n", entriesWS[0].Entry.Edges()[0] == entry.Edges()[0])
 	}
 
-	if _, err := dc.UpdateStatuses(context.TODO(), &Status{}); err == nil {
+	if _, err := dc.UpdateStatuses(context.TODO(), &transport.Status{}); err == nil {
 		fmt.Println("UpdateStatuses success")
 	} else {
 		fmt.Println(err.Error())

pkg/transport/entry_test.go

@@ -1,11 +1,12 @@
-package transport
+package transport_test
 
 import (
 	"fmt"
 
 	"github.com/google/uuid"
 
 	"github.com/skycoin/skywire/pkg/cipher"
+	"github.com/skycoin/skywire/pkg/transport"
 )
 
 // ExampleNewEntry shows that with different order of edges:
@@ -15,8 +16,8 @@ func ExampleNewEntry() {
 	pkA, _ := cipher.GenerateKeyPair()
 	pkB, _ := cipher.GenerateKeyPair()
 
-	entryAB := NewEntry(pkA, pkB, "", true)
-	entryBA := NewEntry(pkB, pkA, "", true)
+	entryAB := transport.NewEntry(pkA, pkB, "", true)
+	entryBA := transport.NewEntry(pkB, pkA, "", true)
 
 	if entryAB.ID == entryBA.ID {
 		fmt.Println("entryAB.ID == entryBA.ID")
@@ -32,14 +33,14 @@ func ExampleEntry_Edges() {
 	pkA, _ := cipher.GenerateKeyPair()
 	pkB, _ := cipher.GenerateKeyPair()
 
-	entryAB := Entry{
+	entryAB := transport.Entry{
 		ID:       uuid.UUID{},
 		EdgeKeys: [2]cipher.PubKey{pkA, pkB},
 		Type:     "",
 		Public:   true,
 	}
 
-	entryBA := Entry{
+	entryBA := transport.Entry{
 		ID:       uuid.UUID{},
 		EdgeKeys: [2]cipher.PubKey{pkB, pkA},
 		Type:     "",
@@ -62,7 +63,7 @@ func ExampleEntry_SetEdges() {
 	pkA, _ := cipher.GenerateKeyPair()
 	pkB, _ := cipher.GenerateKeyPair()
 
-	entryAB, entryBA := Entry{}, Entry{}
+	entryAB, entryBA := transport.Entry{}, transport.Entry{}
 
 	entryAB.SetEdges([2]cipher.PubKey{pkA, pkB})
 	entryBA.SetEdges([2]cipher.PubKey{pkA, pkB})
@@ -85,8 +86,8 @@ func ExampleSignedEntry_Sign() {
 	pkA, skA := cipher.GenerateKeyPair()
 	pkB, skB := cipher.GenerateKeyPair()
 
-	entry := NewEntry(pkA, pkB, "mock", true)
-	sEntry := &SignedEntry{Entry: entry}
+	entry := transport.NewEntry(pkA, pkB, "mock", true)
+	sEntry := &transport.SignedEntry{Entry: entry}
 
 	if sEntry.Signatures[0].Null() && sEntry.Signatures[1].Null() {
 		fmt.Println("No signatures set")
@@ -119,8 +120,8 @@ func ExampleSignedEntry_Signature() {
 	pkA, skA := cipher.GenerateKeyPair()
 	pkB, skB := cipher.GenerateKeyPair()
 
-	entry := NewEntry(pkA, pkB, "mock", true)
-	sEntry := &SignedEntry{Entry: entry}
+	entry := transport.NewEntry(pkA, pkB, "mock", true)
+	sEntry := &transport.SignedEntry{Entry: entry}
 	if ok := sEntry.Sign(pkA, skA); !ok {
 		fmt.Println("Error signing sEntry with (pkA,skA)")
 	}

pkg/transport/handshake_test.go

@@ -199,30 +199,6 @@ func TestSettlementHandshake(t *testing.T) {
 
 }
 
-/*
-func TestSettlementHandshakeInvalidSig(t *testing.T) {
-	mockEnv := newHsMockEnv()
-
-	require.NoError(t, mockEnv.err1)
-	require.NoError(t, mockEnv.err2)
-
-	go settlementInitiatorHandshake(true)(mockEnv.m2, mockEnv.tr1) // nolint: errcheck
-	_, err := settlementResponderHandshake(mockEnv.m2, mockEnv.tr2)
-	require.Error(t, err)
-	assert.Equal(t, "Recovered pubkey does not match pubkey", err.Error())
-
-	in, out := net.Pipe()
-	tr1 := NewMockTransport(in, mockEnv.pk1, mockEnv.pk2)
-	tr2 := NewMockTransport(out, mockEnv.pk2, mockEnv.pk1)
-
-	go settlementResponderHandshake(mockEnv.m1, tr2) // nolint: errcheck
-	_, err = settlementInitiatorHandshake(true)(mockEnv.m1, tr1)
-	require.Error(t, err)
-	assert.Equal(t, "Recovered pubkey does not match pubkey", err.Error())
-
-}
-*/
-
 func TestSettlementHandshakePrivate(t *testing.T) {
 	mockEnv := newHsMockEnv()
 

pkg/transport/log_test.go

@@ -1,4 +1,4 @@
-package transport
+package transport_test
 
 import (
 	"io/ioutil"
@@ -7,17 +7,18 @@ import (
 	"testing"
 
 	"github.com/google/uuid"
+	"github.com/skycoin/skywire/pkg/transport"
 	"github.com/stretchr/testify/assert"
 	"github.com/stretchr/testify/require"
 )
 
-func testTransportLogStore(t *testing.T, logStore LogStore) {
+func testTransportLogStore(t *testing.T, logStore transport.LogStore) {
 	t.Helper()
 
 	id1 := uuid.New()
-	entry1 := &LogEntry{big.NewInt(100), big.NewInt(200)}
+	entry1 := &transport.LogEntry{big.NewInt(100), big.NewInt(200)}
 	id2 := uuid.New()
-	entry2 := &LogEntry{big.NewInt(300), big.NewInt(400)}
+	entry2 := &transport.LogEntry{big.NewInt(300), big.NewInt(400)}
 
 	require.NoError(t, logStore.Record(id1, entry1))
 	require.NoError(t, logStore.Record(id2, entry2))
@@ -29,15 +30,15 @@ func testTransportLogStore(t *testing.T, logStore LogStore) {
 }
 
 func TestInMemoryTransportLogStore(t *testing.T) {
-	testTransportLogStore(t, InMemoryTransportLogStore())
+	testTransportLogStore(t, transport.InMemoryTransportLogStore())
 }
 
 func TestFileTransportLogStore(t *testing.T) {
 	dir, err := ioutil.TempDir("", "log_store")
 	require.NoError(t, err)
 	defer os.RemoveAll(dir)
 
-	ls, err := FileTransportLogStore(dir)
+	ls, err := transport.FileTransportLogStore(dir)
 	require.NoError(t, err)
 	testTransportLogStore(t, ls)
 }

pkg/transport/managed_transport.go

@@ -3,7 +3,6 @@ package transport
 import (
 	"math/big"
 	"sync"
-	"sync/atomic"
 
 	"github.com/google/uuid"
 )
@@ -17,10 +16,10 @@ type ManagedTransport struct {
 	Accepted bool
 	LogEntry *LogEntry
 
-	doneChan  chan struct{}
-	errChan   chan error
-	isClosing int32
-	mu        sync.RWMutex
+	doneChan chan struct{}
+	errChan  chan error
+	mu       sync.RWMutex
+	once     sync.Once
 
 	readLogChan  chan int
 	writeLogChan chan int
@@ -34,8 +33,8 @@ func newManagedTransport(id uuid.UUID, tr Transport, public bool, accepted bool)
 		Accepted:     accepted,
 		doneChan:     make(chan struct{}),
 		errChan:      make(chan error),
-		readLogChan:  make(chan int),
-		writeLogChan: make(chan int),
+		readLogChan:  make(chan int, 16),
+		writeLogChan: make(chan int, 16),
 		LogEntry:     &LogEntry{new(big.Int), new(big.Int)},
 	}
 }
@@ -45,22 +44,12 @@ func (tr *ManagedTransport) Read(p []byte) (n int, err error) {
 	tr.mu.RLock()
 	n, err = tr.Transport.Read(p) // TODO: data race.
 	tr.mu.RUnlock()
-	if err == nil {
-		select {
-		case <-tr.doneChan:
-			return
-		case tr.readLogChan <- n:
-		}
 
-		return
-	}
-
-	select {
-	case <-tr.doneChan:
-		return
-	case tr.errChan <- err:
+	if err != nil {
+		tr.errChan <- err
 	}
 
+	tr.readLogChan <- n
 	return
 }
 
@@ -69,42 +58,41 @@ func (tr *ManagedTransport) Write(p []byte) (n int, err error) {
 	tr.mu.RLock()
 	n, err = tr.Transport.Write(p)
 	tr.mu.RUnlock()
-	if err == nil {
-		select {
-		case <-tr.doneChan:
-			return
-		case tr.writeLogChan <- n:
-		}
 
+	if err != nil {
+		tr.errChan <- err
 		return
 	}
-
-	select {
-	case <-tr.doneChan:
-		return
-	case tr.errChan <- err:
-	}
+	tr.writeLogChan <- n
 
 	return
 }
 
+// killWorker sends signal to Manager.manageTransport goroutine to exit
+// it's safe to call it multiple times
+func (tr *ManagedTransport) killWorker() {
+	tr.once.Do(func() {
+		close(tr.doneChan)
+	})
+}
+
 // Close closes underlying
 func (tr *ManagedTransport) Close() error {
-
-	atomic.StoreInt32(&tr.isClosing, 1)
-
 	tr.mu.RLock()
 	err := tr.Transport.Close()
 	tr.mu.RUnlock()
 
+	tr.killWorker()
+	return err
+}
+
+func (tr *ManagedTransport) isClosing() bool {
 	select {
 	case <-tr.doneChan:
+		return true
 	default:
-
-		close(tr.doneChan)
+		return false
 	}
-
-	return err
 }
 
 func (tr *ManagedTransport) updateTransport(newTr Transport) {