diff --git a/internal/db/sqlite/basedb.go b/internal/db/sqlite/basedb.go index 168fb5c11..71ca3105b 100644 --- a/internal/db/sqlite/basedb.go +++ b/internal/db/sqlite/basedb.go @@ -49,7 +49,7 @@ type baseDB struct { tplInput map[string]any } -func openBase(path string, maxConns int, pragmas, schemaScripts, migrationScripts []string) (*baseDB, error) { +func openBase(path string, maxOpenConns, maxIdleConns int, pragmas, schemaScripts, migrationScripts []string) (*baseDB, error) { // Open the database with options to enable foreign keys and recursive // triggers (needed for the delete+insert triggers on row replace). pathURL := url.URL{ @@ -62,8 +62,8 @@ func openBase(path string, maxConns int, pragmas, schemaScripts, migrationScript return nil, wrap(err) } - sqlDB.SetMaxOpenConns(maxConns) - sqlDB.SetMaxIdleConns(maxConns) + sqlDB.SetMaxOpenConns(maxOpenConns) + sqlDB.SetMaxIdleConns(maxIdleConns) for _, pragma := range pragmas { if _, err := sqlDB.Exec("PRAGMA " + pragma); err != nil { diff --git a/internal/db/sqlite/db_open.go b/internal/db/sqlite/db_open.go index 0f55282cb..000ba005d 100644 --- a/internal/db/sqlite/db_open.go +++ b/internal/db/sqlite/db_open.go @@ -11,6 +11,7 @@ import ( "log/slog" "os" "path/filepath" + "runtime" "sync" "time" @@ -20,10 +21,23 @@ import ( ) const ( - maxDBConns = 6 + // maxIdleConns is set so that most db operations will usually fit + // within one of those connections and hence use their persistent page + // cache. Additional connections on top of these will allocate their own + // page cache (same as any other), but it will be deallocated on + // connection close. + maxIdleConns = 4 + minDeleteRetention = 24 * time.Hour ) +// maxOpenConns is sized for handling spikes. The primary driver is the +// Copiers folder option which may result in up to 2*NumCPU concurrent +// iterations. We reserve additional space on top of this to serve +// additional operations, some of which may be reentrant (queries within +// iterators) without deadlock. +var maxOpenConns = max(16, 4*runtime.NumCPU()) + type DB struct { *baseDB @@ -69,7 +83,7 @@ func Open(path string, opts ...Option) (*DB, error) { initTmpDir(path) mainPath := filepath.Join(path, "main.db") - mainBase, err := openBase(mainPath, maxDBConns, pragmas, schemas, migrations) + mainBase, err := openBase(mainPath, maxOpenConns, maxIdleConns, pragmas, schemas, migrations) if err != nil { return nil, err } @@ -120,7 +134,7 @@ func OpenForMigration(path string) (*DB, error) { initTmpDir(path) mainPath := filepath.Join(path, "main.db") - mainBase, err := openBase(mainPath, 1, pragmas, schemas, migrations) + mainBase, err := openBase(mainPath, 1, 1, pragmas, schemas, migrations) if err != nil { return nil, err } diff --git a/internal/db/sqlite/folderdb_open.go b/internal/db/sqlite/folderdb_open.go index b1b4ec95d..6b5da5704 100644 --- a/internal/db/sqlite/folderdb_open.go +++ b/internal/db/sqlite/folderdb_open.go @@ -38,7 +38,7 @@ func openFolderDB(folder, path string, deleteRetention time.Duration) (*folderDB "sql/migrations/folder/*", } - base, err := openBase(path, maxDBConns, pragmas, schemas, migrations) + base, err := openBase(path, maxOpenConns, maxIdleConns, pragmas, schemas, migrations) if err != nil { return nil, err } @@ -75,7 +75,7 @@ func openFolderDBForMigration(folder, path string, deleteRetention time.Duration "sql/schema/folder/*", } - base, err := openBase(path, 1, pragmas, schemas, nil) + base, err := openBase(path, 1, 1, pragmas, schemas, nil) if err != nil { return nil, err } diff --git a/lib/model/folder_sendrecv.go b/lib/model/folder_sendrecv.go index e86011982..316cb8bf7 100644 --- a/lib/model/folder_sendrecv.go +++ b/lib/model/folder_sendrecv.go @@ -15,6 +15,7 @@ import ( "io" "log/slog" "path/filepath" + "runtime" "slices" "strconv" "strings" @@ -148,6 +149,8 @@ func newSendReceiveFolder(model *model, ignores *ignore.Matcher, cfg config.Fold // copy step. TODO: Rename this config option at some point. f.Copiers = defaultCopiers } + // Cap the copiers at 2*NumCPU, so that we have a known upper bound. + f.Copiers = min(f.Copiers, 2*runtime.NumCPU()) // If the configured max amount of pending data is zero, we use the // default. If it's configured to something non-zero but less than the @@ -1438,15 +1441,15 @@ func (f *sendReceiveFolder) copyBlock(ctx context.Context, block protocol.BlockI // Returns true when the block was successfully copied. // The passed buffer must be large enough to accommodate the block. func (f *sendReceiveFolder) copyBlockFromFolder(ctx context.Context, folderID string, block protocol.BlockInfo, state copyBlocksState, ffs fs.Filesystem, buf []byte) bool { - for e, err := range itererr.Zip(f.model.sdb.AllLocalBlocksWithHash(folderID, block.Hash)) { - if err != nil { - // We just ignore this and continue pulling instead (though - // there's a good chance that will fail too, if the DB is - // unhealthy). - f.sl.DebugContext(ctx, "Failed to get block information from database", "blockHash", block.Hash, slogutil.FilePath(state.file.Name), slogutil.Error(err)) - return false - } + candidates, err := itererr.Collect(f.model.sdb.AllLocalBlocksWithHash(folderID, block.Hash)) + if err != nil { + // We just ignore this and continue pulling instead (though there's + // a good chance that will fail too, if the DB is unhealthy). + f.sl.DebugContext(ctx, "Failed to get block information from database", "blockHash", block.Hash, slogutil.FilePath(state.file.Name), slogutil.Error(err)) + return false + } + for _, e := range candidates { if !f.copyBlockFromFile(ctx, e.FileName, e.Offset, state, ffs, block, buf) { if state.failed() != nil { return false