Fixed routing logic.

* Fixed various race conditions in 'app' module. * Fix chance where app.appConn does not get closed. * Fixed various memory leaks in managedTransport (where channels are not closed). * Improved transport logging so that we don't need to write to disk on every network operation.
skycoin · Jul 2, 2019 · 25cbf6d · 25cbf6d
1 parent 91283de
commit 25cbf6d
Show file tree

Hide file tree

Showing 13 changed files with 262 additions and 144 deletions.
diff --git a/cmd/apps/skychat/chat.go b/cmd/apps/skychat/chat.go
@@ -108,10 +108,10 @@ func messageHandler(w http.ResponseWriter, req *http.Request) {
 
 	addr := &app.Addr{PubKey: pk, Port: 1}
 	connsMu.Lock()
-	conn := chatConns[pk]
+	conn, ok := chatConns[pk]
 	connsMu.Unlock()
 
-	if conn == nil {
+	if !ok {
 		var err error
 		err = r.Do(func() error {
 			conn, err = chatApp.Dial(addr)

diff --git a/integration/test-messaging.sh b/integration/test-messaging.sh
@@ -1,4 +1,4 @@
 #!/usr/bin/env bash
 source ./integration/generic/env-vars.sh
-# curl --data  {'"recipient":"'$PK_A'", "message":"Hello Joe!"}' -X POST  $CHAT_C  
+curl --data  {'"recipient":"'$PK_A'", "message":"Hello Joe!"}' -X POST  $CHAT_C
 curl --data  {'"recipient":"'$PK_C'", "message":"Hello Mike!"}' -X POST  $CHAT_A
diff --git a/pkg/app/app.go b/pkg/app/app.go
@@ -166,6 +166,8 @@ func (app *App) handleProto() {
 }
 
 func (app *App) serveConn(addr *LoopAddr, conn io.ReadWriteCloser) {
+	defer conn.Close()
+
 	for {
 		buf := make([]byte, 32*1024)
 		n, err := conn.Read(buf)
@@ -179,11 +181,10 @@ func (app *App) serveConn(addr *LoopAddr, conn io.ReadWriteCloser) {
 		}
 	}
 
-	if app.conns[*addr] != nil {
+	app.mu.Lock()
+	if _, ok := app.conns[*addr]; ok {
 		app.proto.Send(FrameClose, &addr, nil) // nolint: errcheck
 	}
-
-	app.mu.Lock()
 	delete(app.conns, *addr)
 	app.mu.Unlock()
 }
@@ -251,13 +252,12 @@ func (app *App) confirmLoop(data []byte) error {
 
 type appConn struct {
 	net.Conn
-	rw    io.ReadWriteCloser
-	laddr *Addr
-	raddr *Addr
+	laddr  *Addr
+	raddr  *Addr
 }
 
 func newAppConn(conn net.Conn, laddr, raddr *Addr) *appConn {
-	return &appConn{conn, conn, laddr, raddr}
+	return &appConn{conn, laddr, raddr}
 }
 
 func (conn *appConn) LocalAddr() net.Addr {
@@ -269,13 +269,13 @@ func (conn *appConn) RemoteAddr() net.Addr {
 }
 
 func (conn *appConn) Write(p []byte) (n int, err error) {
-	return conn.rw.Write(p)
+	return conn.Conn.Write(p)
 }
 
 func (conn *appConn) Read(p []byte) (n int, err error) {
-	return conn.rw.Read(p)
+	return conn.Conn.Read(p)
 }
 
 func (conn *appConn) Close() error {
-	return conn.rw.Close()
+	return conn.Conn.Close()
 }
diff --git a/pkg/node/rpc.go b/pkg/node/rpc.go
@@ -55,7 +55,7 @@ func newTransportSummary(tm *transport.Manager, tp *transport.ManagedTransport,
 	}
 
 	summary := &TransportSummary{
-		ID:      tp.ID,
+		ID:      tp.Entry.ID,
 		Local:   tm.Local(),
 		Remote:  remote,
 		Type:    tp.Type(),

diff --git a/pkg/node/rpc_test.go b/pkg/node/rpc_test.go
@@ -246,7 +246,7 @@ func TestRPC(t *testing.T) {
 	t.Run("Transport", func(t *testing.T) {
 		var ids []uuid.UUID
 		node.tm.WalkTransports(func(tp *transport.ManagedTransport) bool {
-			ids = append(ids, tp.ID)
+			ids = append(ids, tp.Entry.ID)
 			return true
 		})
 

diff --git a/pkg/router/router.go b/pkg/router/router.go
@@ -98,6 +98,7 @@ func (r *Router) Serve(ctx context.Context) error {
 			}
 
 			go func(tp transport.Transport) {
+				defer tp.Close()
 				for {
 					if err := serve(tp); err != nil {
 						if err != io.EOF {
@@ -423,7 +424,7 @@ func (r *Router) setupProto(ctx context.Context) (*setup.Protocol, transport.Tra
 	// TODO(evanlinjin): need string constant for tp type.
 	tr, err := r.tm.CreateTransport(ctx, r.config.SetupNodes[0], dmsg.Type, false)
 	if err != nil {
-		return nil, nil, fmt.Errorf("transport: %s", err)
+		return nil, nil, fmt.Errorf("setup transport: %s", err)
 	}
 
 	sProto := setup.NewSetupProtocol(tr)

diff --git a/pkg/router/router_test.go b/pkg/router/router_test.go
@@ -85,7 +85,7 @@ func TestRouterForwarding(t *testing.T) {
 	tr3, err := m3.CreateTransport(context.TODO(), pk2, "mock2", true)
 	require.NoError(t, err)
 
-	rule := routing.ForwardRule(time.Now().Add(time.Hour), 4, tr3.ID)
+	rule := routing.ForwardRule(time.Now().Add(time.Hour), 4, tr3.Entry.ID)
 	routeID, err := rt.AddRule(rule)
 	require.NoError(t, err)
 
@@ -197,9 +197,9 @@ func TestRouterApp(t *testing.T) {
 
 	ni1, ni2 := noiseInstances(t, pk1, pk2, sk1, sk2)
 	raddr := &app.Addr{PubKey: pk2, Port: 5}
-	require.NoError(t, r.pm.SetLoop(6, raddr, &loop{tr.ID, 4, ni1}))
+	require.NoError(t, r.pm.SetLoop(6, raddr, &loop{tr.Entry.ID, 4, ni1}))
 
-	tr2 := m2.Transport(tr.ID)
+	tr2 := m2.Transport(tr.Entry.ID)
 	go proto.Send(app.FrameSend, &app.Packet{Addr: &app.LoopAddr{Port: 6, Remote: *raddr}, Payload: []byte("bar")}, nil) // nolint: errcheck
 
 	packet := make(routing.Packet, 29)
@@ -333,13 +333,13 @@ func TestRouterSetup(t *testing.T) {
 
 	var routeID routing.RouteID
 	t.Run("add route", func(t *testing.T) {
-		routeID, err = setup.AddRule(sProto, routing.ForwardRule(time.Now().Add(time.Hour), 2, tr.ID))
+		routeID, err = setup.AddRule(sProto, routing.ForwardRule(time.Now().Add(time.Hour), 2, tr.Entry.ID))
 		require.NoError(t, err)
 
 		rule, err := rt.Rule(routeID)
 		require.NoError(t, err)
 		assert.Equal(t, routing.RouteID(2), rule.RouteID())
-		assert.Equal(t, tr.ID, rule.TransportID())
+		assert.Equal(t, tr.Entry.ID, rule.TransportID())
 	})
 
 	t.Run("`confirm loop - responder", func(t *testing.T) {
@@ -371,7 +371,7 @@ func TestRouterSetup(t *testing.T) {
 		loop, err := r.pm.GetLoop(2, &app.Addr{PubKey: pk2, Port: 1})
 		require.NoError(t, err)
 		require.NotNil(t, loop)
-		assert.Equal(t, tr.ID, loop.trID)
+		assert.Equal(t, tr.Entry.ID, loop.trID)
 		assert.Equal(t, routing.RouteID(2), loop.routeID)
 
 		addrs := [2]*app.Addr{}
@@ -427,7 +427,7 @@ func TestRouterSetup(t *testing.T) {
 		l, err := r.pm.GetLoop(2, &app.Addr{PubKey: pk2, Port: 1})
 		require.NoError(t, err)
 		require.NotNil(t, l)
-		assert.Equal(t, tr.ID, l.trID)
+		assert.Equal(t, tr.Entry.ID, l.trID)
 		assert.Equal(t, routing.RouteID(2), l.routeID)
 
 		addrs := [2]*app.Addr{}

diff --git a/pkg/setup/node_test.go b/pkg/setup/node_test.go
@@ -99,12 +99,12 @@ func TestCreateLoop(t *testing.T) {
 
 	l := &routing.Loop{LocalPort: 1, RemotePort: 2, Expiry: time.Now().Add(time.Hour),
 		Forward: routing.Route{
-			&routing.Hop{From: pk1, To: pk2, Transport: tr1.ID},
-			&routing.Hop{From: pk2, To: pk3, Transport: tr3.ID},
+			&routing.Hop{From: pk1, To: pk2, Transport: tr1.Entry.ID},
+			&routing.Hop{From: pk2, To: pk3, Transport: tr3.Entry.ID},
 		},
 		Reverse: routing.Route{
-			&routing.Hop{From: pk3, To: pk2, Transport: tr3.ID},
-			&routing.Hop{From: pk2, To: pk1, Transport: tr1.ID},
+			&routing.Hop{From: pk3, To: pk2, Transport: tr3.Entry.ID},
+			&routing.Hop{From: pk2, To: pk1, Transport: tr1.Entry.ID},
 		},
 	}
 
@@ -132,25 +132,25 @@ func TestCreateLoop(t *testing.T) {
 	assert.Equal(t, uint16(1), rule.LocalPort())
 	rule = rules[2]
 	assert.Equal(t, routing.RuleForward, rule.Type())
-	assert.Equal(t, tr1.ID, rule.TransportID())
+	assert.Equal(t, tr1.Entry.ID, rule.TransportID())
 	assert.Equal(t, routing.RouteID(2), rule.RouteID())
 
 	rules = n2.getRules()
 	require.Len(t, rules, 2)
 	rule = rules[1]
 	assert.Equal(t, routing.RuleForward, rule.Type())
-	assert.Equal(t, tr1.ID, rule.TransportID())
+	assert.Equal(t, tr1.Entry.ID, rule.TransportID())
 	assert.Equal(t, routing.RouteID(1), rule.RouteID())
 	rule = rules[2]
 	assert.Equal(t, routing.RuleForward, rule.Type())
-	assert.Equal(t, tr3.ID, rule.TransportID())
+	assert.Equal(t, tr3.Entry.ID, rule.TransportID())
 	assert.Equal(t, routing.RouteID(2), rule.RouteID())
 
 	rules = n3.getRules()
 	require.Len(t, rules, 2)
 	rule = rules[1]
 	assert.Equal(t, routing.RuleForward, rule.Type())
-	assert.Equal(t, tr3.ID, rule.TransportID())
+	assert.Equal(t, tr3.Entry.ID, rule.TransportID())
 	assert.Equal(t, routing.RouteID(1), rule.RouteID())
 	rule = rules[2]
 	assert.Equal(t, routing.RuleApp, rule.Type())

diff --git a/pkg/transport/log.go b/pkg/transport/log.go
@@ -2,6 +2,7 @@ package transport
 
 import (
 	"encoding/json"
+	"errors"
 	"fmt"
 	"math/big"
 	"os"
@@ -14,8 +15,69 @@ import (
 // LogEntry represents a logging entry for a given Transport.
 // The entry is updated every time a packet is received or sent.
 type LogEntry struct {
-	ReceivedBytes *big.Int `json:"received"` // Total received bytes.
-	SentBytes     *big.Int `json:"sent"`     // Total sent bytes.
+	RecvBytes big.Int `json:"recv"` // Total received bytes.
+	SentBytes big.Int `json:"sent"` // Total sent bytes.
+	rMx       sync.Mutex
+	sMx       sync.Mutex
+}
+
+// AddRecv records read.
+func (le *LogEntry) AddRecv(n int64) {
+	le.rMx.Lock()
+	le.RecvBytes.Add(&le.RecvBytes, big.NewInt(n))
+	le.rMx.Unlock()
+}
+
+// AddSent records write.
+func (le *LogEntry) AddSent(n int64) {
+	le.sMx.Lock()
+	le.SentBytes.Add(&le.SentBytes, big.NewInt(n))
+	le.sMx.Unlock()
+}
+
+// MarshalJSON implements json.Marshaller
+func (le *LogEntry) MarshalJSON() ([]byte, error) {
+	le.rMx.Lock()
+	recv := le.RecvBytes.String()
+	le.rMx.Unlock()
+
+	le.sMx.Lock()
+	sent := le.SentBytes.String()
+	le.sMx.Unlock()
+
+	data := `{"recv":` + recv + `,"sent":` + sent + `}`
+	return []byte(data), nil
+}
+
+// GobEncode implements gob.GobEncoder
+func (le *LogEntry) GobEncode() ([]byte, error) {
+	le.rMx.Lock()
+	rb, err := le.RecvBytes.GobEncode()
+	le.rMx.Unlock()
+	if err != nil {
+		return nil, err
+	}
+	le.sMx.Lock()
+	sb, err := le.SentBytes.GobEncode()
+	le.sMx.Unlock()
+	if err != nil {
+		return nil, err
+	}
+	return append(rb, sb...), err
+}
+
+// GobDecode implements gob.GobDecoder
+func (le *LogEntry) GobDecode(b []byte) error {
+	le.rMx.Lock()
+	err := le.RecvBytes.GobDecode(b)
+	le.rMx.Unlock()
+	if err != nil {
+		return err
+	}
+	le.sMx.Lock()
+	err = le.SentBytes.GobDecode(b)
+	le.sMx.Unlock()
+	return err
 }
 
 // LogStore stores transport log entries.
@@ -32,14 +94,17 @@ type inMemoryTransportLogStore struct {
 // InMemoryTransportLogStore implements in-memory TransportLogStore.
 func InMemoryTransportLogStore() LogStore {
 	return &inMemoryTransportLogStore{
-		entries: map[uuid.UUID]*LogEntry{},
+		entries: make(map[uuid.UUID]*LogEntry),
 	}
 }
 
 func (tls *inMemoryTransportLogStore) Entry(id uuid.UUID) (*LogEntry, error) {
 	tls.mu.Lock()
-	entry := tls.entries[id]
+	entry, ok := tls.entries[id]
 	tls.mu.Unlock()
+	if !ok {
+		return entry, errors.New("not found")
+	}
 
 	return entry, nil
 }

diff --git a/pkg/transport/log_test.go b/pkg/transport/log_test.go
@@ -1,8 +1,9 @@
 package transport_test
 
 import (
+	"encoding/json"
+	"fmt"
 	"io/ioutil"
-	"math/big"
 	"os"
 	"testing"
 
@@ -17,16 +18,21 @@ func testTransportLogStore(t *testing.T, logStore transport.LogStore) {
 	t.Helper()
 
 	id1 := uuid.New()
-	entry1 := &transport.LogEntry{big.NewInt(100), big.NewInt(200)}
+	entry1 := new(transport.LogEntry)
+	entry1.AddRecv(100)
+	entry1.AddSent(200)
+
 	id2 := uuid.New()
-	entry2 := &transport.LogEntry{big.NewInt(300), big.NewInt(400)}
+	entry2 := new(transport.LogEntry)
+	entry2.AddRecv(300)
+	entry2.AddSent(400)
 
 	require.NoError(t, logStore.Record(id1, entry1))
 	require.NoError(t, logStore.Record(id2, entry2))
 
 	entry, err := logStore.Entry(id2)
 	require.NoError(t, err)
-	assert.Equal(t, int64(300), entry.ReceivedBytes.Int64())
+	assert.Equal(t, int64(300), entry.RecvBytes.Int64())
 	assert.Equal(t, int64(400), entry.SentBytes.Int64())
 }
 
@@ -43,3 +49,24 @@ func TestFileTransportLogStore(t *testing.T) {
 	require.NoError(t, err)
 	testTransportLogStore(t, ls)
 }
+
+func TestLogEntry_MarshalJSON(t *testing.T) {
+	entry := new(transport.LogEntry)
+	entry.AddSent(10)
+	entry.AddRecv(100)
+	b, err := json.Marshal(entry)
+	require.NoError(t, err)
+	fmt.Println(string(b))
+	b, err = json.MarshalIndent(entry, "", "\t")
+	require.NoError(t, err)
+	fmt.Println(string(b))
+}
+
+func TestLogEntry_GobEncode(t *testing.T) {
+	var entry transport.LogEntry
+
+	enc, err := entry.GobEncode()
+	require.NoError(t, err)
+
+	require.NoError(t, entry.GobDecode(enc))
+}