From 9f6efc8b53a1644255db1a8bc93bd800b6b728e9 Mon Sep 17 00:00:00 2001 From: Chris Lu Date: Wed, 2 Sep 2026 23:45:45 -0700 Subject: [PATCH] filer: a listing over a hard link no longer deadlocks a bounded SQL pool (#11118) * filer: give the SQL stores' key-value reads their own connections A listing holds the connection its rows are on for the whole iteration, and FilerStoreWrapper calls maybeReadHardLink -> KvGet from inside that iteration, so a hard-linked entry needs a second connection while the first is still busy. Out of one bounded pool that is a deadlock: the listings fill the pool and then wait for a connection none of them will release, and the wrapper's context.WithoutCancel leaves the waiters without a deadline, so the filer stays wedged rather than erroring. The sqlite store shows it at its sharpest -- it allows a single connection, so one listing over one hard-linked entry never returns. On postgres with connection_max_open = 50, 60 concurrent listings over hard-linked entries made no progress at all. Key-value reads now run on their own pool, carved out of connection_max_open rather than added to it, so the operator's cap still bounds what the store opens against the database. An unbounded pool keeps a single pool: nothing can wait there. sqlite's single connection becomes two, one per pool, and its writes get a busy timeout so a write that meets the reader waits instead of failing. Claude-Session: https://claude.ai/code/session_018DWwctzD4T2DmnczPRM47t * sqlite: keep both pools on one database, whatever the dbFile spells A dbFile that already carries URI options got a second "?" appended, which the driver reads as part of the preceding option value, and a bare :memory: is private to each connection, so the key-value pool would open its own empty database and every key-value operation would fail on a missing filemeta. Claude-Session: https://claude.ai/code/session_018DWwctzD4T2DmnczPRM47t * sqlite: assert the busy timeout on the in-memory DSN too Claude-Session: https://claude.ai/code/session_018DWwctzD4T2DmnczPRM47t --- weed/filer/abstract_sql/abstract_sql_store.go | 48 ++++++++++ .../abstract_sql/abstract_sql_store_kv.go | 27 +++--- .../abstract_sql/abstract_sql_store_test.go | 20 ++++ weed/filer/mysql/mysql_store.go | 10 +- weed/filer/mysql2/mysql2_store.go | 17 ++-- weed/filer/postgres/postgres_store.go | 7 +- weed/filer/postgres2/postgres2_store.go | 7 +- weed/filer/sqlite/sqlite_store.go | 35 +++++-- weed/filer/sqlite/sqlite_store_test.go | 92 +++++++++++++++++++ 9 files changed, 228 insertions(+), 35 deletions(-) create mode 100644 weed/filer/sqlite/sqlite_store_test.go diff --git a/weed/filer/abstract_sql/abstract_sql_store.go b/weed/filer/abstract_sql/abstract_sql_store.go index 9d01bdbba..b35a24e8e 100644 --- a/weed/filer/abstract_sql/abstract_sql_store.go +++ b/weed/filer/abstract_sql/abstract_sql_store.go @@ -6,6 +6,7 @@ import ( "fmt" "strings" "sync" + "time" "github.com/seaweedfs/seaweedfs/weed/filer" "github.com/seaweedfs/seaweedfs/weed/glog" @@ -29,6 +30,7 @@ type SqlGenerator interface { type AbstractSqlStore struct { SqlGenerator DB *sql.DB + KvDB *sql.DB SupportBucketTable bool dbs map[string]bool dbsLock sync.Mutex @@ -97,6 +99,49 @@ func (store *AbstractSqlStore) RollbackTransaction(ctx context.Context) error { return nil } +// A listing holds its connection for the whole row iteration while its callback +// reads a hard link through KvGet, so both cannot come out of one bounded pool: +// the listings fill it and then wait for a connection none of them will release. +// Give the key-value reads their own slice of connection_max_open. A cap of 1 is +// the exception -- it has to become 2, or a single listing cannot finish. +func splitPoolForKv(maxOpen int) (mainOpen, kvOpen int) { + if maxOpen <= 0 { + return maxOpen, 0 + } + kvOpen = min(max(maxOpen/4, 1), maxKvPoolSize) + return max(maxOpen-kvOpen, 1), kvOpen +} + +const maxKvPoolSize = 8 + +// UseConnectionPools sizes the store's pool and, when it is bounded, opens the +// separate pool the key-value reads run on. +func (store *AbstractSqlStore) UseConnectionPools(db *sql.DB, openKv func() (*sql.DB, error), maxIdle, maxOpen, maxLifetimeSeconds int) error { + + lifetime := time.Duration(maxLifetimeSeconds) * time.Second + mainOpen, kvOpen := splitPoolForKv(maxOpen) + + db.SetMaxIdleConns(maxIdle) + db.SetMaxOpenConns(mainOpen) + db.SetConnMaxLifetime(lifetime) + store.DB = db + + if kvOpen == 0 { + return nil + } + + kvDB, err := openKv() + if err != nil { + return err + } + kvDB.SetMaxIdleConns(kvOpen) + kvDB.SetMaxOpenConns(kvOpen) + kvDB.SetConnMaxLifetime(lifetime) + store.KvDB = kvDB + + return nil +} + func (store *AbstractSqlStore) getTxOrDB(ctx context.Context, fullpath util.FullPath, isForChildren bool) (txOrDB TxOrDB, bucket string, shortPath util.FullPath, err error) { shortPath = fullpath @@ -400,6 +445,9 @@ func (store *AbstractSqlStore) ListDirectoryEntries(ctx context.Context, dirPath func (store *AbstractSqlStore) Shutdown() { store.DB.Close() + if store.KvDB != nil { + store.KvDB.Close() + } } func isValidBucket(bucket string) bool { diff --git a/weed/filer/abstract_sql/abstract_sql_store_kv.go b/weed/filer/abstract_sql/abstract_sql_store_kv.go index ca8950381..fd5ba3566 100644 --- a/weed/filer/abstract_sql/abstract_sql_store_kv.go +++ b/weed/filer/abstract_sql/abstract_sql_store_kv.go @@ -12,12 +12,21 @@ import ( "github.com/seaweedfs/seaweedfs/weed/util" ) +// Key-value reads run outside the listing pool: a listing calls KvGet from +// inside its own callback, still holding the connection its rows are on. +func (store *AbstractSqlStore) kvDB(ctx context.Context) TxOrDB { + if tx, ok := ctx.Value("tx").(*sql.Tx); ok { + return tx + } + if store.KvDB != nil { + return store.KvDB + } + return store.DB +} + func (store *AbstractSqlStore) KvPut(ctx context.Context, key []byte, value []byte) (err error) { - db, _, _, err := store.getTxOrDB(ctx, "", false) - if err != nil { - return fmt.Errorf("findDB: %w", err) - } + db := store.kvDB(ctx) dirStr, dirHash, name := GenDirAndName(key) @@ -49,10 +58,7 @@ func (store *AbstractSqlStore) KvPut(ctx context.Context, key []byte, value []by func (store *AbstractSqlStore) KvGet(ctx context.Context, key []byte) (value []byte, err error) { - db, _, _, err := store.getTxOrDB(ctx, "", false) - if err != nil { - return nil, fmt.Errorf("findDB: %w", err) - } + db := store.kvDB(ctx) dirStr, dirHash, name := GenDirAndName(key) row := db.QueryRowContext(ctx, store.GetSqlFind(DEFAULT_TABLE), dirHash, name, dirStr) @@ -72,10 +78,7 @@ func (store *AbstractSqlStore) KvGet(ctx context.Context, key []byte) (value []b func (store *AbstractSqlStore) KvDelete(ctx context.Context, key []byte) (err error) { - db, _, _, err := store.getTxOrDB(ctx, "", false) - if err != nil { - return fmt.Errorf("findDB: %w", err) - } + db := store.kvDB(ctx) dirStr, dirHash, name := GenDirAndName(key) diff --git a/weed/filer/abstract_sql/abstract_sql_store_test.go b/weed/filer/abstract_sql/abstract_sql_store_test.go index 156b84c85..540183e9e 100644 --- a/weed/filer/abstract_sql/abstract_sql_store_test.go +++ b/weed/filer/abstract_sql/abstract_sql_store_test.go @@ -51,3 +51,23 @@ func TestGetTxOrDBRealBucket(t *testing.T) { t.Errorf("shortPath = %q, want /dir/file.txt", shortPath) } } + +func TestSplitPoolForKv(t *testing.T) { + cases := []struct { + maxOpen, mainOpen, kvOpen int + }{ + {0, 0, 0}, // unbounded: nothing waits, one pool is enough + {-1, -1, 0}, // same, however the operator spelled it + {50, 42, 8}, + {16, 12, 4}, + {4, 3, 1}, + {2, 1, 1}, + {1, 1, 1}, // a listing needs two, so a cap of 1 has to become 2 + } + for _, c := range cases { + mainOpen, kvOpen := splitPoolForKv(c.maxOpen) + if mainOpen != c.mainOpen || kvOpen != c.kvOpen { + t.Errorf("splitPoolForKv(%d) = %d, %d, want %d, %d", c.maxOpen, mainOpen, kvOpen, c.mainOpen, c.kvOpen) + } + } +} diff --git a/weed/filer/mysql/mysql_store.go b/weed/filer/mysql/mysql_store.go index 5a7cf9da0..1da17fd20 100644 --- a/weed/filer/mysql/mysql_store.go +++ b/weed/filer/mysql/mysql_store.go @@ -8,7 +8,6 @@ import ( "fmt" "os" "strings" - "time" "github.com/go-sql-driver/mysql" "github.com/seaweedfs/seaweedfs/weed/filer" @@ -151,10 +150,11 @@ func (store *MysqlStore) initialize(dsn string, upsertQuery string, enableUpsert return fmt.Errorf("can not create mysql connector for %s error:%w", maskedDSN(cfg), err) } - store.DB = sql.OpenDB(connector) - store.DB.SetMaxIdleConns(maxIdle) - store.DB.SetMaxOpenConns(maxOpen) - store.DB.SetConnMaxLifetime(time.Duration(maxLifetimeSeconds) * time.Second) + if err = store.UseConnectionPools(sql.OpenDB(connector), func() (*sql.DB, error) { + return sql.OpenDB(connector), nil + }, maxIdle, maxOpen, maxLifetimeSeconds); err != nil { + return err + } if err = store.DB.Ping(); err != nil { return fmt.Errorf("connect to %s error:%v", maskedDSN(cfg), err) diff --git a/weed/filer/mysql2/mysql2_store.go b/weed/filer/mysql2/mysql2_store.go index e8b0af261..87269376e 100644 --- a/weed/filer/mysql2/mysql2_store.go +++ b/weed/filer/mysql2/mysql2_store.go @@ -5,7 +5,6 @@ import ( "database/sql" "fmt" "strings" - "time" _ "github.com/go-sql-driver/mysql" "github.com/seaweedfs/seaweedfs/weed/filer" @@ -84,19 +83,19 @@ func (store *MysqlStore2) initialize(createTable, upsertQuery string, enableUpse adaptedSqlUrl += "&interpolateParams=true" } - var dbErr error - store.DB, dbErr = sql.Open("mysql", sqlUrl) + db, dbErr := sql.Open("mysql", sqlUrl) if dbErr != nil { - if store.DB != nil { - store.DB.Close() + if db != nil { + db.Close() } - store.DB = nil return fmt.Errorf("can not connect to %s error:%w", adaptedSqlUrl, dbErr) } - store.DB.SetMaxIdleConns(maxIdle) - store.DB.SetMaxOpenConns(maxOpen) - store.DB.SetConnMaxLifetime(time.Duration(maxLifetimeSeconds) * time.Second) + if err = store.UseConnectionPools(db, func() (*sql.DB, error) { + return sql.Open("mysql", sqlUrl) + }, maxIdle, maxOpen, maxLifetimeSeconds); err != nil { + return err + } if err = store.DB.Ping(); err != nil { return fmt.Errorf("connect to %s error:%v", adaptedSqlUrl, err) diff --git a/weed/filer/postgres/postgres_store.go b/weed/filer/postgres/postgres_store.go index 62f27b327..73926071d 100644 --- a/weed/filer/postgres/postgres_store.go +++ b/weed/filer/postgres/postgres_store.go @@ -8,6 +8,7 @@ package postgres import ( + "database/sql" "strconv" "github.com/seaweedfs/seaweedfs/weed/filer" @@ -119,7 +120,11 @@ func (store *PostgresStore) initialize(upsertQuery string, enableUpsert bool, us if openErr != nil { return openErr } - store.DB = db + if err = store.UseConnectionPools(db, func() (*sql.DB, error) { + return OpenPGXDB(sqlUrl, adaptedSqlUrl, pgbouncerCompatible, maxIdle, maxOpen, maxLifetimeSeconds) + }, maxIdle, maxOpen, maxLifetimeSeconds); err != nil { + return err + } ConfigureListOrdering(store.DB, gen) diff --git a/weed/filer/postgres2/postgres2_store.go b/weed/filer/postgres2/postgres2_store.go index 2eb85764a..dfd7ce387 100644 --- a/weed/filer/postgres2/postgres2_store.go +++ b/weed/filer/postgres2/postgres2_store.go @@ -9,6 +9,7 @@ package postgres2 import ( "context" + "database/sql" "fmt" "strconv" @@ -128,7 +129,11 @@ func (store *PostgresStore2) initialize(createTable, upsertQuery string, enableU if openErr != nil { return openErr } - store.DB = db + if err = store.UseConnectionPools(db, func() (*sql.DB, error) { + return postgres.OpenPGXDB(sqlUrl, adaptedSqlUrl, pgbouncerCompatible, maxIdle, maxOpen, maxLifetimeSeconds) + }, maxIdle, maxOpen, maxLifetimeSeconds); err != nil { + return err + } if err = store.CreateTable(context.Background(), abstract_sql.DEFAULT_TABLE); err != nil { return fmt.Errorf("init table %s: %v", abstract_sql.DEFAULT_TABLE, err) diff --git a/weed/filer/sqlite/sqlite_store.go b/weed/filer/sqlite/sqlite_store.go index 0e9fb7396..31d696ca7 100644 --- a/weed/filer/sqlite/sqlite_store.go +++ b/weed/filer/sqlite/sqlite_store.go @@ -8,6 +8,8 @@ import ( "context" "database/sql" "fmt" + "strings" + "time" "github.com/seaweedfs/seaweedfs/weed/filer" "github.com/seaweedfs/seaweedfs/weed/filer/abstract_sql" @@ -49,6 +51,21 @@ func (store *SqliteStore) Initialize(configuration util.Configuration, prefix st ) } +// sqliteDSN keeps the store's two pools on one database and lets a writer that +// meets the other pool's reader wait for it instead of failing outright. A bare +// :memory: is private to each connection, so it has to be named and shared. +func sqliteDSN(dbFile string) string { + dsn := dbFile + if dsn == ":memory:" { + dsn = fmt.Sprintf("file:seaweedfs%d?mode=memory&cache=shared", time.Now().UnixNano()) + } + separator := "?" + if strings.Contains(dsn, "?") { + separator = "&" + } + return dsn + separator + "_pragma=busy_timeout(10000)" +} + func (store *SqliteStore) initialize(dbFile, createTable, upsertQuery string) (err error) { store.SupportBucketTable = true @@ -58,21 +75,25 @@ func (store *SqliteStore) initialize(dbFile, createTable, upsertQuery string) (e UpsertQueryTemplate: upsertQuery, } - var dbErr error - store.DB, dbErr = sql.Open("sqlite", dbFile) + dsn := sqliteDSN(dbFile) + + db, dbErr := sql.Open("sqlite", dsn) if dbErr != nil { - if store.DB != nil { - store.DB.Close() - store.DB = nil + if db != nil { + db.Close() } return fmt.Errorf("can not connect to %s error:%v", dbFile, dbErr) } - if err = store.DB.Ping(); err != nil { + if err = db.Ping(); err != nil { return fmt.Errorf("connect to %s error:%v", dbFile, err) } - store.DB.SetMaxOpenConns(1) + if err = store.UseConnectionPools(db, func() (*sql.DB, error) { + return sql.Open("sqlite", dsn) + }, 1, 1, 0); err != nil { + return err + } if err = store.CreateTable(context.Background(), abstract_sql.DEFAULT_TABLE); err != nil { return fmt.Errorf("init table %s: %v", abstract_sql.DEFAULT_TABLE, err) diff --git a/weed/filer/sqlite/sqlite_store_test.go b/weed/filer/sqlite/sqlite_store_test.go new file mode 100644 index 000000000..2ada1e5bc --- /dev/null +++ b/weed/filer/sqlite/sqlite_store_test.go @@ -0,0 +1,92 @@ +//go:build (linux || darwin || windows) && sqlite + +package sqlite + +import ( + "context" + "path/filepath" + "strings" + "testing" + "time" + + "github.com/seaweedfs/seaweedfs/weed/filer" + "github.com/seaweedfs/seaweedfs/weed/util" +) + +const ( + createTableSql = `CREATE TABLE IF NOT EXISTS "%s" (dirhash BIGINT, name VARCHAR(1000), directory TEXT, meta BLOB, PRIMARY KEY (dirhash, name)) WITHOUT ROWID;` + upsertQuerySql = `INSERT INTO "%s"(dirhash,name,directory,meta)VALUES(?,?,?,?) ON CONFLICT(dirhash,name) DO UPDATE SET directory=excluded.directory, meta=excluded.meta;` +) + +// A listing whose callback reads a hard link needs a second connection while it +// still holds the one its rows are on, and the store allows exactly one. +func TestListDirectoryOverHardLinkDoesNotDeadlock(t *testing.T) { + store := &SqliteStore{} + if err := store.initialize(filepath.Join(t.TempDir(), "filer.db"), createTableSql, upsertQuerySql); err != nil { + t.Fatal(err) + } + defer store.Shutdown() + + wrapper := filer.NewFilerStoreWrapper(store) + ctx := context.Background() + entry := &filer.Entry{ + FullPath: util.FullPath("/buckets/hlbucket/f0001"), + Attr: filer.Attr{Mode: 0644, Mtime: time.Now(), Crtime: time.Now()}, + HardLinkId: filer.HardLinkId("hardlink-00000001"), + } + if err := wrapper.InsertEntry(ctx, entry); err != nil { + t.Fatal(err) + } + + done := make(chan error, 1) + go func() { + _, err := wrapper.ListDirectoryEntries(ctx, util.FullPath("/buckets/hlbucket"), "", true, 10, + func(*filer.Entry) (bool, error) { return true, nil }) + done <- err + }() + select { + case err := <-done: + if err != nil { + t.Errorf("list: %v", err) + } + case <-time.After(10 * time.Second): + t.Fatalf("one listing over one hard-linked entry never returned: %+v", store.DB.Stats()) + } +} + +func TestSqliteDSN(t *testing.T) { + if got, want := sqliteDSN("/data/filer.db"), "/data/filer.db?_pragma=busy_timeout(10000)"; got != want { + t.Errorf("sqliteDSN = %q, want %q", got, want) + } + if got, want := sqliteDSN("file:/data/filer.db?cache=shared"), "file:/data/filer.db?cache=shared&_pragma=busy_timeout(10000)"; got != want { + t.Errorf("sqliteDSN with options = %q, want %q", got, want) + } + memory := sqliteDSN(":memory:") + if !strings.HasPrefix(memory, "file:seaweedfs") || !strings.Contains(memory, "mode=memory&cache=shared") || + !strings.HasSuffix(memory, "&_pragma=busy_timeout(10000)") { + t.Errorf("sqliteDSN(:memory:) = %q, want a named shared-memory URI carrying the busy timeout", memory) + } +} + +// Both pools have to open the same database: the table is created on one and +// the key-value operations run on the other. +func TestInMemoryStoreSharesOneDatabase(t *testing.T) { + store := &SqliteStore{} + if err := store.initialize(":memory:", createTableSql, upsertQuerySql); err != nil { + t.Fatal(err) + } + defer store.Shutdown() + + ctx := context.Background() + key := []byte("hardlink-00000001") + if err := store.KvPut(ctx, key, []byte("value")); err != nil { + t.Fatalf("KvPut: %v", err) + } + value, err := store.KvGet(ctx, key) + if err != nil { + t.Fatalf("KvGet: %v", err) + } + if string(value) != "value" { + t.Errorf("KvGet = %q, want value", value) + } +}