Fix comm stack resource consumtion

- set reasonable read buffer size
- cleanup subcon when they are closed
- more efficient sessionID computation
- remove delayed subcon close
- remove delayed context deletion after view execution

Signed-off-by: Marcus Brandenburger <bur@zurich.ibm.com>

Author: mbrandenburger

platform/view/services/comm/host/rest/host.go

@@ -9,8 +9,8 @@ package rest
 import (
 	"context"
 	"crypto/tls"
-	"fmt"
 	"net/http"
+	"strings"
 
 	"github.com/hyperledger-labs/fabric-smart-client/pkg/utils/errors"
 	"github.com/hyperledger-labs/fabric-smart-client/platform/common/services/logging"
@@ -86,5 +86,13 @@ func (h *host) Close() error {
 func (h *host) Wait() {}
 
 func StreamHash(info host2.StreamInfo) host2.StreamHash {
-	return fmt.Sprintf("%s.%s.%s.%s", info.RemotePeerID, info.RemotePeerAddress, info.SessionID, info.ContextID)
+	var sb strings.Builder
+	sb.WriteString(info.RemotePeerID)
+	sb.WriteRune('.')
+	sb.WriteString(info.RemotePeerAddress)
+	sb.WriteRune('.')
+	sb.WriteString(info.SessionID)
+	sb.WriteRune('.')
+	sb.WriteString(info.ContextID)
+	return sb.String()
 }

platform/view/services/comm/host/rest/websocket/multiplexed_provider.go

@@ -275,10 +275,7 @@ func (c *multiplexedServerConn) readIncoming(newStreamCallback func(pStream host
 			logger.Debugf("server subconn errored: %v", mm.Err)
 		} else if mm.Err != "" {
 			logger.Debugf("Server subconn [%s] errored: %v", mm.ID, mm.Err)
-			go func() {
-				time.Sleep(1 * time.Second) // TODO: Find the point when the connection must close
-				_ = sc.Close()
-			}()
+			_ = sc.Close()
 		} else {
 			sc.receiverChan <- result{value: mm.Msg}
 		}
@@ -429,5 +426,10 @@ func (c *subConn) Close() error {
 	// try to send closing handshake but ignore any error (in case connection is already closed)
 	_ = c.parentConn.write(MultiplexedMessage{ID: c.id, Err: io.EOF.Error()})
 
+	// we need to clean up the parents subConns map
+	c.parentConn.mu.Lock()
+	delete(c.parentConn.subConns, c.id)
+	c.parentConn.mu.Unlock()
+
 	return nil
 }

platform/view/services/comm/host/rest/websocket/stream.go

@@ -14,6 +14,7 @@ import (
 	"github.com/hyperledger-labs/fabric-smart-client/platform/common/services/logging"
 	host2 "github.com/hyperledger-labs/fabric-smart-client/platform/view/services/comm/host"
 	"github.com/hyperledger-labs/fabric-smart-client/platform/view/services/comm/host/rest"
+	"go.uber.org/zap/zapcore"
 )
 
 type connection interface {
@@ -70,7 +71,9 @@ func (s *stream) readMessages(ctx context.Context) {
 				_ = s.Close()
 				return
 			}
-			logger.Debugf("Read message of length [%d] on [%s]", len(msg), s.Hash())
+			if logger.IsEnabledFor(zapcore.DebugLevel) {
+				logger.Debugf("Read message of length [%d] on [%s]", len(msg), s.Hash())
+			}
 			s.reads <- result{value: msg, err: err}
 		}
 	}

platform/view/services/comm/host/rest/websocket/stream_test.go

@@ -21,6 +21,7 @@ import (
 	"github.com/hyperledger-labs/fabric-smart-client/platform/view/services/comm/host"
 	"github.com/hyperledger-labs/fabric-smart-client/platform/view/services/comm/host/rest/websocket"
 	"github.com/stretchr/testify/assert"
+	"go.uber.org/goleak"
 )
 
 func newMockStream(conn *mockConn) host.P2PStream {
@@ -65,6 +66,9 @@ func (c *mockConn) WrittenValues() <-chan []byte {
 }
 
 func TestWriter(t *testing.T) {
+	// let check that at the end of this test all our go routines are stopped
+	defer goleak.VerifyNone(t, goleak.IgnoreCurrent())
+
 	conn := &mockConn{
 		written: make(chan []byte, 100),
 		read:    make(chan []byte, 100),
@@ -100,6 +104,9 @@ func TestWriter(t *testing.T) {
 }
 
 func TestReader(t *testing.T) {
+	// let check that at the end of this test all our go routines are stopped
+	defer goleak.VerifyNone(t, goleak.IgnoreCurrent())
+
 	conn := &mockConn{
 		written: make(chan []byte, 100),
 		read:    make(chan []byte, 100),
@@ -117,23 +124,19 @@ func TestReader(t *testing.T) {
 		assert.NoError(t, conn.ReadValue(message))
 	}
 	wg := sync.WaitGroup{}
-	wg.Add(len(input))
+	wg.Add(1)
 
-	output := make([]*comm.ViewPacket, 0, len(input))
-	m := sync.RWMutex{}
 	go func() {
-		for {
+		defer wg.Done()
+		for _, in := range input {
 			read := &comm.ViewPacket{}
 			assert.NoError(t, r.ReadMsg(read))
-			m.Lock()
-			output = append(output, read)
-			m.Unlock()
-			wg.Done()
+			assert.True(t, proto.Equal(in, read))
 		}
 	}()
 	wg.Wait()
 
-	assert.Equal(t, len(input), len(output))
+	assert.NoError(t, stream.Close())
 }
 
 func messageOfSize(size int) proto.Message {

platform/view/services/comm/ids.go

@@ -6,8 +6,13 @@ SPDX-License-Identifier: Apache-2.0
 
 package comm
 
-import "encoding/base64"
+import (
+	"crypto/sha256"
+	"encoding/hex"
+)
 
 func computeInternalSessionID(topic string, pkid []byte) string {
-	return topic + "." + base64.StdEncoding.EncodeToString(pkid)
+	hasher := sha256.New()
+	hasher.Write(pkid)
+	return topic + "." + hex.EncodeToString(hasher.Sum(nil))
 }

platform/view/services/comm/p2p.go

@@ -24,9 +24,10 @@ import (
 )
 
 const (
-	masterSession                    = "master of puppets I'm pulling your strings"
-	contextIDLabel tracing.LabelName = "context_id"
-	sessionIDLabel tracing.LabelName = "session_id"
+	masterSession                       = "master of puppets I'm pulling your strings"
+	contextIDLabel    tracing.LabelName = "context_id"
+	sessionIDLabel    tracing.LabelName = "session_id"
+	defaultBufferSize                   = 4096
 )
 
 var errStreamNotFound = errors.New("stream not found")
@@ -209,7 +210,7 @@ func (p *P2PNode) handleIncomingStream(stream host2.P2PStream) {
 func (p *P2PNode) handleStream(stream host2.P2PStream) {
 	sh := &streamHandler{
 		stream: stream,
-		reader: io.NewVarintProtoReader(stream, 655360*2),
+		reader: io.NewVarintProtoReader(stream, defaultBufferSize),
 		writer: io.NewVarintProtoWriter(stream),
 		node:   p,
 	}

platform/view/services/comm/session.go

@@ -12,6 +12,7 @@ import (
 
 	"github.com/hyperledger-labs/fabric-smart-client/platform/view/services/comm/host"
 	"github.com/hyperledger-labs/fabric-smart-client/platform/view/view"
+	"go.uber.org/zap/zapcore"
 )
 
 // NetworkStreamSession implements view.Session
@@ -82,7 +83,9 @@ func (n *NetworkStreamSession) closeInternal() {
 	logger.Debugf("closing session [%s] with [%d] streams", n.sessionID, len(n.streams))
 	toClose := make([]*streamHandler, 0, len(n.streams))
 	for stream := range n.streams {
-		logger.Debugf("session [%s], stream [%s], refCtr [%d]", n.sessionID, stream.stream.Hash(), stream.refCtr)
+		if logger.IsEnabledFor(zapcore.DebugLevel) {
+			logger.Debugf("session [%s], stream [%s], refCtr [%d]", n.sessionID, stream.stream.Hash(), stream.refCtr)
+		}
 		stream.refCtr--
 		if stream.refCtr == 0 {
 			toClose = append(toClose, stream)

platform/view/services/view/manager.go

@@ -11,7 +11,6 @@ import (
 	"reflect"
 	"runtime/debug"
 	"sync"
-	"time"
 
 	"github.com/hyperledger-labs/fabric-smart-client/pkg/utils/errors"
 	"github.com/hyperledger-labs/fabric-smart-client/platform/common/services/logging"
@@ -277,12 +276,7 @@ func (cm *Manager) respond(responder view.View, id view.Identity, msg *view.Mess
 	if isNew {
 		// delete context at the end of the execution
 		res, err = func(ctx view.Context, responder view.View) (interface{}, error) {
-			defer func() {
-				// TODO: this is a workaround
-				// give some time to flush anything can be in queues
-				time.Sleep(5 * time.Second)
-				cm.deleteContext(id, ctx.ID())
-			}()
+			defer cm.deleteContext(id, ctx.ID())
 			return ctx.RunView(responder)
 		}(ctx, responder)
 	} else {