colin
/
pocketbase
1
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

updated code comments and renamed async/sync db to concurrent/nonconcurrent db

This commit is contained in:
Gani Georgiev 2022-12-16 13:07:58 +02:00
parent c25e67e13d
commit 89de29fc84
8 changed files with 77 additions and 78 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

2
CHANGELOG.md
View File

@ -6,7 +6,7 @@
- Reduced memory consumption (you can expect ~20% less allocated memory). - Reduced memory consumption (you can expect ~20% less allocated memory).
- Added support for split (async and sync) DB connections pool increasing even further the concurrent throughput. - Added support for split (concurrent and nonconcurrent) DB connections pool increasing even further the concurrent throughput without blocking reads on heavy write load.
- Improved record references delete performance. - Improved record references delete performance.

4
core/app.go
View File

@ -19,7 +19,7 @@ type App interface {
// Deprecated: // Deprecated:
// This method may get removed in the near future. // This method may get removed in the near future.
// It is recommended to access the logs db instance from app.Dao().DB() or // It is recommended to access the logs db instance from app.Dao().DB() or
// if you want more flexibility - app.Dao().AsyncDB() and app.Dao().SyncDB(). // if you want more flexibility - app.Dao().ConcurrentDB() and app.Dao().NonconcurrentDB().
// //
// DB returns the default app database instance. // DB returns the default app database instance.
DB() *dbx.DB DB() *dbx.DB
@ -34,7 +34,7 @@ type App interface {
// Deprecated: // Deprecated:
// This method may get removed in the near future. // This method may get removed in the near future.
// It is recommended to access the logs db instance from app.LogsDao().DB() or // It is recommended to access the logs db instance from app.LogsDao().DB() or
// if you want more flexibility - app.LogsDao().AsyncDB() and app.LogsDao().SyncDB(). // if you want more flexibility - app.LogsDao().ConcurrentDB() and app.LogsDao().NonconcurrentDB().
// //
// LogsDB returns the app logs database instance. // LogsDB returns the app logs database instance.
LogsDB() *dbx.DB LogsDB() *dbx.DB

60
core/base.go
View File

@ -316,19 +316,19 @@ func (app *BaseApp) Bootstrap() error {
// (eg. closing db connections). // (eg. closing db connections).
func (app *BaseApp) ResetBootstrapState() error { func (app *BaseApp) ResetBootstrapState() error {
if app.Dao() != nil { if app.Dao() != nil {
if err := app.Dao().AsyncDB().(*dbx.DB).Close(); err != nil { if err := app.Dao().ConcurrentDB().(*dbx.DB).Close(); err != nil {
return err return err
} }
if err := app.Dao().SyncDB().(*dbx.DB).Close(); err != nil { if err := app.Dao().NonconcurrentDB().(*dbx.DB).Close(); err != nil {
return err return err
} }
} }
if app.LogsDao() != nil { if app.LogsDao() != nil {
if err := app.LogsDao().AsyncDB().(*dbx.DB).Close(); err != nil { if err := app.LogsDao().ConcurrentDB().(*dbx.DB).Close(); err != nil {
return err return err
} }
if err := app.LogsDao().SyncDB().(*dbx.DB).Close(); err != nil { if err := app.LogsDao().NonconcurrentDB().(*dbx.DB).Close(); err != nil {
return err return err
} }
} }
@ -343,7 +343,7 @@ func (app *BaseApp) ResetBootstrapState() error {
// Deprecated: // Deprecated:
// This method may get removed in the near future. // This method may get removed in the near future.
// It is recommended to access the db instance from app.Dao().DB() or // It is recommended to access the db instance from app.Dao().DB() or
// if you want more flexibility - app.Dao().AsyncDB() and app.Dao().SyncDB(). // if you want more flexibility - app.Dao().ConcurrentDB() and app.Dao().NonconcurrentDB().
// //
// DB returns the default app database instance. // DB returns the default app database instance.
func (app *BaseApp) DB() *dbx.DB { func (app *BaseApp) DB() *dbx.DB {
@ -367,7 +367,7 @@ func (app *BaseApp) Dao() *daos.Dao {
// Deprecated: // Deprecated:
// This method may get removed in the near future. // This method may get removed in the near future.
// It is recommended to access the logs db instance from app.LogsDao().DB() or // It is recommended to access the logs db instance from app.LogsDao().DB() or
// if you want more flexibility - app.LogsDao().AsyncDB() and app.LogsDao().SyncDB(). // if you want more flexibility - app.LogsDao().ConcurrentDB() and app.LogsDao().NonconcurrentDB().
// //
// LogsDB returns the app logs database instance. // LogsDB returns the app logs database instance.
func (app *BaseApp) LogsDB() *dbx.DB { func (app *BaseApp) LogsDB() *dbx.DB {
@ -826,23 +826,23 @@ func (app *BaseApp) initLogsDB() error {
maxIdleConns = app.logsMaxIdleConns maxIdleConns = app.logsMaxIdleConns
} }
asyncDB, err := connectDB(filepath.Join(app.DataDir(), "logs.db")) concurrentDB, err := connectDB(filepath.Join(app.DataDir(), "logs.db"))
if err != nil { if err != nil {
return err return err
} }
asyncDB.DB().SetMaxOpenConns(maxOpenConns) concurrentDB.DB().SetMaxOpenConns(maxOpenConns)
asyncDB.DB().SetMaxIdleConns(maxIdleConns) concurrentDB.DB().SetMaxIdleConns(maxIdleConns)
asyncDB.DB().SetConnMaxIdleTime(5 * time.Minute) concurrentDB.DB().SetConnMaxIdleTime(5 * time.Minute)
syncDB, err := connectDB(filepath.Join(app.DataDir(), "logs.db")) nonconcurrentDB, err := connectDB(filepath.Join(app.DataDir(), "logs.db"))
if err != nil { if err != nil {
return err return err
} }
syncDB.DB().SetMaxOpenConns(1) nonconcurrentDB.DB().SetMaxOpenConns(1)
syncDB.DB().SetMaxIdleConns(1) nonconcurrentDB.DB().SetMaxIdleConns(1)
syncDB.DB().SetConnMaxIdleTime(5 * time.Minute) nonconcurrentDB.DB().SetConnMaxIdleTime(5 * time.Minute)
app.logsDao = daos.NewMultiDB(asyncDB, syncDB) app.logsDao = daos.NewMultiDB(concurrentDB, nonconcurrentDB)
return nil return nil
} }
@ -857,41 +857,41 @@ func (app *BaseApp) initDataDB() error {
maxIdleConns = app.dataMaxIdleConns maxIdleConns = app.dataMaxIdleConns
} }
asyncDB, err := connectDB(filepath.Join(app.DataDir(), "data.db")) concurrentDB, err := connectDB(filepath.Join(app.DataDir(), "data.db"))
if err != nil { if err != nil {
return err return err
} }
asyncDB.DB().SetMaxOpenConns(maxOpenConns) concurrentDB.DB().SetMaxOpenConns(maxOpenConns)
asyncDB.DB().SetMaxIdleConns(maxIdleConns) concurrentDB.DB().SetMaxIdleConns(maxIdleConns)
asyncDB.DB().SetConnMaxIdleTime(5 * time.Minute) concurrentDB.DB().SetConnMaxIdleTime(5 * time.Minute)
syncDB, err := connectDB(filepath.Join(app.DataDir(), "data.db")) nonconcurrentDB, err := connectDB(filepath.Join(app.DataDir(), "data.db"))
if err != nil { if err != nil {
return err return err
} }
syncDB.DB().SetMaxOpenConns(1) nonconcurrentDB.DB().SetMaxOpenConns(1)
syncDB.DB().SetMaxIdleConns(1) nonconcurrentDB.DB().SetMaxIdleConns(1)
syncDB.DB().SetConnMaxIdleTime(5 * time.Minute) nonconcurrentDB.DB().SetConnMaxIdleTime(5 * time.Minute)
if app.IsDebug() { if app.IsDebug() {
syncDB.QueryLogFunc = func(ctx context.Context, t time.Duration, sql string, rows *sql.Rows, err error) { nonconcurrentDB.QueryLogFunc = func(ctx context.Context, t time.Duration, sql string, rows *sql.Rows, err error) {
color.HiBlack("[%.2fms] %v\n", float64(t.Milliseconds()), sql) color.HiBlack("[%.2fms] %v\n", float64(t.Milliseconds()), sql)
} }
asyncDB.QueryLogFunc = syncDB.QueryLogFunc concurrentDB.QueryLogFunc = nonconcurrentDB.QueryLogFunc
syncDB.ExecLogFunc = func(ctx context.Context, t time.Duration, sql string, result sql.Result, err error) { nonconcurrentDB.ExecLogFunc = func(ctx context.Context, t time.Duration, sql string, result sql.Result, err error) {
color.HiBlack("[%.2fms] %v\n", float64(t.Milliseconds()), sql) color.HiBlack("[%.2fms] %v\n", float64(t.Milliseconds()), sql)
} }
asyncDB.ExecLogFunc = syncDB.ExecLogFunc concurrentDB.ExecLogFunc = nonconcurrentDB.ExecLogFunc
} }
app.dao = app.createDaoWithHooks(asyncDB, syncDB) app.dao = app.createDaoWithHooks(concurrentDB, nonconcurrentDB)
return nil return nil
} }
func (app *BaseApp) createDaoWithHooks(asyncDB, syncDB dbx.Builder) *daos.Dao { func (app *BaseApp) createDaoWithHooks(concurrentDB, nonconcurrentDB dbx.Builder) *daos.Dao {
dao := daos.NewMultiDB(asyncDB, syncDB) dao := daos.NewMultiDB(concurrentDB, nonconcurrentDB)
dao.BeforeCreateFunc = func(eventDao *daos.Dao, m models.Model) error { dao.BeforeCreateFunc = func(eventDao *daos.Dao, m models.Model) error {
return app.OnModelBeforeCreate().Trigger(&ModelEvent{eventDao, m}) return app.OnModelBeforeCreate().Trigger(&ModelEvent{eventDao, m})

8
core/base_test.go
View File

@ -143,16 +143,16 @@ func TestBaseAppGetters(t *testing.T) {
t.Fatalf("Expected app.Dao %v, got %v", app.Dao(), app.dao) t.Fatalf("Expected app.Dao %v, got %v", app.Dao(), app.dao)
} }
if app.dao.AsyncDB() != app.DB() { if app.dao.ConcurrentDB() != app.DB() {
t.Fatalf("Expected app.DB %v, got %v", app.DB(), app.dao.AsyncDB()) t.Fatalf("Expected app.DB %v, got %v", app.DB(), app.dao.ConcurrentDB())
} }
if app.logsDao != app.LogsDao() { if app.logsDao != app.LogsDao() {
t.Fatalf("Expected app.LogsDao %v, got %v", app.LogsDao(), app.logsDao) t.Fatalf("Expected app.LogsDao %v, got %v", app.LogsDao(), app.logsDao)
} }
if app.logsDao.AsyncDB() != app.LogsDB() { if app.logsDao.ConcurrentDB() != app.LogsDB() {
t.Fatalf("Expected app.LogsDB %v, got %v", app.LogsDB(), app.logsDao.AsyncDB()) t.Fatalf("Expected app.LogsDB %v, got %v", app.LogsDB(), app.logsDao.ConcurrentDB())
} }
if app.dataDir != app.DataDir() { if app.dataDir != app.DataDir() {

52
daos/base.go
View File

@ -25,10 +25,10 @@ func New(db dbx.Builder) *Dao {
// New creates a new Dao instance with the provided dedicated // New creates a new Dao instance with the provided dedicated
// async and sync db builders. // async and sync db builders.
func NewMultiDB(asyncDB, syncDB dbx.Builder) *Dao { func NewMultiDB(concurrentDB, nonconcurrentDB dbx.Builder) *Dao {
return &Dao{ return &Dao{
asyncDB: asyncDB, concurrentDB: concurrentDB,
syncDB: syncDB, nonconcurrentDB: nonconcurrentDB,
} }
} }
@ -36,8 +36,8 @@ func NewMultiDB(asyncDB, syncDB dbx.Builder) *Dao {
// Think of Dao as a repository and service layer in one. // Think of Dao as a repository and service layer in one.
type Dao struct { type Dao struct {
// in a transaction both refer to the same *dbx.TX instance // in a transaction both refer to the same *dbx.TX instance
asyncDB dbx.Builder concurrentDB dbx.Builder
syncDB dbx.Builder nonconcurrentDB dbx.Builder
// @todo delete after removing Block and Continue // @todo delete after removing Block and Continue
sem *semaphore.Weighted sem *semaphore.Weighted
@ -53,26 +53,28 @@ type Dao struct {
// DB returns the default dao db builder (*dbx.DB or *dbx.TX). // DB returns the default dao db builder (*dbx.DB or *dbx.TX).
// //
// Currently the default db builder is dao.asyncDB but that may change in the future. // Currently the default db builder is dao.concurrentDB but that may change in the future.
func (dao *Dao) DB() dbx.Builder { func (dao *Dao) DB() dbx.Builder {
return dao.AsyncDB() return dao.ConcurrentDB()
} }
// AsyncDB returns the dao asynchronous db builder (*dbx.DB or *dbx.TX). // ConcurrentDB returns the dao concurrent (aka. multiple open connections)
// db builder (*dbx.DB or *dbx.TX).
// //
// In a transaction the asyncDB and syncDB refer to the same *dbx.TX instance. // In a transaction the concurrentDB and nonconcurrentDB refer to the same *dbx.TX instance.
func (dao *Dao) AsyncDB() dbx.Builder { func (dao *Dao) ConcurrentDB() dbx.Builder {
return dao.asyncDB return dao.concurrentDB
} }
// SyncDB returns the dao synchronous db builder (*dbx.DB or *dbx.TX). // NonconcurrentDB returns the dao nonconcurrent (aka. single open connection)
// db builder (*dbx.DB or *dbx.TX).
// //
// In a transaction the asyncDB and syncDB refer to the same *dbx.TX instance. // In a transaction the concurrentDB and nonconcurrentDB refer to the same *dbx.TX instance.
func (dao *Dao) SyncDB() dbx.Builder { func (dao *Dao) NonconcurrentDB() dbx.Builder {
return dao.syncDB return dao.nonconcurrentDB
} }
// Deprecated: Will be removed in the next releases. Use [Dao.SyncDB()] instead. // Deprecated: Will be removed in the next releases. Use [Dao.NonconcurrentDB()] instead.
// //
// Block acquires a lock and blocks all other go routines that uses // Block acquires a lock and blocks all other go routines that uses
// the Dao instance until dao.Continue() is called, effectively making // the Dao instance until dao.Continue() is called, effectively making
@ -105,7 +107,7 @@ func (dao *Dao) Block(ctx context.Context) error {
return dao.sem.Acquire(ctx, 1) return dao.sem.Acquire(ctx, 1)
} }
// Deprecated: Will be removed in the next releases. Use [Dao.SyncDB()] instead. // Deprecated: Will be removed in the next releases. Use [Dao.NonconcurrentDB()] instead.
// //
// Continue releases the previously acquired Block() lock. // Continue releases the previously acquired Block() lock.
func (dao *Dao) Continue() { func (dao *Dao) Continue() {
@ -139,7 +141,7 @@ type afterCallGroup struct {
// //
// It is safe to nest RunInTransaction calls as long as you use the txDao. // It is safe to nest RunInTransaction calls as long as you use the txDao.
func (dao *Dao) RunInTransaction(fn func(txDao *Dao) error) error { func (dao *Dao) RunInTransaction(fn func(txDao *Dao) error) error {
switch txOrDB := dao.SyncDB().(type) { switch txOrDB := dao.NonconcurrentDB().(type) {
case *dbx.Tx: case *dbx.Tx:
// nested transactions are not supported by default // nested transactions are not supported by default
// so execute the function within the current transaction // so execute the function within the current transaction
@ -213,7 +215,7 @@ func (dao *Dao) RunInTransaction(fn func(txDao *Dao) error) error {
return txError return txError
} }
return errors.New("Failed to start transaction (unknown dao.db)") return errors.New("failed to start transaction (unknown dao.NonconcurrentDB() instance)")
} }
// Delete deletes the provided model. // Delete deletes the provided model.
@ -229,7 +231,7 @@ func (dao *Dao) Delete(m models.Model) error {
} }
} }
if err := retryDao.SyncDB().Model(m).Delete(); err != nil { if err := retryDao.NonconcurrentDB().Model(m).Delete(); err != nil {
return err return err
} }
@ -274,7 +276,7 @@ func (dao *Dao) update(m models.Model) error {
if v, ok := any(m).(models.ColumnValueMapper); ok { if v, ok := any(m).(models.ColumnValueMapper); ok {
dataMap := v.ColumnValueMap() dataMap := v.ColumnValueMap()
_, err := dao.SyncDB().Update( _, err := dao.NonconcurrentDB().Update(
m.TableName(), m.TableName(),
dataMap, dataMap,
dbx.HashExp{"id": m.GetId()}, dbx.HashExp{"id": m.GetId()},
@ -284,7 +286,7 @@ func (dao *Dao) update(m models.Model) error {
return err return err
} }
} else { } else {
if err := dao.SyncDB().Model(m).Update(); err != nil { if err := dao.NonconcurrentDB().Model(m).Update(); err != nil {
return err return err
} }
} }
@ -325,12 +327,12 @@ func (dao *Dao) create(m models.Model) error {
dataMap["id"] = m.GetId() dataMap["id"] = m.GetId()
} }
_, err := dao.SyncDB().Insert(m.TableName(), dataMap).Execute() _, err := dao.NonconcurrentDB().Insert(m.TableName(), dataMap).Execute()
if err != nil { if err != nil {
return err return err
} }
} else { } else {
if err := dao.SyncDB().Model(m).Insert(); err != nil { if err := dao.NonconcurrentDB().Model(m).Insert(); err != nil {
return err return err
} }
} }
@ -353,7 +355,7 @@ Retry:
if attempts == 2 { if attempts == 2 {
// assign new Dao without the before hooks to avoid triggering // assign new Dao without the before hooks to avoid triggering
// the already fired before callbacks multiple times // the already fired before callbacks multiple times
retryDao = NewMultiDB(dao.asyncDB, dao.syncDB) retryDao = NewMultiDB(dao.concurrentDB, dao.nonconcurrentDB)
retryDao.AfterCreateFunc = dao.AfterCreateFunc retryDao.AfterCreateFunc = dao.AfterCreateFunc
retryDao.AfterUpdateFunc = dao.AfterUpdateFunc retryDao.AfterUpdateFunc = dao.AfterUpdateFunc
retryDao.AfterDeleteFunc = dao.AfterDeleteFunc retryDao.AfterDeleteFunc = dao.AfterDeleteFunc

8
daos/base_test.go
View File

@ -24,17 +24,17 @@ func TestNewMultiDB(t *testing.T) {
testApp, _ := tests.NewTestApp() testApp, _ := tests.NewTestApp()
defer testApp.Cleanup() defer testApp.Cleanup()
dao := daos.NewMultiDB(testApp.Dao().AsyncDB(), testApp.Dao().SyncDB()) dao := daos.NewMultiDB(testApp.Dao().ConcurrentDB(), testApp.Dao().NonconcurrentDB())
if dao.DB() != testApp.Dao().AsyncDB() { if dao.DB() != testApp.Dao().ConcurrentDB() {
t.Fatal("[db-asyncdb] The 2 db instances are different") t.Fatal("[db-asyncdb] The 2 db instances are different")
} }
if dao.AsyncDB() != testApp.Dao().AsyncDB() { if dao.ConcurrentDB() != testApp.Dao().ConcurrentDB() {
t.Fatal("[asyncdb-asyncdb] The 2 db instances are different") t.Fatal("[asyncdb-asyncdb] The 2 db instances are different")
} }
if dao.SyncDB() != testApp.Dao().SyncDB() { if dao.NonconcurrentDB() != testApp.Dao().NonconcurrentDB() {
t.Fatal("[syncdb-syncdb] The 2 db instances are different") t.Fatal("[syncdb-syncdb] The 2 db instances are different")
} }
} }

8
daos/record_test.go
View File

@ -634,16 +634,16 @@ func TestDeleteRecord(t *testing.T) {
// delete existing record + cascade // delete existing record + cascade
// --- // ---
calledQueries := []string{} calledQueries := []string{}
app.Dao().SyncDB().(*dbx.DB).QueryLogFunc = func(ctx context.Context, t time.Duration, sql string, rows *sql.Rows, err error) { app.Dao().NonconcurrentDB().(*dbx.DB).QueryLogFunc = func(ctx context.Context, t time.Duration, sql string, rows *sql.Rows, err error) {
calledQueries = append(calledQueries, sql) calledQueries = append(calledQueries, sql)
} }
app.Dao().AsyncDB().(*dbx.DB).QueryLogFunc = func(ctx context.Context, t time.Duration, sql string, rows *sql.Rows, err error) { app.Dao().ConcurrentDB().(*dbx.DB).QueryLogFunc = func(ctx context.Context, t time.Duration, sql string, rows *sql.Rows, err error) {
calledQueries = append(calledQueries, sql) calledQueries = append(calledQueries, sql)
} }
app.Dao().SyncDB().(*dbx.DB).ExecLogFunc = func(ctx context.Context, t time.Duration, sql string, result sql.Result, err error) { app.Dao().NonconcurrentDB().(*dbx.DB).ExecLogFunc = func(ctx context.Context, t time.Duration, sql string, result sql.Result, err error) {
calledQueries = append(calledQueries, sql) calledQueries = append(calledQueries, sql)
} }
app.Dao().AsyncDB().(*dbx.DB).ExecLogFunc = func(ctx context.Context, t time.Duration, sql string, result sql.Result, err error) { app.Dao().ConcurrentDB().(*dbx.DB).ExecLogFunc = func(ctx context.Context, t time.Duration, sql string, result sql.Result, err error) {
calledQueries = append(calledQueries, sql) calledQueries = append(calledQueries, sql)
} }
rec3, _ := app.Dao().FindRecordById("users", "oap640cot4yru2s") rec3, _ := app.Dao().FindRecordById("users", "oap640cot4yru2s")

13
pocketbase.go
View File

@ -148,12 +148,6 @@ func (pb *PocketBase) Start() error {
// This method differs from pb.Start() by not registering the default // This method differs from pb.Start() by not registering the default
// system commands! // system commands!
func (pb *PocketBase) Execute() error { func (pb *PocketBase) Execute() error {
// Run the bootstrap process if the command exist and it is not
// the default cobra version command to prevent creating
// unnecessary the initialization system files.
//
// https://github.com/pocketbase/pocketbase/issues/404
// https://github.com/pocketbase/pocketbase/discussions/1267
if !pb.skipBootstrap() { if !pb.skipBootstrap() {
if err := pb.Bootstrap(); err != nil { if err := pb.Bootstrap(); err != nil {
return err return err
@ -218,11 +212,14 @@ func (pb *PocketBase) eagerParseFlags(config *Config) error {
return pb.RootCmd.ParseFlags(os.Args[1:]) return pb.RootCmd.ParseFlags(os.Args[1:])
} }
// eager check if the app should skip the bootstap process: // skipBootstrap eagerly checks if the app should skip the bootstap process:
// - already bootstrapped // - already bootstrapped
// - is unknown command // - is unknown command
// - is the default help command // - is the default help command
// - is the default version command // - is the default version command
//
// https://github.com/pocketbase/pocketbase/issues/404
// https://github.com/pocketbase/pocketbase/discussions/1267
func (pb *PocketBase) skipBootstrap() bool { func (pb *PocketBase) skipBootstrap() bool {
flags := []string{ flags := []string{
"-h", "-h",
@ -258,7 +255,7 @@ func (pb *PocketBase) skipBootstrap() bool {
return false return false
} }
// tries to find the base executable directory and how it was run // inspectRuntime tries to find the base executable directory and how it was run.
func inspectRuntime() (baseDir string, withGoRun bool) { func inspectRuntime() (baseDir string, withGoRun bool) {
if strings.HasPrefix(os.Args[0], os.TempDir()) { if strings.HasPrefix(os.Args[0], os.TempDir()) {
// probably ran with go run // probably ran with go run