From 912ad59b9e4b23019d179904309aa1a3ea55712f Mon Sep 17 00:00:00 2001 From: Patrick Pacher Date: Tue, 2 Aug 2022 10:07:13 +0200 Subject: [PATCH 1/4] Add connection pooling for sqlite database access --- go.mod | 1 + go.sum | 2 + netquery/database.go | 110 ++++++++++++++++++++++++++++--------------- 3 files changed, 76 insertions(+), 37 deletions(-) diff --git a/go.mod b/go.mod index 8ee22764..741cee16 100644 --- a/go.mod +++ b/go.mod @@ -45,6 +45,7 @@ require ( github.com/gorilla/websocket v1.5.0 // indirect github.com/hashicorp/errwrap v1.1.0 // indirect github.com/inconshreveable/mousetrap v1.0.0 // indirect + github.com/jackc/puddle/v2 v2.0.0-beta.1 // indirect github.com/josharian/native v1.0.0 // indirect github.com/kardianos/osext v0.0.0-20190222173326-2bc1f35cddc0 // indirect github.com/mattn/go-isatty v0.0.14 // indirect diff --git a/go.sum b/go.sum index 587e7aeb..fcfb45d3 100644 --- a/go.sum +++ b/go.sum @@ -569,6 +569,8 @@ github.com/imdario/mergo v0.3.5/go.mod h1:2EnlNZ0deacrJVfApfmtdGgDfMuh/nq6Ok1EcJ github.com/imdario/mergo v0.3.12/go.mod h1:jmQim1M+e3UYxmgPu/WyfjB3N3VflVyUjjjwH0dnCYA= github.com/inconshreveable/mousetrap v1.0.0 h1:Z8tu5sraLXCXIcARxBp/8cbvlwVa7Z1NHg9XEKhtSvM= github.com/inconshreveable/mousetrap v1.0.0/go.mod h1:PxqpIevigyE2G7u3NXJIT2ANytuPF1OarO4DADm73n8= +github.com/jackc/puddle/v2 v2.0.0-beta.1 h1:Y4Ao+kFWANtDhWUkdw1JcbH+x84/aq6WUfhVQ1wdib8= +github.com/jackc/puddle/v2 v2.0.0-beta.1/go.mod h1:itE7ZJY8xnoo0JqJEpSMprN0f+NQkMCuEV/N9j8h0oc= github.com/jbenet/go-context v0.0.0-20150711004518-d14ea06fba99/go.mod h1:1lJo3i6rXxKeerYnT8Nvf0QmHCRC1n8sfWVwXF2Frvo= github.com/jessevdk/go-flags v1.5.0/go.mod h1:Fw0T6WPc1dYxT4mKEZRfG5kJhaTDP9pj1c2EWnYs/m4= github.com/jmespath/go-jmespath v0.0.0-20160202185014-0b12d6b521d8/go.mod h1:Nht3zPeWKUH0NzdCt2Blrr5ys8VGpn0CEB0cQHVjt7k= diff --git a/netquery/database.go b/netquery/database.go index ed1dcd8b..b53d8c14 100644 --- a/netquery/database.go +++ b/netquery/database.go @@ -12,6 +12,7 @@ import ( "zombiezen.com/go/sqlite" "zombiezen.com/go/sqlite/sqlitex" + "github.com/jackc/puddle/v2" "github.com/safing/portbase/log" "github.com/safing/portmaster/netquery/orm" "github.com/safing/portmaster/network" @@ -20,7 +21,7 @@ import ( ) // InMemory is the "file path" to open a new in-memory database. -const InMemory = ":memory:" +const InMemory = "file:inmemdb" // Available connection types as their string representation. const ( @@ -47,8 +48,10 @@ type ( Database struct { Schema *orm.TableSchema - l sync.Mutex - conn *sqlite.Conn + pool *puddle.Pool[*sqlite.Conn] + + l sync.Mutex + // conn *sqlite.Conn } // Conn is a network connection that is stored in a SQLite database and accepted @@ -109,26 +112,38 @@ type ( // SIGBUS/SIGSEGV when used with out a dedicated mutex in *Database. // func New(path string) (*Database, error) { - c, err := sqlite.OpenConn( - path, - sqlite.OpenCreate, - sqlite.OpenReadWrite, - sqlite.OpenFullMutex, - sqlite.OpenWAL, - sqlite.OpenSharedCache, - ) - if err != nil { - return nil, fmt.Errorf("failed to open sqlite at %s: %w", path, err) + constructor := func(ctx context.Context) (*sqlite.Conn, error) { + c, err := sqlite.OpenConn( + path, + sqlite.OpenCreate, + sqlite.OpenReadWrite, + sqlite.OpenNoMutex, + sqlite.OpenWAL, + sqlite.OpenSharedCache, + sqlite.OpenMemory, + ) + if err != nil { + return nil, fmt.Errorf("failed to open sqlite at %s: %w", path, err) + } + + return c, nil } + destructor := func(resource *sqlite.Conn) { + resource.Close() + } + + pool := puddle.NewPool(constructor, destructor, 10) + schema, err := orm.GenerateTableSchema("connections", Conn{}) if err != nil { return nil, err } return &Database{ + pool: pool, Schema: schema, - conn: c, + // conn: c, }, nil } @@ -156,14 +171,26 @@ func NewInMemory() (*Database, error) { // become/use a full migration system -- use zombiezen.com/go/sqlite/sqlitemigration. func (db *Database) ApplyMigrations() error { // get the create-table SQL statement from the inferred schema - sql := db.Schema.CreateStatement(false) + sql := db.Schema.CreateStatement(true) - // execute the SQL - if err := sqlitex.ExecuteTransient(db.conn, sql, nil); err != nil { - return fmt.Errorf("failed to create schema: %w", err) + return db.withConn(context.Background(), func(conn *sqlite.Conn) error { + // execute the SQL + if err := sqlitex.ExecuteTransient(conn, sql, nil); err != nil { + return fmt.Errorf("failed to create schema: %w", err) + } + + return nil + }) +} + +func (db *Database) withConn(ctx context.Context, fn func(conn *sqlite.Conn) error) error { + res, err := db.pool.Acquire(ctx) + if err != nil { + return err } + defer res.Release() - return nil + return fn(res.Value()) } // Execute executes a custom SQL query against the SQLite database used by db. @@ -173,7 +200,9 @@ func (db *Database) Execute(ctx context.Context, sql string, args ...orm.QueryOp db.l.Lock() defer db.l.Unlock() - return orm.RunQuery(ctx, db.conn, sql, args...) + return db.withConn(ctx, func(conn *sqlite.Conn) error { + return orm.RunQuery(ctx, conn, sql, args...) + }) } // CountRows returns the number of rows stored in the database. @@ -239,17 +268,20 @@ func (db *Database) dumpTo(ctx context.Context, w io.Writer) error { //nolint:un defer db.l.Unlock() var conns []Conn - if err := sqlitex.ExecuteTransient(db.conn, "SELECT * FROM connections", &sqlitex.ExecOptions{ - ResultFunc: func(stmt *sqlite.Stmt) error { - var c Conn - if err := orm.DecodeStmt(ctx, db.Schema, stmt, &c, orm.DefaultDecodeConfig); err != nil { - return err - } + err := db.withConn(ctx, func(conn *sqlite.Conn) error { + return sqlitex.ExecuteTransient(conn, "SELECT * FROM connections", &sqlitex.ExecOptions{ + ResultFunc: func(stmt *sqlite.Stmt) error { + var c Conn + if err := orm.DecodeStmt(ctx, db.Schema, stmt, &c, orm.DefaultDecodeConfig); err != nil { + return err + } - conns = append(conns, c) - return nil - }, - }); err != nil { + conns = append(conns, c) + return nil + }, + }) + }) + if err != nil { return err } @@ -294,12 +326,14 @@ func (db *Database) Save(ctx context.Context, conn Conn) error { strings.Join(updateSets, ", "), ) - if err := sqlitex.ExecuteTransient(db.conn, sql, &sqlitex.ExecOptions{ - Named: values, - ResultFunc: func(stmt *sqlite.Stmt) error { - log.Errorf("netquery: got result statement with %d columns", stmt.ColumnCount()) - return nil - }, + if err := db.withConn(ctx, func(conn *sqlite.Conn) error { + return sqlitex.ExecuteTransient(conn, sql, &sqlitex.ExecOptions{ + Named: values, + ResultFunc: func(stmt *sqlite.Stmt) error { + log.Errorf("netquery: got result statement with %d columns", stmt.ColumnCount()) + return nil + }, + }) }); err != nil { log.Errorf("netquery: failed to execute:\n\t%q\n\treturned error was: %s\n\tparameters: %+v", sql, err, values) return err @@ -311,5 +345,7 @@ func (db *Database) Save(ctx context.Context, conn Conn) error { // Close closes the underlying database connection. db should and cannot be // used after Close() has returned. func (db *Database) Close() error { - return db.conn.Close() + return nil + + //return db.conn.Close() } From 84b66e994905c7e36daf27d13806b8cd00e1922c Mon Sep 17 00:00:00 2001 From: Patrick Pacher Date: Tue, 2 Aug 2022 10:13:12 +0200 Subject: [PATCH 2/4] Remove locking from SQlite access --- netquery/database.go | 16 ++-------------- 1 file changed, 2 insertions(+), 14 deletions(-) diff --git a/netquery/database.go b/netquery/database.go index b53d8c14..f9ce1bce 100644 --- a/netquery/database.go +++ b/netquery/database.go @@ -6,7 +6,6 @@ import ( "fmt" "io" "strings" - "sync" "time" "zombiezen.com/go/sqlite" @@ -21,7 +20,7 @@ import ( ) // InMemory is the "file path" to open a new in-memory database. -const InMemory = "file:inmemdb" +const InMemory = "file:inmem.db" // Available connection types as their string representation. const ( @@ -49,9 +48,6 @@ type ( Schema *orm.TableSchema pool *puddle.Pool[*sqlite.Conn] - - l sync.Mutex - // conn *sqlite.Conn } // Conn is a network connection that is stored in a SQLite database and accepted @@ -121,6 +117,7 @@ func New(path string) (*Database, error) { sqlite.OpenWAL, sqlite.OpenSharedCache, sqlite.OpenMemory, + sqlite.OpenURI, ) if err != nil { return nil, fmt.Errorf("failed to open sqlite at %s: %w", path, err) @@ -197,9 +194,6 @@ func (db *Database) withConn(ctx context.Context, fn func(conn *sqlite.Conn) err // It uses orm.RunQuery() under the hood so please refer to the orm package for // more information about available options. func (db *Database) Execute(ctx context.Context, sql string, args ...orm.QueryOption) error { - db.l.Lock() - defer db.l.Unlock() - return db.withConn(ctx, func(conn *sqlite.Conn) error { return orm.RunQuery(ctx, conn, sql, args...) }) @@ -264,9 +258,6 @@ func (db *Database) Cleanup(ctx context.Context, threshold time.Time) (int, erro // as JSON to w. // Any error aborts dumping rows and is returned. func (db *Database) dumpTo(ctx context.Context, w io.Writer) error { //nolint:unused - db.l.Lock() - defer db.l.Unlock() - var conns []Conn err := db.withConn(ctx, func(conn *sqlite.Conn) error { return sqlitex.ExecuteTransient(conn, "SELECT * FROM connections", &sqlitex.ExecOptions{ @@ -310,9 +301,6 @@ func (db *Database) Save(ctx context.Context, conn Conn) error { updateSets = append(updateSets, fmt.Sprintf("%s = :%s", key, key)) } - db.l.Lock() - defer db.l.Unlock() - // TODO(ppacher): make sure this one can be cached to speed up inserting // and save some CPU cycles for the user sql := fmt.Sprintf( From 6a86596736a2133caf2229111e787ec016e1db86 Mon Sep 17 00:00:00 2001 From: Patrick Pacher Date: Tue, 2 Aug 2022 10:46:16 +0200 Subject: [PATCH 3/4] Use a dedicated write connection for persisting in sqlite and switch pool to read-only --- netquery/database.go | 54 +++++++++++++++++++++++++++++--------------- 1 file changed, 36 insertions(+), 18 deletions(-) diff --git a/netquery/database.go b/netquery/database.go index f9ce1bce..a6333a77 100644 --- a/netquery/database.go +++ b/netquery/database.go @@ -6,6 +6,7 @@ import ( "fmt" "io" "strings" + "sync" "time" "zombiezen.com/go/sqlite" @@ -48,6 +49,9 @@ type ( Schema *orm.TableSchema pool *puddle.Pool[*sqlite.Conn] + + l sync.Mutex + writeConn *sqlite.Conn } // Conn is a network connection that is stored in a SQLite database and accepted @@ -111,16 +115,14 @@ func New(path string) (*Database, error) { constructor := func(ctx context.Context) (*sqlite.Conn, error) { c, err := sqlite.OpenConn( path, - sqlite.OpenCreate, - sqlite.OpenReadWrite, + sqlite.OpenReadOnly, sqlite.OpenNoMutex, - sqlite.OpenWAL, sqlite.OpenSharedCache, sqlite.OpenMemory, sqlite.OpenURI, ) if err != nil { - return nil, fmt.Errorf("failed to open sqlite at %s: %w", path, err) + return nil, fmt.Errorf("failed to open read-only sqlite connection at %s: %w", path, err) } return c, nil @@ -137,10 +139,24 @@ func New(path string) (*Database, error) { return nil, err } + writeConn, err := sqlite.OpenConn( + path, + sqlite.OpenCreate, + sqlite.OpenReadWrite, + sqlite.OpenNoMutex, + sqlite.OpenWAL, + sqlite.OpenSharedCache, + sqlite.OpenMemory, + sqlite.OpenURI, + ) + if err != nil { + return nil, fmt.Errorf("failed to open sqlite at %s: %w", path, err) + } + return &Database{ - pool: pool, - Schema: schema, - // conn: c, + pool: pool, + Schema: schema, + writeConn: writeConn, }, nil } @@ -283,6 +299,9 @@ func (db *Database) dumpTo(ctx context.Context, w io.Writer) error { //nolint:un // Save inserts the connection conn into the SQLite database. If conn // already exists the table row is updated instead. +// +// Save uses the database write connection instead of relying on the +// connection pool. func (db *Database) Save(ctx context.Context, conn Conn) error { connMap, err := orm.ToParamMap(ctx, conn, "", orm.DefaultEncodeConfig) if err != nil { @@ -301,6 +320,9 @@ func (db *Database) Save(ctx context.Context, conn Conn) error { updateSets = append(updateSets, fmt.Sprintf("%s = :%s", key, key)) } + db.l.Lock() + defer db.l.Unlock() + // TODO(ppacher): make sure this one can be cached to speed up inserting // and save some CPU cycles for the user sql := fmt.Sprintf( @@ -314,14 +336,12 @@ func (db *Database) Save(ctx context.Context, conn Conn) error { strings.Join(updateSets, ", "), ) - if err := db.withConn(ctx, func(conn *sqlite.Conn) error { - return sqlitex.ExecuteTransient(conn, sql, &sqlitex.ExecOptions{ - Named: values, - ResultFunc: func(stmt *sqlite.Stmt) error { - log.Errorf("netquery: got result statement with %d columns", stmt.ColumnCount()) - return nil - }, - }) + if err := sqlitex.Execute(db.writeConn, sql, &sqlitex.ExecOptions{ + Named: values, + ResultFunc: func(stmt *sqlite.Stmt) error { + log.Errorf("netquery: got result statement with %d columns", stmt.ColumnCount()) + return nil + }, }); err != nil { log.Errorf("netquery: failed to execute:\n\t%q\n\treturned error was: %s\n\tparameters: %+v", sql, err, values) return err @@ -333,7 +353,5 @@ func (db *Database) Save(ctx context.Context, conn Conn) error { // Close closes the underlying database connection. db should and cannot be // used after Close() has returned. func (db *Database) Close() error { - return nil - - //return db.conn.Close() + return db.writeConn.Close() } From 01f7ab24b3be94cecd8d2e20f10db0864f338122 Mon Sep 17 00:00:00 2001 From: Patrick Pacher Date: Tue, 2 Aug 2022 14:04:12 +0200 Subject: [PATCH 4/4] Fix write/readonly connection usage and linter error --- netquery/database.go | 58 ++++++++++++++++++++++++++------------------ 1 file changed, 35 insertions(+), 23 deletions(-) diff --git a/netquery/database.go b/netquery/database.go index a6333a77..b857f310 100644 --- a/netquery/database.go +++ b/netquery/database.go @@ -41,14 +41,10 @@ type ( // It's use is tailored for persistence and querying of network.Connection. // Access to the underlying SQLite database is synchronized. // - // TODO(ppacher): somehow I'm receiving SIGBUS or SIGSEGV when no doing - // synchronization in *Database. Check what exactly sqlite.OpenFullMutex, etc.. - // are actually supposed to do. - // Database struct { Schema *orm.TableSchema - pool *puddle.Pool[*sqlite.Conn] + readConnPool *puddle.Pool[*sqlite.Conn] l sync.Mutex writeConn *sqlite.Conn @@ -105,11 +101,12 @@ type ( } ) -// New opens a new database at path. The database is opened with Full-Mutex, Write-Ahead-Log (WAL) -// and Shared-Cache enabled. +// New opens a new in-memory database named path. // -// TODO(ppacher): check which sqlite "open flags" provide the best performance and don't cause -// SIGBUS/SIGSEGV when used with out a dedicated mutex in *Database. +// The returned Database used connection pooling for read-only connections +// (see Execute). To perform database writes use either Save() or ExecuteWrite(). +// Note that write connections are serialized by the Database object before being +// handed over to SQLite. // func New(path string) (*Database, error) { constructor := func(ctx context.Context) (*sqlite.Conn, error) { @@ -129,7 +126,9 @@ func New(path string) (*Database, error) { } destructor := func(resource *sqlite.Conn) { - resource.Close() + if err := resource.Close(); err != nil { + log.Errorf("failed to close pooled SQlite database connection: %s", err) + } } pool := puddle.NewPool(constructor, destructor, 10) @@ -154,9 +153,9 @@ func New(path string) (*Database, error) { } return &Database{ - pool: pool, - Schema: schema, - writeConn: writeConn, + readConnPool: pool, + Schema: schema, + writeConn: writeConn, }, nil } @@ -186,18 +185,19 @@ func (db *Database) ApplyMigrations() error { // get the create-table SQL statement from the inferred schema sql := db.Schema.CreateStatement(true) - return db.withConn(context.Background(), func(conn *sqlite.Conn) error { - // execute the SQL - if err := sqlitex.ExecuteTransient(conn, sql, nil); err != nil { - return fmt.Errorf("failed to create schema: %w", err) - } + db.l.Lock() + defer db.l.Unlock() - return nil - }) + // execute the SQL + if err := sqlitex.ExecuteTransient(db.writeConn, sql, nil); err != nil { + return fmt.Errorf("failed to create schema: %w", err) + } + + return nil } func (db *Database) withConn(ctx context.Context, fn func(conn *sqlite.Conn) error) error { - res, err := db.pool.Acquire(ctx) + res, err := db.readConnPool.Acquire(ctx) if err != nil { return err } @@ -206,7 +206,19 @@ func (db *Database) withConn(ctx context.Context, fn func(conn *sqlite.Conn) err return fn(res.Value()) } -// Execute executes a custom SQL query against the SQLite database used by db. +// ExecuteWrite executes a custom SQL query using a writable connection against the SQLite +// database used by db. +// It uses orm.RunQuery() under the hood so please refer to the orm package for +// more information about available options. +func (db *Database) ExecuteWrite(ctx context.Context, sql string, args ...orm.QueryOption) error { + db.l.Lock() + defer db.l.Unlock() + + return orm.RunQuery(ctx, db.writeConn, sql, args...) +} + +// Execute executes a custom SQL query using a read-only connection against the SQLite +// database used by db. // It uses orm.RunQuery() under the hood so please refer to the orm package for // more information about available options. func (db *Database) Execute(ctx context.Context, sql string, args ...orm.QueryOption) error { @@ -262,7 +274,7 @@ func (db *Database) Cleanup(ctx context.Context, threshold time.Time) (int, erro return 0, fmt.Errorf("unexpected number of rows, expected 1 got %d", len(result)) } - err := db.Execute(ctx, sql, args) + err := db.ExecuteWrite(ctx, sql, args) if err != nil { return 0, err }