add configurable maxIdleConns and maxIdleTime (#713)

Signed-off-by: Arne Rutjes <arne123@gmail.com>

Author: arner

docs/core-fabric.md

@@ -480,8 +480,10 @@ persistence:
     dataSource: file:/some/path/fsc.sqlite&_txlock=immediate
     tablePrefix: fsc  # optional
     skipCreateTable: true # tells FSC _not_ to create a table when starting up (because it already exists).
-    skipPragmas: true # if this is false, the pragmas we set in the datasource will be overridden with the defaults.
-    maxOpenConns: 50  # optional: max open read connections to the database. Defaults to unlimited.
+    skipPragmas: true # if this is false, the pragmas we set in the datasource will be overridden with the defaults (sqlite specific).
+    maxOpenConns: 20  # optional: max open read connections to the database. Defaults to unlimited. See https://go.dev/doc/database/manage-connections.
+    maxIdleConns: 20  # optional: max idle read connections to the database. Defaults to 2.
+    maxIdleTime: 30s  # optional: max duration a connection can be idle before it is closed. Defaults to 1 minute.
 ```
 
 By default we set the following pragmas (unless you do `skipPragmas: true`. Make sure you always have `_pragma=journal_mode(WAL`):
@@ -510,4 +512,11 @@ persistence:
   opts:
     driver: postgres
     dataSource: host=localhost port=5432 user=postgres password=example dbname=tokendb sslmode=disable
+    maxOpenConns: 25  # optional: max open read connections to the database. Defaults to unlimited. 
+    maxIdleConns: 25  # optional: max idle read connections to the database. Defaults to 2.
+    maxIdleTime: 30s  # optional: max duration a connection can be idle before it is closed. Defaults to 1 minute.
 ```
+
+For more info about managing connections, see https://go.dev/doc/database/manage-connections. Keep in mind that Fabric Smart Client
+maintains two independent database instances: one for KVS and one for the Vault. The combined maxOpenConns should not exceed the
+configured max_connections in the postgres server (100 by default).

platform/common/core/generic/vault/db/helpers.go

@@ -52,7 +52,7 @@ func openBadger[V any](tempDir, dir string, open opener[V]) (V, error) {
 
 type dbConfig common.Opts
 
-func (c *dbConfig) IsSet(string) bool { panic("not supported") }
+func (c *dbConfig) IsSet(string) bool { return false }
 func (c *dbConfig) UnmarshalKey(key string, rawVal interface{}) error {
 	if len(key) > 0 {
 		return errors.New("invalid key")

platform/view/services/db/dbtest/bench.go

@@ -8,6 +8,7 @@ package dbtest
 
 import (
 	"fmt"
+	"math"
 	"testing"
 
 	"github.com/hyperledger-labs/fabric-smart-client/platform/view/services/db/driver"
@@ -69,6 +70,8 @@ func WriteOne(b *testing.B, db driver.TransactionalVersionedPersistence) {
 func WriteMany(b *testing.B, db driver.TransactionalVersionedPersistence) {
 	var err error
 	var k string
+	b.Logf("before: %+v", db.Stats())
+	mid := math.Round(float64(b.N) / 2)
 	b.ResetTimer()
 	for i := 0; i < b.N; i++ {
 		k = fmt.Sprintf("key_%d", i)
@@ -78,9 +81,50 @@ func WriteMany(b *testing.B, db driver.TransactionalVersionedPersistence) {
 		err = db.SetState(namespace, k, driver.VersionedValue{Raw: payload})
 		_ = err
 		err = db.Commit()
+
+		if i == int(mid) {
+			b.Logf("mid: %+v", db.Stats())
+		}
 	}
 	b.StopTimer()
 	returnErr = err
 	assert.NoError(b, returnErr)
 	b.Logf("%.0f writes per second to different keys", float64(b.N)/b.Elapsed().Seconds())
+	b.Logf("after: %+v", db.Stats())
+}
+
+func WriteParallel(b *testing.B, db driver.TransactionalUnversionedPersistence) {
+	var err error
+	var k string
+	var i int
+	b.Logf("    before: %+v", db.Stats())
+	mid := math.Round(float64(b.N) / 2)
+	b.ResetTimer()
+
+	b.RunParallel(func(pb *testing.PB) {
+		for pb.Next() {
+			i++                          // i is not unique because of paralellism but that's ok
+			k = fmt.Sprintf("key_%d", i) // this could be optimized by moving the key creation out of the benchmark
+			if i == int(mid) {
+				b.Logf("    mid (%d): %+v", int(mid), db.Stats())
+			}
+
+			tx, err := db.NewWriteTransaction()
+			if err != nil {
+				b.Error(err)
+			}
+			if err := tx.SetState(namespace, k, payload); err != nil {
+				b.Error(err)
+			}
+			if err := tx.Commit(); err != nil {
+				b.Error(err)
+			}
+		}
+	})
+
+	b.StopTimer()
+	returnErr = err
+	assert.NoError(b, returnErr)
+	b.Logf("    after: %+v", db.Stats())
+	b.Logf("%.0f writes per second to different keys", float64(b.N)/b.Elapsed().Seconds())
 }

platform/view/services/db/driver/badger/badger.go

@@ -248,6 +248,10 @@ func (db *DB) NewWriteTransaction() (driver.WriteTransaction, error) {
 	}, nil
 }
 
+func (db *DB) Stats() any {
+	return db.db.BlockCacheMetrics()
+}
+
 type txMgr interface {
 	Commit() error
 	Discard() error

platform/view/services/db/driver/driver.go

@@ -73,6 +73,8 @@ type BasePersistence[V any, R any] interface {
 	Commit() error
 	// Discard discards the changes since BeginUpdate
 	Discard() error
+	// Stats returns driver specific statistics of the datastore
+	Stats() any
 }
 
 // UnversionedPersistence models a key-value storage place

platform/view/services/db/driver/notifier/persistence.go

@@ -108,6 +108,10 @@ func (db *UnversionedPersistenceNotifier[P]) Close() error { return db.Persisten
 
 func (db *UnversionedPersistenceNotifier[P]) BeginUpdate() error { return db.Persistence.BeginUpdate() }
 
+func (db *UnversionedPersistenceNotifier[P]) Stats() any {
+	return db.Persistence.Stats()
+}
+
 func (db *UnversionedPersistenceNotifier[P]) Subscribe(callback driver.TriggerCallback) error {
 	return db.Notifier.Subscribe(callback)
 }
@@ -210,6 +214,10 @@ func (db *VersionedPersistenceNotifier[P]) GetStateSetIterator(ns driver2.Namesp
 	return db.Persistence.GetStateSetIterator(ns, keys...)
 }
 
+func (db *VersionedPersistenceNotifier[P]) Stats() any {
+	return db.Persistence.Stats()
+}
+
 func (db *VersionedPersistenceNotifier[P]) Close() error { return db.Persistence.Close() }
 
 func (db *VersionedPersistenceNotifier[P]) BeginUpdate() error { return db.Persistence.BeginUpdate() }

platform/view/services/db/driver/sql/common/base.go

@@ -10,6 +10,7 @@ import (
 	"database/sql"
 	"fmt"
 	"strings"
+	"time"
 
 	errors2 "github.com/hyperledger-labs/fabric-smart-client/pkg/utils/errors"
 	driver2 "github.com/hyperledger-labs/fabric-smart-client/platform/common/driver"
@@ -54,7 +55,7 @@ type BasePersistence[V any, R any] struct {
 
 func NewBasePersistence[V any, R any](writeDB *sql.DB, readDB *sql.DB, table string, readScanner readScanner[R], valueScanner ValueScanner[V], errorWrapper driver.SQLErrorWrapper, ci Interpreter, newTransaction func() (*sql.Tx, error)) *BasePersistence[V, R] {
 	return &BasePersistence[V, R]{
-		BaseDB:       common.NewBaseDB[*sql.Tx](func() (*sql.Tx, error) { return newTransaction() }),
+		BaseDB:       common.NewBaseDB(func() (*sql.Tx, error) { return newTransaction() }),
 		readDB:       readDB,
 		writeDB:      writeDB,
 		table:        table,
@@ -301,6 +302,13 @@ func (db *BasePersistence[V, R]) TableName() string {
 	return db.table
 }
 
+func (db *BasePersistence[V, R]) Stats() any {
+	if db.readDB != nil {
+		return db.readDB.Stats()
+	}
+	return nil
+}
+
 type readIterator[V any] struct {
 	txs     *sql.Rows
 	scanner readScanner[V]
@@ -329,6 +337,8 @@ type Opts struct {
 	SkipCreateTable bool
 	SkipPragmas     bool
 	MaxOpenConns    int
+	MaxIdleConns    int
+	MaxIdleTime     time.Duration
 }
 
 func generateParamSet(offset, count int) string {

platform/view/services/db/driver/sql/common/utils.go

@@ -12,6 +12,7 @@ import (
 	"regexp"
 	"runtime/debug"
 	"strings"
+	"time"
 
 	errors2 "github.com/hyperledger-labs/fabric-smart-client/pkg/utils/errors"
 	"github.com/hyperledger-labs/fabric-smart-client/platform/view/services/db/driver"
@@ -91,6 +92,13 @@ func GetOpts(config driver.Config, optsKey string) (*Opts, error) {
 	if opts.DataSource == "" {
 		return nil, notSetError(optsKey + ".dataSource")
 	}
+	if !config.IsSet(optsKey + ".maxIdleConns") {
+		opts.MaxIdleConns = 2 // go default
+	}
+	if !config.IsSet(optsKey + ".maxIdleTime") {
+		opts.MaxIdleTime = time.Minute
+	}
+
 	return &opts, nil
 }
 

platform/view/services/db/driver/sql/common/versioned.go

@@ -100,7 +100,7 @@ func NewVersionedPersistence(base basePersistence[driver.VersionedValue, driver.
 }
 
 func NewVersioned(readDB *sql.DB, writeDB *sql.DB, table string, errorWrapper driver.SQLErrorWrapper, ci Interpreter) *VersionedPersistence {
-	base := NewBasePersistence[driver.VersionedValue, driver.VersionedRead](writeDB, readDB, table, &versionedReadScanner{}, &versionedValueScanner{}, errorWrapper, ci, writeDB.Begin)
+	base := NewBasePersistence(writeDB, readDB, table, &versionedReadScanner{}, &versionedValueScanner{}, errorWrapper, ci, writeDB.Begin)
 	return NewVersionedPersistence(base, base.table, base.errorWrapper, base.readDB, base.writeDB)
 }
 
@@ -176,6 +176,10 @@ func (db *VersionedPersistence) NewWriteTransaction() (driver.WriteTransaction,
 	return NewWriteTransaction(txn, db), nil
 }
 
+func (db *VersionedPersistence) Stats() any {
+	return db.basePersistence.Stats()
+}
+
 type versionedPersistence interface {
 	SetStateWithTx(tx *sql.Tx, ns driver2.Namespace, pkey string, value driver.VersionedValue) error
 	DeleteStateWithTx(tx *sql.Tx, ns driver2.Namespace, key driver2.PKey) error

platform/view/services/db/driver/sql/postgres/bench_test.go

@@ -8,8 +8,10 @@ package postgres
 
 import (
 	"testing"
+	"time"
 
 	"github.com/hyperledger-labs/fabric-smart-client/platform/view/services/db/dbtest"
+	"github.com/hyperledger-labs/fabric-smart-client/platform/view/services/db/driver/unversioned"
 )
 
 func BenchmarkReadExistingPostgres(b *testing.B) {
@@ -18,7 +20,7 @@ func BenchmarkReadExistingPostgres(b *testing.B) {
 		b.Fatal(err)
 	}
 	defer terminate()
-	db, err := initPersistence(NewVersioned, pgConnStr, "benchmark", 50)
+	db, err := initPersistence(NewVersioned, pgConnStr, "benchmark", 50, 2, time.Minute)
 	if err != nil {
 		b.Fatal(err)
 	}
@@ -33,7 +35,7 @@ func BenchmarkReadNonExistingPostgres(b *testing.B) {
 		b.Fatal(err)
 	}
 	defer terminate()
-	db, err := initPersistence(NewVersioned, pgConnStr, "benchmark", 50)
+	db, err := initPersistence(NewVersioned, pgConnStr, "benchmark", 50, 2, time.Minute)
 	if err != nil {
 		b.Fatal(err)
 	}
@@ -48,7 +50,7 @@ func BenchmarkWriteOnePostgres(b *testing.B) {
 		b.Fatal(err)
 	}
 	defer terminate()
-	db, err := initPersistence(NewVersioned, pgConnStr, "benchmark", 50)
+	db, err := initPersistence(NewVersioned, pgConnStr, "benchmark", 50, 2, time.Minute)
 	if err != nil {
 		b.Fatal(err)
 	}
@@ -63,11 +65,27 @@ func BenchmarkWriteManyPostgres(b *testing.B) {
 		b.Fatal(err)
 	}
 	defer terminate()
-	db, err := initPersistence(NewVersioned, pgConnStr, "benchmark", 50)
+	db, err := initPersistence(NewVersioned, pgConnStr, "benchmark", 50, 2, time.Minute)
 	if err != nil {
 		b.Fatal(err)
 	}
 	defer db.Close()
 
 	dbtest.WriteMany(b, db)
 }
+
+func BenchmarkWriteManyPostgresWithIdle(b *testing.B) {
+	terminate, pgConnStr, err := StartPostgres(b, false)
+	if err != nil {
+		b.Fatal(err)
+	}
+	defer terminate()
+	p, err := initPersistence(NewVersioned, pgConnStr, "benchmark", 50, 50, time.Minute)
+	if err != nil {
+		b.Fatal(err)
+	}
+	db := &unversioned.Transactional{TransactionalVersioned: p}
+	defer db.Close()
+
+	dbtest.WriteParallel(b, db)
+}