stress test update + bug fixes (#682)

Signed-off-by: Angelo De Caro <[email protected]>

Author: adecaro

integration/nwo/token/common/ne.go

@@ -71,7 +71,7 @@ func (p *NetworkHandler) DBPath(root string, tms *topology2.TMS) string {
 		filepath.Join(
 			root,
 			fmt.Sprintf("%s_%s_%s", tms.Network, tms.Channel, tms.Namespace)+"_db.sqlite",
-		) + "?_pragma=journal_mode(WAL)&_pragma=busy_timeout(5000)"
+		) + "?_pragma=journal_mode(WAL)&_pragma=busy_timeout(20000)"
 }
 
 func (p *NetworkHandler) FSCNodeKVSDir(uniqueName string) string {

integration/nwo/token/fabric/template.go

@@ -43,7 +43,7 @@ token:
           opts:
             createSchema: true 
             driver: {{ SQLDriver }}    
-            maxOpenConns: 10
+            maxOpenConns: 200
             dataSource: {{ SQLDataSource }}
       {{ if not OnlyUnity }} 
       tokendb:
@@ -53,7 +53,7 @@ token:
             createSchema: true 
             tablePrefix: tokens  
             driver: {{ TokensSQLDriver }}    
-            maxOpenConns: 10
+            maxOpenConns: 200
             dataSource: {{ TokensSQLDataSource }}
       {{ end }}
       {{ if Wallets }}

integration/nwo/token/orion/orion.go

@@ -179,9 +179,6 @@ func (p *NetworkHandler) GenerateExtension(tms *topology2.TMS, node *sfcnode.Nod
 	Expect(os.MkdirAll(p.IdentityDBSQLDataSourceDir(uniqueName), 0775)).ToNot(HaveOccurred(), "failed to create [%s]", p.IdentityDBSQLDataSourceDir(uniqueName))
 
 	t, err := template.New("peer").Funcs(template.FuncMap{
-		"TMSID":   func() string { return tms.ID() },
-		"TMS":     func() *topology2.TMS { return tms },
-		"Wallets": func() *generators.Wallets { return p.GetEntry(tms).Wallets[node.Name] },
 		"IsCustodian": func() bool {
 			custodianNode, ok := tms.BackendParams[Custodian]
 			if !ok {
@@ -192,10 +189,14 @@ func (p *NetworkHandler) GenerateExtension(tms *topology2.TMS, node *sfcnode.Nod
 		"CustodianID": func() string {
 			return tms.BackendParams[Custodian].(string)
 		},
+		"TMSID":               func() string { return tms.ID() },
+		"TMS":                 func() *topology2.TMS { return tms },
+		"Wallets":             func() *generators.Wallets { return p.GetEntry(tms).Wallets[node.Name] },
 		"SQLDriver":           func() string { return GetTokenPersistenceDriver(node.Options) },
 		"SQLDataSource":       func() string { return p.GetSQLDataSource(node.Options, uniqueName, tms) },
 		"TokensSQLDriver":     func() string { return GetTokenPersistenceDriver(node.Options) },
 		"TokensSQLDataSource": func() string { return p.GetTokensSQLDataSource(node.Options, uniqueName, tms) },
+		"OnlyUnity":           func() bool { return common2.IsOnlyUnity(tms) },
 	}).Parse(Extension)
 	Expect(err).NotTo(HaveOccurred())
 

integration/nwo/token/orion/template.go

@@ -6,73 +6,118 @@ SPDX-License-Identifier: Apache-2.0
 
 package orion
 
-const Extension = `
+const (
+	Extension = `
 token:
-  tms: 
+  # TMS stands for Token Management Service. A TMS is uniquely identified by a network, channel, and 
+  # namespace identifiers. The network identifier should refer to a configure network (Fabric, Orion, and so on).
+  # The meaning of channel and namespace are network dependant. For Fabric, the meaning is clear.
+  # For Orion, channel is empty and namespace is the DB name to use.
+  tms:
     {{ TMSID }}:
+      # Network identifier this TMS refers to
       network: {{ TMS.Network }}
+      # Channel identifier within the specified network
       channel: {{ TMS.Channel }}
+      # Namespace identifier within the specified channel
       namespace: {{ TMS.Namespace }}
       orion:
         custodian:
           id: {{ CustodianID }}
           enabled: {{ IsCustodian }}
       certification: 
-        interactive:
+        {{ if TMS.Certifiers }} interactive:
           ids: {{ range TMS.Certifiers }}
-          - {{ . }}{{ end }}
+          - {{ . }}{{ end }}{{ end }}
       db:
         persistence:
+          type: unity
           opts:
             createSchema: true 
             driver: {{ SQLDriver }}    
-            maxOpenConns: 10
+            maxOpenConns: 200
             dataSource: {{ SQLDataSource }}
-      identitydb:
-        persistence:
-          type: unity
-      ttxdb:
-        persistence:
-          type: unity
-      auditdb:
-        persistence:
-          type: unity
-      tokenlockdb:
-        persistence:
-          type: unity
+      {{ if not OnlyUnity }} 
       tokendb:
         persistence:
           type: sql
           opts:
             createSchema: true 
             tablePrefix: tokens  
             driver: {{ TokensSQLDriver }}    
-            maxOpenConns: 10
+            maxOpenConns: 200
             dataSource: {{ TokensSQLDataSource }}
-      {{ if Wallets }}wallets:{{ if Wallets.Certifiers }}
+      {{ end }}
+      {{ if Wallets }}
+      # Wallets associated with this TMS
+      wallets:
+        defaultCacheSize: 200
+        {{ if Wallets.Certifiers }}
+        # Certifiers wallets are used to certify tokens
         certifiers: {{ range Wallets.Certifiers }}
         - id: {{ .ID }}
           default: {{ .Default }}
           path: {{ .Path }}
         {{ end }}
-      {{ end }}{{ if Wallets.Issuers }}
+        {{ end }}{{ if Wallets.Issuers }}
+        # Issuers wallets are used to issue tokens
         issuers: {{ range Wallets.Issuers }}
         - id: {{ .ID }}
           default: {{ .Default }}
           path: {{ .Path }}
+          opts:
+            BCCSP:
+              Default: {{ .Opts.Default }}
+              # Settings for the SW crypto provider (i.e. when DEFAULT: SW)
+              SW:
+                 Hash: {{ .Opts.SW.Hash }}
+                 Security: {{ .Opts.SW.Security }}
+              # Settings for the PKCS#11 crypto provider (i.e. when DEFAULT: PKCS11)
+              PKCS11:
+                 # Location of the PKCS11 module library
+                 Library: {{ .Opts.PKCS11.Library }}
+                 # Token Label
+                 Label: {{ .Opts.PKCS11.Label }}
+                 # User PIN
+                 Pin: {{ .Opts.PKCS11.Pin }}
+                 Hash: {{ .Opts.PKCS11.Hash }}
+                 Security: {{ .Opts.PKCS11.Security }}
         {{ end }}
       {{ end }}{{ if Wallets.Owners }}
+        # Owners wallets are used to own tokens
         owners: {{ range Wallets.Owners }}
         - id: {{ .ID }}
           default: {{ .Default }}
           path: {{ .Path }}
+          {{ if .Type }}
+          type: {{ .Type }}
+          {{ end }}
         {{ end }}
       {{ end }}{{ if Wallets.Auditors }}
+        # Auditors wallets are used to audit tokens
         auditors: {{ range Wallets.Auditors }}
         - id: {{ .ID }}
           default: {{ .Default }}
           path: {{ .Path }}
+          opts:
+            BCCSP:
+              Default: {{ .Opts.Default }}
+              # Settings for the SW crypto provider (i.e. when DEFAULT: SW)
+              SW:
+                 Hash: {{ .Opts.SW.Hash }}
+                 Security: {{ .Opts.SW.Security }}
+              # Settings for the PKCS#11 crypto provider (i.e. when DEFAULT: PKCS11)
+              PKCS11:
+                 # Location of the PKCS11 module library
+                 Library: {{ .Opts.PKCS11.Library }}
+                 # Token Label
+                 Label: {{ .Opts.PKCS11.Label }}
+                 # User PIN
+                 Pin: {{ .Opts.PKCS11.Pin }}
+                 Hash: {{ .Opts.PKCS11.Hash }}
+                 Security: {{ .Opts.PKCS11.Security }}
         {{ end }}
       {{ end }}
       {{ end }}
-`
+  `
+)

integration/ports.go

@@ -44,7 +44,9 @@ var (
 		ReplicationFactor: token.None,
 	}
 
-	WebSocketNoReplicationOnly = []*InfrastructureType{WebSocketNoReplication}
+	WebSocketNoReplicationOnly = []*InfrastructureType{
+		WebSocketNoReplication,
+	}
 
 	AllTestTypes = []*InfrastructureType{
 		WebSocketNoReplication,

integration/token/common/views/finality.go

@@ -8,7 +8,6 @@ package views
 
 import (
 	"encoding/json"
-	"fmt"
 
 	"github.com/hyperledger-labs/fabric-smart-client/platform/view/services/assert"
 	"github.com/hyperledger-labs/fabric-smart-client/platform/view/view"
@@ -69,6 +68,6 @@ type finalityListener struct {
 }
 
 func (l *finalityListener) OnStatus(txID string, status int, message string, tokenRequestHash []byte) {
-	fmt.Printf("Received finality from network for TX [%s][%d]", txID, status)
+	//fmt.Printf("Received finality from network for TX [%s][%d]", txID, status)
 	l.errs <- nil
 }

integration/token/fungible/dlogstress/dlog_test.go

@@ -9,7 +9,7 @@ package dlog
 import (
 	"github.com/hyperledger-labs/fabric-smart-client/integration/nwo/fsc"
 	"github.com/hyperledger-labs/fabric-smart-client/pkg/api"
-	token2 "github.com/hyperledger-labs/fabric-token-sdk/integration/token"
+	"github.com/hyperledger-labs/fabric-token-sdk/integration/token"
 	"github.com/hyperledger-labs/fabric-token-sdk/integration/token/common"
 	"github.com/hyperledger-labs/fabric-token-sdk/integration/token/common/sdk/fdlog"
 	"github.com/hyperledger-labs/fabric-token-sdk/integration/token/common/sdk/odlog"
@@ -19,7 +19,10 @@ import (
 )
 
 var _ = Describe("Stress EndToEnd", func() {
-	for _, backend := range []string{"fabric", "orion"} {
+	for _, backend := range []string{
+		"fabric",
+		"orion",
+	} {
 		Describe("Stress test", Label(backend), func() {
 			ts, selector := newTestSuite(backend)
 			AfterEach(ts.TearDown)
@@ -35,16 +38,17 @@ var sdks = map[string]api.SDK{
 	"orion":  &odlog.SDK{},
 }
 
-func newTestSuite(backend string) (*token2.TestSuite, *token2.ReplicaSelector) {
-	opts, selector := token2.NewReplicationOptions(token2.None)
-	ts := token2.NewTestSuite(opts.SQLConfigs, StartPortDlog, topology.Topology(
+func newTestSuite(backend string) (*token.TestSuite, *token.ReplicaSelector) {
+	//opts, selector := token.NewReplicationOptions(token.None)
+	opts, selector := token.NewReplicationOptions(1, "alice", "bob", "charlie", "issuer", "auditor")
+	ts := token.NewTestSuite(opts.SQLConfigs, StartPortDlog, topology.Topology(
 		common.Opts{
 			Backend:         backend,
 			TokenSDKDriver:  "dlog",
 			Aries:           true,
 			ReplicationOpts: opts,
-			CommType:        fsc.WebSocket,
-			//FSCLogSpec:     "token-sdk=debug:fabric-sdk=debug:info",
+			CommType:        fsc.LibP2P,
+			//FSCLogSpec:      "token-sdk=debug:orion-sdk=debug:info",
 			SDKs:       []api.SDK{sdks[backend]},
 			Monitoring: true,
 		},

integration/token/fungible/dlogstress/support/worker.go

@@ -7,32 +7,87 @@ SPDX-License-Identifier: Apache-2.0
 package support
 
 import (
+	"context"
+	"fmt"
 	"sync"
+	"sync/atomic"
+	"time"
 )
 
 type Task func()
 
 type Pool struct {
 	taskQueue chan Task
 	wg        sync.WaitGroup
-	shutdown  chan struct{}
+	ctx       context.Context
+	cancel    context.CancelFunc
+	label     string
+	stop      atomic.Bool
 }
 
-func NewPool(numWorkers int) *Pool {
+func NewPool(label string, numWorkers int) *Pool {
+	ctx, cancel := context.WithCancel(context.Background())
 	pool := &Pool{
 		taskQueue: make(chan Task, numWorkers),
 		wg:        sync.WaitGroup{},
+		ctx:       ctx,
+		cancel:    cancel,
+		label:     label,
 	}
 
 	pool.wg.Add(numWorkers)
+	pool.stop.Store(false)
 	for i := 0; i < numWorkers; i++ {
 		go func() {
-			defer pool.wg.Done()
+			counter := atomic.Int64{}
+			var taskDurations []time.Duration
+			workerStart := time.Now()
+
+			defer func() {
+				fmt.Printf("Context done for [%s], shutdown after, computing statistics\n", pool.label)
+				workerEnd := time.Since(workerStart)
+				var sum int64
+				numTasks := counter.Load()
+				for _, duration := range taskDurations {
+					sum += duration.Milliseconds()
+				}
+				avgDuration := sum / numTasks
+				throughput := workerEnd.Milliseconds() / numTasks
+
+				fmt.Printf(
+					"Context done for [%s], shutdown after # task [%d], avg duration [%d], throughput (%d,%d): [%v] \n",
+					pool.label,
+					numTasks,
+					avgDuration,
+					throughput,
+					workerEnd.Milliseconds(),
+					taskDurations,
+				)
+				pool.wg.Done()
+			}()
 			for {
-				if task, ok := <-pool.taskQueue; ok {
-					task()
-				} else {
+				select {
+				case <-pool.ctx.Done():
+					fmt.Printf("Context done for [%s], shutdown after # %d workers\n", pool.label, numWorkers)
+					pool.stop.Store(true)
 					return
+				case task := <-pool.taskQueue:
+					if pool.stop.Load() {
+						fmt.Printf("No more tasks for [%s] to run...\n", pool.label)
+						return
+					}
+					if task == nil {
+						fmt.Printf("Got nil task for [%s], shutdown...\n", pool.label)
+						return
+					}
+					fmt.Printf("Schedule new task for [%s]: [%d]\n", pool.label, counter.Add(1))
+					start := time.Now()
+					task()
+					end := time.Since(start)
+					taskDurations = append(taskDurations, end)
+					fmt.Printf("Task for [%s][%d], took [%v]\n", pool.label, counter.Load(), end.Milliseconds())
+				default:
+					fmt.Printf("Nothing to do for [%s]\n", pool.label)
 				}
 			}
 		}()
@@ -41,17 +96,33 @@ func NewPool(numWorkers int) *Pool {
 }
 
 func (p *Pool) ScheduleTask(task Task) {
-	for {
-		select {
-		case <-p.shutdown:
-			return
-		default:
-			p.taskQueue <- task
+	go func() {
+		for {
+			select {
+			case <-p.ctx.Done():
+				fmt.Printf("Context done for [%s], shutdown scheduler\n", p.label)
+				p.stop.Store(true)
+				return
+			default:
+				if p.stop.Load() {
+					fmt.Printf("Stop for [%s], shutdown scheduler\n", p.label)
+					return
+				}
+				p.taskQueue <- task
+			}
 		}
-	}
+	}()
+}
+
+func (p *Pool) Stop() {
+	fmt.Printf("Shutting down workers for [%s]\n", p.label)
+	p.stop.Store(true)
+	fmt.Printf("Shutting down workers for [%s], cancel context...\n", p.label)
+	p.cancel()
 }
 
-func (p *Pool) Shutdown() {
-	close(p.taskQueue)
+func (p *Pool) Wait() {
+	fmt.Printf("Shutting down workers for [%s], wait workers...\n", p.label)
 	p.wg.Wait()
+	fmt.Printf("All workers shut down for [%s]\n", p.label)
 }