featurebase/rbf/db.go
CLoZengineer daceee2ab6
merge: featurebase merge updates for 2022-10-28 (#2188)
* use t.Fatal(f) to abort tests, not panic

* make perf_able run at all, make it debug a bit better

switch perf-able to using same node type we use for other spot instances,
because otherwise it never finds any available capacity.

we switch the perf-able script to use the standard get_value function
instead of direct jq calls.

we try to grab server logs if the restore fails in the hopes of finding
out why the restore very occasionally fails.

* Fix some issues with running IDK tests in docker. (#2248)

*Stop running TestKafkaSourceIntegration with t.Parallel()

This test can't be run in parallel as it's currently written. Doing so
allows for interleaving of messages to the same kafka topic between
tests.

I didn't attempt to modify the test so it could be run in parallel. That
could be done, but left for someone more ambitious.

* Remove idk/testenv/certs which got accidentally committed.

also update .gitignore to include those.

* changes to add bool support in idk (#2240)

* initial changes to add bool support in idk

* modifying some default parameters for testing, will revert them later

* adding support for bool in making fragments function

* boolean values implementation without supporting empty or null values at this point

* Implement bool support in batch using a map (and a slice for nulls) (#2247)

* Implement bool support in batch using a map (and a slice for nulls)

* Keep the PackBools default for now

But set it explicity in the ingest tests which rely on it.

* Modify batch to construct bool update like mutex

The code in API.ImportRoaringShard has a switch statement which causes
bool fields to be handled like mutex fields. This means, that the
viewUpdate.Clear value should only contain data in the first "row" of
the fragment, which it will treat as records to clear for *all* rows.
This makes more sense for mutex fields; for bool fields, there's only
one other row to clear. But since the code is currently handling them
the same, we need to construct viewUpdate.Clear such that it conforms to
that pattern.

This commit also adds a test which covers this logic.

* Remove commented code; revert config for testing

This commit also removes the DELETE_SENTINEL case for non-packed bools,
since that isn't supported anyway.

* Revert default setting

* remove inconsistent type scope

* correcting the logic of string converstion to bool

* resolving an error in a test

* adding tests to cover code related to bool support in batch.go file and interface.go files

* modifying interfaces test

* added one more test case

Co-authored-by: Travis Turner <travis@pilosa.com>
Co-authored-by: Travis Turner <travis@molecula.com>

* resolving bool null field ingestion error (#2254)

* resolving bool null field ingestion error

* testing issues

* adding null support for bools

* updating the null bool field ingestion

* trying to resolve issue when ingesting null value for bool type

* adding a clearing support for bool type

* resolving issues with bool null value ingestion

* updating the jwt go package version and removing changes made in docker compose file

* reverting jwt go version

* removing v4 of jwt

* adding a comment in test file to see if sonar cloud accepts this file

* don't obtain stack traces on rbf.Tx creation

We thought stack traces were mildly expensive. We were very wrong.
Due to a complicated issue in the Go runtime, simultaneous requests
for stack traces end up contending on a lock even when they're not
actually contending on any resources. I've filed a ticket in the
Go issue tracker for this:

	https://github.com/golang/go/issues/56400

In the mean time: Under some workloads, we were seeing 85% of all
CPU time go into the stack backtraces, of which 81% went into the
contention on those locks. But even if you take away the contention,
that leaves us with 4/19 of all CPU time in our code going into
building those stack backtraces. That's a lot of overhead for a
feature we virtually never use.

We might consider adding a backtrace functionality here, possibly
using `runtime.Callers` which is much lower overhead, and allows us
to generate a backtrace on demand (no argument values available,
but then, we never read those because they're unformatted hex
values), but I don't think it's actually very informative to know
what the stack traces were of the Tx; they don't necessarily reflect
the current state of any ongoing use of the Tx, so we can't necessarily
correlate them to goroutine stack dumps, and so on.

* fb-1729 Enriched Table Metadata (#2255)

enriched metadata for tables

added support for the concept of a table and field owners in metadata; mechanism to derive owner from http request metadata; metadata for table description

* tightened up is/is not null filter expressions (FB-1741) (#2260)

Covers tightening up handling filter expressions that contain is/is not null ops. These filters may have to be translated into PQL calls to be passed to the executor and even though sql3 language supports nullability for any data type, currently only BSI fields are nullable at the storage engine level (there is a ticket to add support for non-BSI field here FB-1689: IS SQL Argument returns incorrect error) so when these fields are used in filter conditions we need to handle BSI and non-BSI fields differently.

* added a test to cover the keyword replace as being synonymous with insert (#2261)

* update molecula references to featurebase (#2262)

Co-authored-by: Seebs <seebs@molecula.com>
Co-authored-by: Travis Turner <travis@pilosa.com>
Co-authored-by: Pranitha-malae <56414132+Pranitha-malae@users.noreply.github.com>
Co-authored-by: Travis Turner <travis@molecula.com>
Co-authored-by: pokeeffe-molecula <85502298+pokeeffe-molecula@users.noreply.github.com>
Co-authored-by: Stephanie Yang <stephanie@pilosa.com>
2022-10-28 13:08:23 -04:00

907 lines
25 KiB
Go

// Copyright 2022 Molecula Corp. (DBA FeatureBase).
// SPDX-License-Identifier: Apache-2.0
package rbf
import (
"fmt"
"io"
"os"
"path/filepath"
"sort"
"sync"
"syscall"
"unsafe"
"github.com/pkg/errors"
"github.com/benbjohnson/immutable"
"github.com/featurebasedb/featurebase/v3/logger"
rbfcfg "github.com/featurebasedb/featurebase/v3/rbf/cfg"
"github.com/featurebasedb/featurebase/v3/syswrap"
)
var (
ErrClosed = errors.New("rbf: database closed")
)
// shared cursor pool across all DB instances.
// Cursors are returned on Cursor.Close().
var cursorSyncPool = &sync.Pool{
New: func() interface{} {
return &Cursor{}
},
}
// txWaiter is a representation of "i need to wait for txs to complete".
// it is created with a function, and will run that function, with the db
// lock held, at some point after every Tx that was open when it was created
// has closed. WARNING: A txWaiter may hold db.rwmu.
type txWaiter struct {
ready chan struct{}
waitingOn map[*Tx]struct{}
callback func()
}
// DB options like MaxSize, FsyncEnabled, DoAllocZero
// can be set before calling DB.Open().
type DB struct {
cfg rbfcfg.Config
data []byte // database mmap
file *os.File // database file descriptor
rootRecords *immutable.SortedMap // cached root records
pageMap *PageMap // pgno-to-WALID mapping
txs map[*Tx]struct{} // active transactions
opened bool // true if open
logger logger.Logger // for diagnostics from async things
wal []byte // wal mmap
walFile *os.File // wal file descriptor
walPageN int // wal page count
baseWALID int64 // WAL ID of first page
mu sync.RWMutex // general mutex
rwmu sync.Mutex // mutex for restricting single writer
haltCond *sync.Cond // condition for resuming txs after checkpoint
txWaiters []*txWaiter // things waiting for Txs to close
isDead error // this database died in an unrecoverable way, error out opens
// Path represents the path to the database file.
Path string
freelistCursor Cursor // cursor to reuse for freelist operations
}
// NewDB returns a new instance of DB.
// If cfg is nil we will use the rbfcfg.DefaultConfig().
func NewDB(path string, cfg *rbfcfg.Config) *DB {
if cfg == nil {
cfg = rbfcfg.NewDefaultConfig()
}
db := &DB{
cfg: *cfg,
txs: make(map[*Tx]struct{}),
pageMap: NewPageMap(),
Path: path,
logger: cfg.Logger,
}
if db.logger == nil {
// default to writing to stdout if not told otherwise
db.logger = logger.NewStandardLogger(os.Stderr)
}
db.haltCond = sync.NewCond(&db.mu)
return db
}
// DataPath returns the path to the data file for the DB.
func (db *DB) DataPath() string {
return filepath.Join(db.Path, "data")
}
// WALPath returns the path to the WAL file.
func (db *DB) WALPath() string {
return filepath.Join(db.Path, "wal")
}
func CreateDirIfNotExist(path string) {
dir := filepath.Dir(path)
if _, err := os.Stat(dir); os.IsNotExist(err) {
err = os.MkdirAll(dir, 0755)
if err != nil {
panic(err)
}
}
}
// TxN returns the number of active transactions.
func (db *DB) TxN() int {
db.mu.RLock()
defer db.mu.RUnlock()
return len(db.txs)
}
// Open opens a database with the file specified in Path.
// Creates a new file if one does not already exist.
func (db *DB) Open() (err error) {
db.mu.Lock()
defer db.mu.Unlock()
if err := os.MkdirAll(db.Path, 0755); err != nil {
return err
} else if db.file, err = os.OpenFile(db.DataPath(), os.O_WRONLY|os.O_CREATE, 0600); err != nil {
return fmt.Errorf("open file: %w", err)
}
// Open read-only database mmap.
if f, err := os.OpenFile(db.DataPath(), os.O_RDONLY, 0600); err != nil {
return fmt.Errorf("open mmap file: %w", err)
} else if db.data, err = syswrap.Mmap(int(f.Fd()), 0, int(db.cfg.MaxSize), syscall.PROT_READ, syscall.MAP_SHARED); err != nil {
f.Close()
return fmt.Errorf("open mmap file: %w", err)
} else if err := f.Close(); err != nil {
return fmt.Errorf("cannot close mmap file: %w", err)
}
// Initialize file if it is too small.
if fi, err := db.file.Stat(); err != nil {
return fmt.Errorf("stat: %w", err)
} else if fi.Size() < PageSize {
if err := db.init(); err != nil {
return fmt.Errorf("init: %w", err)
}
}
// TODO(BBJ): Obtain advisory lock on file.
db.opened = true
// Open write-ahead log & checkpoint to the end since no transactions are open.
if err := db.openWAL(); err != nil {
return fmt.Errorf("wal open: %w", err)
} else {
// checkpoint wants to hold the rwmu lock.
db.rwmu.Lock()
if err := db.checkpoint(); err != nil {
return fmt.Errorf("startup checkpoint: %w", err)
}
}
return nil
}
func (db *DB) openWAL() (err error) {
// Open WAL file writer.
if db.walFile, err = os.OpenFile(db.WALPath(), os.O_WRONLY|os.O_CREATE, 0600); err != nil {
return fmt.Errorf("open wal file: %w", err)
}
// Open read-only mmap.
if f, err := os.OpenFile(db.WALPath(), os.O_RDONLY, 0600); err != nil {
return fmt.Errorf("open wal mmap file: %w", err)
} else if db.wal, err = syswrap.Mmap(int(f.Fd()), 0, int(db.cfg.MaxWALSize), syscall.PROT_READ, syscall.MAP_SHARED); err != nil {
f.Close()
return fmt.Errorf("map wal mmap file: %w", err)
} else if err := f.Close(); err != nil {
return fmt.Errorf("cannot close wal mmap file: %w", err)
}
// Determine the number of whole pages in the WAL.
var pageN int
var fileSize int64
if fi, err := db.walFile.Stat(); err != nil {
return fmt.Errorf("wal stat: %w", err)
} else {
fileSize = fi.Size()
pageN = int(fileSize / PageSize)
}
// Read backwards through the WAL to find the last valid meta page.
for ; pageN > 0; pageN-- {
if page, err := db.readWALPageAt(pageN - 1); err != nil {
return err
} else if IsMetaPage(page) {
// We now face a challenge. Probably this is a meta page.
// But consider a sequence of pages written which gets
// interrupted right before the meta page is written.
// If the last page is a bitmap page, it could LOOK LIKE a meta
// page. So we have to check the page before it. If that page
// is a bitmap header, then actually this is a bitmap page, right?
// If that page doesn't exist, of course, we're fine, except
// for the philosophical question of why we wrote a meta page
// when no pages had changed.
if pageN > 1 {
if page, err = db.readWALPageAt(pageN - 2); err != nil {
return err
}
if IsBitmapHeader(page) {
// But wait!
// What if this *is* a meta page, and the page before it is
// actually a *bitmap page* that looks like a bitmap header? And
// so on.
//
// Rather than try to resolve this, in this insanely unlikely
// situation, we read from the beginning which allows us to
// always know what we're seeing, because every bitmap page
// comes *after* a bitmap header page, and thus, we know when
// we might be seeing one.
pageN, err = db.methodicalWALPageN(pageN)
if err != nil {
return err
}
}
}
break
}
}
if fileSize != int64(pageN*PageSize) {
if err := db.walFile.Truncate(int64(pageN) * PageSize); err != nil {
return fmt.Errorf("wal truncate: %w", err)
}
}
if _, err := db.walFile.Seek(int64(pageN)*PageSize, io.SeekStart); err != nil {
return fmt.Errorf("wal seek: %w", err)
}
db.walPageN = pageN
db.baseWALID = readMetaWALID(db.data)
return nil
}
// methodicalWALPageN tries to determine the last meta page in a very reliable
// but slow way. This handles the theoretical but hard to imagine creating
// edge case where we have a bitmap page which happens to look like a meta
// page, and the write got interrupted before the meta page got written.
func (db *DB) methodicalWALPageN(pageN int) (lastMeta int, err error) {
for i := 0; i < pageN; i++ {
var page []byte
if page, err = db.readWALPageAt(i); err != nil {
return -1, err
}
switch {
case IsMetaPage(page):
lastMeta = i + 1
case IsBitmapHeader(page):
// skip the bitmap page, which we can't usefully evaluate
i++
}
}
return lastMeta, nil
}
// Checkpoint performs a manual checkpoint. This is not necessary except for tests.
func (db *DB) Checkpoint() error {
db.mu.Lock()
defer db.mu.Unlock()
db.rwmu.Lock()
return db.checkpoint()
}
// checkpoint moves all WAL pages to the main DB file. Must be called
// while holding both db.mu and db.rwmu. Should release db.rwmu, but not
// db.mu.
func (db *DB) checkpoint() (err error) {
// if we don't spin off a possible async waiter, we should release the
// write lock, if we do, that will release it.
releaseLock := true
defer func() {
if releaseLock {
db.rwmu.Unlock()
}
}()
if !db.opened {
return nil
}
// Check if there are any WAL pages, if not do nothing as
// checkpointing and calling fsync can be very expensive even if
// there are no writes.
if db.walPageN == 0 {
return nil
}
// wake up things waiting on haltCond when we're done, even if we fail.
// Otherwise, we deadlock with them all stuck waiting on that forever.
defer func() {
if err != nil && db.isDead == nil {
db.isDead = err
}
db.haltCond.Broadcast()
}()
// Copy the pages from the WAL back to the database outside of the lock.
var pageN uint32
if err := func() error {
db.mu.Unlock() // This is intentionally reversed so run w/o lock
defer db.mu.Lock()
var page []byte
// We might have either a *PageMap or just the file. If we have the file,
// building the PageMap is fairly expensive because it's fancy and immutable.
// If we have the PageMap *or* some other map, that's two different things
// to iterate. If we have the PageMap, building a map from it is relatively
// cheap, so we'll do it that way.
pages := make(map[uint32]int)
if db.pageMap.size == 0 {
// you'd think we're done, but actually this PROBABLY means that
// this is initial startup, and we haven't read the file yet. We scan
// the file for pages, because it turns out most of them probably
// got overwritten.
for i := 0; i < db.walPageN; i++ {
page, err = db.readWALPageAt(i)
if err != nil {
return fmt.Errorf("reading WAL page %d: %w", i, err)
}
// Determine page number. Meta pages are always on zero & bitmap
// headers specify the page number of the next page in the WAL.
// All other pages have their page number in the page data.
var pgno uint32
if IsBitmapHeader(page) {
pgno = readPageNo(page)
if i+1 < db.walPageN {
if _, err = db.readWALPageAt(i + 1); err != nil {
return err
}
} else {
return fmt.Errorf("last page of WAL file (%d) is bitmap header", i)
}
i++ // bitmaps in WAL are two pages
} else if !IsMetaPage(page) {
pgno = readPageNo(page)
}
// record where in the file we have this page
pages[pgno] = i
}
} else {
itr := db.pageMap.Iterator()
itr.First()
for k, v, ok := itr.Next(); ok; k, v, ok = itr.Next() {
pages[k] = int(v - db.baseWALID - 1)
}
}
for pgno, walID := range pages {
page, err = db.readWALPageAt(walID)
if err != nil {
return fmt.Errorf("reading page %d [page number %d]: %v", walID, pgno, err)
}
// Determine new database size from the page size in meta page.
if pgno == 0 {
pageN = readMetaPageN(page)
}
// Write data to the data file.
if err = db.writeDBPage(pgno, page); err != nil {
return fmt.Errorf("writing page %d: %v", pgno, err)
}
}
// Ensure database file is synced and then truncate the WAL file.
if err = db.fsync(db.file); err != nil {
return fmt.Errorf("db file sync: %w", err)
}
return nil
}(); err != nil {
return err
}
// now we've updated the file. There are existing transactions that are still
// using the WAL, though. So we wait for them to terminate before we unlock
// the rwmu and update the metadata about the WAL.
releaseLock = false
db.walPageN = 0
db.pageMap = NewPageMap()
db.afterCurrentTx(func() {
defer db.rwmu.Unlock()
db.baseWALID = readMetaWALID(db.data)
db.mu.Unlock()
defer db.mu.Lock()
if err = db.walFile.Truncate(0); err != nil {
db.logger.Errorf("truncate wal file: %w", err)
} else if err = db.fsync(db.walFile); err != nil {
db.logger.Errorf("wal file sync: %w", err)
} else if _, err = db.walFile.Seek(0, io.SeekStart); err != nil {
db.logger.Errorf("seek wal file: %w", err)
}
// Truncate data file if it has shrunk.
if fi, err := db.file.Stat(); err != nil {
db.logger.Errorf("stat db file: %w", err)
} else if sz := int64(pageN) * PageSize; sz > 0 && fi.Size() > sz {
if err := db.file.Truncate(sz); err != nil {
db.logger.Errorf("truncate db file: %w", err)
}
}
})
return nil
}
// Close closes the database.
func (db *DB) Close() (err error) {
// mark db as closed, spawn a thing to wait for existing tx to drain, then
// release the lock so they CAN drain. We do this before getting the
// write lock, so if something else is waiting on rwmu.Lock, and will be
// competing with us, we can ensure that it'll exit out quickly.
db.mu.Lock()
db.opened = false
// wait for transactions to complete
ch := make(chan struct{})
db.afterCurrentTx(func() {
close(ch)
})
db.mu.Unlock()
<-ch
// Wait for writer lock.
db.rwmu.Lock()
defer db.rwmu.Unlock()
// and main DB lock.
db.mu.Lock()
defer db.mu.Unlock()
// Close mmap handle.
if db.data != nil {
if e := syswrap.Munmap(db.data); e != nil && err == nil {
err = e
}
db.data = nil
}
// Close writer handler.
if db.file != nil {
err = db.fsync(db.file)
if err != nil {
return
}
if e := db.file.Close(); e != nil && err == nil {
err = e
}
db.file = nil
}
// Close WAL mmap handle.
if db.wal != nil {
if e := syswrap.Munmap(db.wal); e != nil && err == nil {
err = e
}
db.wal = nil
}
// Close wal writer handler.
if db.walFile != nil {
if e := db.walFile.Close(); e != nil && err == nil {
err = e
}
db.walFile = nil
}
return err
}
// HasData with requireOneHotBit=false returns
// hasAnyRecords true if any record has been stored,
// even if the value for that bitmap record turned out to have
// no bits hot (be all zeroes).
//
// In this case, we are taking the attempted storage
// of any named bitmap into the database as evidence
// that the db is in use, and we return hasAnyRecords true.
//
// Conversely, if requireOneHotBit is true, then a
// database consisting of only a named bitmap with
// an all zeroes (no bits hot)
// will return hasAnyRecords false. We must find at
// least a single hot bit inside the db
// in order to return hasAnyRecords true.
//
// HasData is used by backend migration.
//
// If there is a disk error we return (false, error), so always
// check the error before deciding if hasAnyRecords is valid.
//
// We will internally create and rollback a read-only
// transaction to answer this query.
func (db *DB) HasData(requireOneHotBit bool) (hasAnyRecords bool, err error) {
// Read a list of all bitmaps in Tx.
tx, err := db.Begin(false)
if err != nil {
return false, err
}
defer tx.Rollback()
records, err := tx.RootRecords()
if err != nil {
return false, err
}
// Loop over each bitmap and attempt to move to the first cell.
// If we can move to a cell then we have at least one record.
for itr := records.Iterator(); !itr.Done(); {
name, _ := itr.Next()
// Fetch cursor for bitmap.
cur, err := tx.Cursor(name.(string))
if err != nil {
return false, err
}
defer cur.Close()
if !requireOneHotBit {
return true, nil
}
// INVAR: requireOneHotBit true
// Check if we can move to the first cell.
if err := cur.First(); err == io.EOF {
continue // no data in bitmap
} else if err != nil {
return false, err
}
return true, nil
}
return false, nil
}
// Size returns the size of the database & WAL, in bytes.
func (db *DB) Size() (int64, error) {
db.mu.RLock()
defer db.mu.RUnlock()
fi, err := os.Stat(db.Path)
if err != nil {
return 0, err
}
return db.walSize() + fi.Size(), nil
}
// WALSize returns the size of the WAL, in bytes.
func (db *DB) WALSize() int64 {
db.mu.RLock()
defer db.mu.RUnlock()
return db.walSize()
}
func (db *DB) walSize() int64 {
return int64(db.walPageN) * PageSize
}
// init initializes a new database file.
func (db *DB) init() error {
if err := db.initMetaPage(); err != nil {
return fmt.Errorf("meta: %w", err)
} else if err := db.initRootRecordPage(); err != nil {
return fmt.Errorf("root record page: %w", err)
} else if err := db.initFreelistPage(); err != nil {
return fmt.Errorf("freelist page: %w", err)
}
return nil
}
// initMetaPage initializes the meta page.
func (db *DB) initMetaPage() error {
page := allocPage()
writeMetaMagic(page)
writeMetaPageN(page, 3)
writeMetaRootRecordPageNo(page, 1)
writeMetaFreelistPageNo(page, 2)
_, err := db.file.WriteAt(page, 0*PageSize)
return err
}
// initRootRecordPage initializes the initial root record page.
func (db *DB) initRootRecordPage() error {
page := allocPage()
writePageNo(page, 1)
writeFlags(page, PageTypeRootRecord)
_, err := db.file.WriteAt(page, 1*PageSize)
return err
}
// initFreelistPage initializes the initial freelist btree page.
func (db *DB) initFreelistPage() error {
page := allocPage()
writePageNo(page, 2)
writeFlags(page, PageTypeLeaf)
_, err := db.file.WriteAt(page, 2*PageSize)
return err
}
// Begin starts a new transaction.
func (db *DB) Begin(writable bool) (_ *Tx, err error) {
// Ensure only one writable transaction at a time.
if writable {
db.rwmu.Lock()
}
// This local function is called at exit points that occur before we can
// call Rollback() which would normally release these locks.
cleanup := func() {
if writable {
db.rwmu.Unlock()
}
}
db.mu.Lock()
defer db.mu.Unlock()
if !db.opened {
cleanup()
return nil, ErrClosed
}
if db.isDead != nil {
err := db.isDead
cleanup()
return nil, err
}
// Wait for WAL size to be below threshold, if we're going to write.
// Reads don't care.
if writable {
for int64(db.walPageN)*PageSize > db.cfg.MaxWALCheckpointSize {
if db.isDead != nil {
err := db.isDead
cleanup()
return nil, err
}
// This implicitly releases db.mu.Lock and comes back with it
// held again.
db.haltCond.Wait()
}
}
tx := &Tx{
db: db,
rootRecords: db.rootRecords,
pageMap: db.pageMap,
walPageN: db.walPageN,
writable: writable,
DeleteEmptyContainer: true,
}
defer func() {
if err != nil {
tx.rollback(true)
}
}()
if writable {
tx.dirtyPages = make(map[uint32][]byte)
tx.dirtyBitmapPages = make(map[uint32][]byte)
}
// Copy meta page into transaction's buffer.
// This page is only written at the end of a dirty transaction.
page, err := db.readMetaPage()
if err != nil {
return nil, err
}
copy(tx.meta[:], page)
// Attach starting WAL ID to transaction.
tx.walID = readMetaWALID(tx.meta[:])
// Track transaction with the DB.
db.txs[tx] = struct{}{}
// If no root records are cached, build the cache the first time.
// Normally the cache is updated by successful write transactions but
// this avoids recomputing the cache if there are no write txs for a while.
if db.rootRecords == nil {
if db.rootRecords, err = tx.RootRecords(); err != nil {
return nil, err
}
}
return tx, nil
}
// afterCurrentTx runs the provided callback, with the db lock
// held, after all current Tx terminate. It should be called with the db
// lock held.
func (db *DB) afterCurrentTx(callback func()) {
if len(db.txs) == 0 {
callback()
return
}
txw := &txWaiter{}
txw.ready = make(chan struct{})
txw.callback = callback
txw.waitingOn = make(map[*Tx]struct{}, len(db.txs))
for k := range db.txs {
txw.waitingOn[k] = struct{}{}
}
db.txWaiters = append(db.txWaiters, txw)
go func() {
<-txw.ready
db.mu.Lock()
defer db.mu.Unlock()
txw.callback()
}()
}
// removeTx removes an active transaction from the database. it obtains
// the db lock, and currently drops it, but will later possibly be leaving
// it retained by an asynchronous op that wants to happen before we start
// running new tx.
func (db *DB) removeTx(tx *Tx) error {
// We might want to trigger a checkpoint. Only for writable
// transactions, and only when either there's nothing else open or we
// really need to.
checkpoint := false
if tx.writable {
walSize := db.walSize()
if walSize > db.cfg.MinWALCheckpointSize {
// Might be a good time for a checkpoint. We'll do a checkpoint
// if we're the only transaction, or if we have to.
if len(db.txs) == 1 || walSize > db.cfg.MaxWALCheckpointSize {
checkpoint = true
}
}
// During checkpointing, we'll be preventing writes, but allowing reads.
if !checkpoint {
tx.db.rwmu.Unlock()
}
}
// remove ourselves from the list of transactions the db is keeping.
delete(tx.db.txs, tx)
for i := 0; i < len(tx.db.txWaiters); i++ {
txw := tx.db.txWaiters[i]
// in practice this probably never matters, but theoretically the
// goroutine that's waiting on the condition variable may
// not have performed its first test on len(txw.waitingOn) yet.
delete(txw.waitingOn, tx)
// let it know we're done. we've still got db.mu.lock, so it won't
// happen just yet, but it'll be able to continue.
if len(txw.waitingOn) == 0 {
// remove us from the db's list
copy(db.txWaiters[i:], db.txWaiters[i+1:])
db.txWaiters = db.txWaiters[:len(db.txWaiters)-1]
close(txw.ready)
// decrement i so we don't skip an entry we just copied in to [i]
i--
}
}
// Disassociate from db.
tx.db = nil
if checkpoint {
// We need to run a checkpoint. This can be semi-asynchronous.
// It needs to wait until every existing transaction has finished,
// because every existing transaction could want to look up pages
// which are in the database before our operations, but which should
// now be in the WAL. We want them to use the WAL instead.
db.afterCurrentTx(func() {
// We still hold db.rwmu here. checkpoint unlocks it when it's
// ready.
if err := db.checkpoint(); err != nil {
db.logger.Errorf("async checkpoint: %v", err)
}
})
}
return nil
}
// Check performs an integrity check.
func (db *DB) Check() error {
tx, err := db.Begin(false)
if err != nil {
return err
}
defer tx.Rollback()
return tx.Check()
}
// writeDBPage writes a page to the data file.
func (db *DB) writeDBPage(pgno uint32, page []byte) error {
_, err := db.file.WriteAt(page, int64(pgno)*PageSize)
return err
}
func (db *DB) readDBPage(pgno uint32) ([]byte, error) {
offset := int64(pgno) * PageSize
// FB-1381
// Verify page number requested is within the current size of database.
bound := offset + PageSize
if sz := int64(len(db.data)); bound >= sz {
return nil, fmt.Errorf("rbf: page read out of bounds, pgno=%d upper-bound=%d file-size=%d", pgno, bound, sz)
}
return db.data[offset:bound], nil
}
// readWALPageByID reads a WAL page by WAL ID.
func (db *DB) readWALPageByID(id int64) ([]byte, error) {
return db.readWALPageAt(int(id - db.baseWALID - 1))
}
// readWALPageAt reads the i-th page in the WAL file.
func (db *DB) readWALPageAt(i int) ([]byte, error) {
offset := int64(i) * PageSize
return db.wal[offset : offset+PageSize], nil
}
func (db *DB) readMetaPage() ([]byte, error) {
if walID, ok := db.pageMap.Get(uint32(0)); ok {
return db.readWALPageByID(walID)
}
return db.readDBPage(0)
}
// getCursor returns a cursor which has not been zeroed. The only thing
// a caller should need to do is set c.stack's top correctly (it should be
// 0, and the [0] elem should be the root page to start on).
//
// TODO: Should this do anything about c.buffered?
func (db *DB) getCursor(tx *Tx) *Cursor {
c := cursorSyncPool.Get().(*Cursor)
c.tx = tx
return c
}
func (db *DB) DebugInfo() *DebugInfo {
info := &DebugInfo{Path: db.Path}
for tx := range db.txs {
info.Txs = append(info.Txs, tx.DebugInfo())
}
sort.Slice(info.Txs, func(i, j int) bool { return info.Txs[i].Ptr < info.Txs[j].Ptr })
return info
}
type DebugInfo struct {
Path string `json:"path"`
Txs []*TxDebugInfo `json:"txs"`
}
// when we want a cursor to access a free list, we are always doing this in
// a context specific to a write transaction, of which any DB can only have
// one at a time, and the operations modifying the free list don't recurse,
// because that would corrupt the list (see tx.freelistCleanup for the hairy
// details), which means that there is only ever one cursor being used for the
// free list, but also we use that cursor very often, and if we have to allocate
// it or zero it we end up with a lot of excess allocations and zeroing.
func (db *DB) getFreelistCursor(tx *Tx) *Cursor {
c := &db.freelistCursor
c.tx = tx
c.stack.elems[0] = stackElem{pgno: readMetaFreelistPageNo(tx.meta[:])}
c.stack.top = 0
c.buffered = false
return c
}
// Shared pool for in-memory database pages.
// These are used before being flushed to disk.
var pagePool = &sync.Pool{}
func allocPage() []byte {
existing := pagePool.Get()
if existing == nil {
return make([]byte, PageSize)
}
// zero the existing page before returning it
page := existing.(*[PageSize]byte)[:]
for i := range page {
page[i] = 0
}
return page
}
func freePage(page []byte) {
data := (*[PageSize]byte)(unsafe.Pointer(&page[0]))
pagePool.Put(data)
}