scriptoriumadmin/featurebase
1
0
Fork 0
mirror of https://github.com/featurebasedb/featurebase.git synced 2026-09-11 07:11:02 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions

Merge branch 'master' into fb1146

Browse source
This commit is contained in:
reesporte 2022-01-24 15:25:00 -06:00
commit ec543ac094
15 changed files with 23 additions and 1053 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

8
executor_test.go
View file

@ -6915,11 +6915,9 @@ func TestTimelessClearRegression(t *testing.T) {
}
func TestMissingKeyRegression(t *testing.T) {
c := test.MustRunCluster(t, 1, []server.CommandOption{server.OptCommandServerOptions(
pilosa.OptServerStorageConfig(&storage.Config{
Backend: "roaring",
FsyncEnabled: false,
}))})
// this used to be explicitly roaring backend... I'm not sure
// whether it is a useful test in post-roaring world.
c := test.MustRunCluster(t, 1)
defer c.Close()
c.CreateField(t, "i", pilosa.IndexOptions{Keys: true, TrackExistence: true}, "f", pilosa.OptFieldKeys())

242
fragment.go
View file

@ -140,7 +140,6 @@ type fragment struct {
// File-backed storage
flags byte // user-defined flags passed to roaring
gen generation
storage *roaring.Bitmap
opN int // number of ops since snapshot (may be approximate for imports)
ops int // number of higher-level operations, as opposed to bit changes
@ -319,159 +318,15 @@ func (f *fragment) emptyStorage(file *os.File) (bool, error) {
return false, nil
}
// importStorage attempts to import data from storage -- for instance,
// reading in a roaring bitmap from media.
func (f *fragment) importStorage(data []byte, file *os.File, newGen generation, mapped bool) (bool, error) {
f.storage.PreferMapping(mapped)
if len(data) == 0 {
return f.emptyStorage(file)
}
// UnmarshalBinary will have remapped the storage to newGen if it
// succeeded, or if it fails but the error is advisory-only. So we
// optimistically set the source here, but if there's a non-advisory
// error, we'll unmap it and then set the source to nil.
f.storage.SetSource(newGen)
if err := f.storage.UnmarshalBinary(data); err != nil {
// roaring can report advisory-only errors...
cause := errors.Cause(err)
_, ok := cause.(roaring.AdvisoryError)
if !ok {
_, e2 := f.storage.RemapRoaringStorage(nil)
f.storage.SetSource(nil)
if e2 != nil {
return false, fmt.Errorf("unmarshal storage: file=%s, err=%s, clearing old mapping also failed: %v", file.Name(), err, e2)
}
return false, fmt.Errorf("unmarshal storage: file=%s, err=%s", file.Name(), err)
}
f.holder.Logger.Warnf("unmarshal storage, file=%s, err=%v", file.Name(), err)
trunc, ok := cause.(roaring.FileShouldBeTruncatedError)
if ok && !f.holder.Opts.ReadOnly {
// if the holder is ReadOnly, we silently ignore the "advisory"
// error. This may be a bad idea.
// generation code looks for a FileShouldBeTruncatedError
return false, trunc
}
}
f.ops, f.opN = f.storage.Ops()
// For now, we assume that UnmarshalBinary will have mapped at least
// one container if we told it the storage was mapped and it didn't
// error out. This might be wrong in occasional trivial cases, but
// it should be harmless.
return mapped, nil
}
// applyStorage applies storage to a fragment that may already have
// usable data. For instance, this would try to remap existing containers
// to use a new storage as backing store.
func (f *fragment) applyStorage(data []byte, file *os.File, newGen generation, mapped bool) (bool, error) {
if len(data) == 0 {
// This shouldn't be used anyway in this path, but just in
// case, we'll be explicit about it.
f.storage.PreferMapping(false)
if file != nil {
fi, err := file.Stat()
if err != nil {
f.holder.Logger.Errorf("trying to apply new storage to existing bitmap, stat failed: %v", err)
}
if err == nil && fi != nil && fi.Size() == 0 {
return f.emptyStorage(file)
}
}
// if we can't be sure of that, we assume data is 0 because
// we couldn't mmap it, and since all we'd be doing is remapping
// our containers to use that storage *to take advantage of
// mmap*, we'll just make sure our containers aren't pointing to
// old storage and say "nope".
_, _ = f.storage.RemapRoaringStorage(nil)
f.storage.SetSource(nil)
return false, nil
}
// Tell storage to prefer mapping if and only if we think the data
// is mmapped and valid.
f.storage.PreferMapping(mapped)
// RemapRoaringStorage will fix any mapped containers to point either
// to the provided data (if PreferMapping was called with true and
// data is provided and there's a corresponding container) or to
// allocated storage, so when it's done, there's nothing in it that
// is mapped to anything *other than* the provided data.
mapped, err := f.storage.RemapRoaringStorage(data)
if err != nil {
// OOPS! something went wrong, we don't know why, we can't
// sanely recover from that.
_, _ = f.storage.RemapRoaringStorage(nil)
mapped = false
f.storage.SetSource(nil)
} else {
f.storage.SetSource(newGen)
}
return mapped, err
}
func (f *fragment) inspectStorage(data []byte, file *os.File, newGen generation, mapped bool) (didMap bool, err error) {
f.bitmapInfo = &roaring.BitmapInfo{}
f.storage, didMap, err = roaring.InspectBinary(data, mapped, f.bitmapInfo)
return didMap, err
}
// openStorage opens the storage bitmap.
//
// This has been massively reworked recently, and now hands a lot of
// file management off to the generation object and the Done method
// of that object. Similarly, the bitmap mapping/remapping
// logic is now mostly in importStorage (reading in a bitmap) and applyStorage
// (remapping an existing bitmap to match a new backing store).
// openStorage opens the storage bitmap. Does nothing in RBF-world and will be removed soon.
func (f *fragment) openStorage(unmarshalData bool) error {
if !f.idx.NeedsSnapshot() {
f.gen = &NopGeneration{}
f.currdata = struct{ from, to uintptr }{}
f.prevdata = f.currdata
return nil // openStorage becomes a noop under RBF, Badger, etc.
}
// Create a roaring bitmap to serve as storage for the shard.
if f.storage == nil {
f.storage = roaring.NewFileBitmap()
f.storage.Flags = f.flags
// if we didn't actually have storage, we *do* need to
// unmarshal this data in order to have any.
unmarshalData = true
}
var storageOp func([]byte, *os.File, generation, bool) (bool, error)
if f.holder.Opts.Inspect {
// note that this will unmarshal even if we already have
// storage; when Inspect is on for a holder, we actually want
// to be able to report this.
storageOp = f.inspectStorage
} else {
if unmarshalData {
storageOp = f.importStorage
} else {
storageOp = f.applyStorage
}
}
var err error
f.gen, err = newGeneration(f.gen, f.path(), unmarshalData, storageOp, f.holder.Logger)
if f.gen != nil {
scratchData := f.gen.Bytes()
f.prevdata = f.currdata
var scratchAddrs struct{ from, to uintptr }
if scratchData != nil {
scratchAddrs.from = uintptr(unsafe.Pointer(&scratchData[0]))
scratchAddrs.to = scratchAddrs.from + uintptr(len(scratchData))
}
f.currdata = scratchAddrs
}
if generationDebug {
// We might have already done this anyway, if we think we
// mapped stuff, but when debugging we want to do it
// unconditionally, because the test cases otherwise won't
// exercise this code well.
f.storage.SetSource(f.gen)
}
return err
return nil
}
// openCache initializes the cache from row ids persisted to disk.
@ -557,17 +412,13 @@ func (f *fragment) close() error {
return nil
}
// closeStorage marks the current generation as done. It is not necessary
// to call this before openStorage.
// closeStorage is essentially a no-op and will go away soon.
func (f *fragment) closeStorage() error {
// opN is determined by how many bit set/clear operations are in the storage
// write log, so once the storage is closed it should be 0. Opening new
// storage will set opN appropriately.
f.opN = 0
if f.gen != nil {
f.gen.Done()
}
return nil
}
@ -640,10 +491,7 @@ func (f *fragment) rowFromStorage(tx Tx, rowID uint64) (*Row, error) {
func (f *fragment) setBit(tx Tx, rowID, columnID uint64) (changed bool, err error) {
f.mu.Lock() // controls access to the file.
defer f.mu.Unlock()
var wp *io.Writer
if f.storage != nil {
wp = &f.storage.OpWriter
}
doSetFunc := func() error {
// handle mutux field type
if f.mutexVector != nil {
@ -654,16 +502,7 @@ func (f *fragment) setBit(tx Tx, rowID, columnID uint64) (changed bool, err erro
changed, err = f.unprotectedSetBit(tx, rowID, columnID)
return err
}
// avoid crashing when f.gen is nil
if f.gen != nil {
err = f.gen.Transaction(wp, doSetFunc)
} else {
if tx.Type() == RoaringTxn {
return changed, errors.New("internal error: f.gen was nil and tx.Type is RoaringTxn - should never happen under roaring b/c storage should be open")
}
// else transactional backend. Just do it.
err = doSetFunc()
}
err = doSetFunc()
return changed, err
}
@ -728,15 +567,7 @@ func (f *fragment) unprotectedSetBit(tx Tx, rowID, columnID uint64) (changed boo
func (f *fragment) clearBit(tx Tx, rowID, columnID uint64) (changed bool, err error) {
f.mu.Lock()
defer f.mu.Unlock()
var wp *io.Writer
if f.storage != nil {
wp = &f.storage.OpWriter
}
err = f.gen.Transaction(wp, func() error {
changed, err = f.unprotectedClearBit(tx, rowID, columnID)
return err
})
return changed, err
return f.unprotectedClearBit(tx, rowID, columnID)
}
// unprotectedClearBit TODO should be replaced by an invocation of
@ -788,15 +619,7 @@ func (f *fragment) unprotectedClearBit(tx Tx, rowID, columnID uint64) (changed b
func (f *fragment) setRow(tx Tx, row *Row, rowID uint64) (changed bool, err error) {
f.mu.Lock()
defer f.mu.Unlock()
var wp *io.Writer
if f.storage != nil {
wp = &f.storage.OpWriter
}
err = f.gen.Transaction(wp, func() error {
changed, err = f.unprotectedSetRow(tx, row, rowID)
return err
})
return changed, err
return f.unprotectedSetRow(tx, row, rowID)
}
func (f *fragment) unprotectedSetRow(tx Tx, row *Row, rowID uint64) (changed bool, err error) {
@ -854,15 +677,7 @@ func (f *fragment) unprotectedSetRow(tx Tx, row *Row, rowID uint64) (changed boo
func (f *fragment) clearRow(tx Tx, rowID uint64) (changed bool, err error) {
f.mu.Lock()
defer f.mu.Unlock()
var wp *io.Writer
if f.storage != nil {
wp = &f.storage.OpWriter
}
err = f.gen.Transaction(wp, func() error {
changed, err = f.unprotectedClearRow(tx, rowID)
return err
})
return changed, err
return f.unprotectedClearRow(tx, rowID)
}
func (f *fragment) unprotectedClearRow(tx Tx, rowID uint64) (changed bool, err error) {
@ -903,11 +718,7 @@ func (f *fragment) clearBlock(tx Tx, block int) (changed bool, err error) {
defer f.mu.Unlock()
firstRow := uint64(block * HashBlockSize)
var wp *io.Writer
if f.storage != nil {
wp = &f.storage.OpWriter
}
err = f.gen.Transaction(wp, func() error {
err = func() error {
var rowChanged bool
for rowID := uint64(firstRow); rowID < firstRow+HashBlockSize; rowID++ {
if chang, err := f.unprotectedClearRow(tx, rowID); err != nil {
@ -918,7 +729,7 @@ func (f *fragment) clearBlock(tx Tx, block int) (changed bool, err error) {
}
changed = rowChanged
return nil
})
}()
return changed, err
}
@ -1028,11 +839,7 @@ func (f *fragment) setValueBase(txOrig Tx, columnID uint64, bitDepth uint64, val
}()
}
var wp *io.Writer
if f.storage != nil {
wp = &f.storage.OpWriter
}
err = f.gen.Transaction(wp, func() error {
err = func() error {
// Convert value to an unsigned representation.
uvalue := uint64(value)
if value < 0 {
@ -1086,7 +893,7 @@ func (f *fragment) setValueBase(txOrig Tx, columnID uint64, bitDepth uint64, val
}
return nil
})
}()
return changed, err
}
@ -2369,11 +2176,6 @@ func (p parallelSlices) Swap(i, j int) {
// operations to the op log.
func (f *fragment) importPositions(tx Tx, set, clear []uint64, rowSet map[uint64]struct{}) error {
//tx.AddN()
var wp *io.Writer
if f.storage != nil {
wp = &f.storage.OpWriter
}
doFunc := func() error {
if len(set) > 0 {
f.stats.Count(MetricImportingN, int64(len(set)), 1)
@ -2420,16 +2222,7 @@ func (f *fragment) importPositions(tx Tx, set, clear []uint64, rowSet map[uint64
}
return nil
}
var err error
if f.gen != nil {
err = f.gen.Transaction(wp, doFunc)
} else {
if tx.Type() == RoaringTxn {
return errors.New("internal error: f.gen was nil and tx.Type is RoaringTxn - should never happen under roaring b/c storage should be open")
}
err = doFunc()
}
err := doFunc()
if err != nil && f.storage != nil {
// we got an error. it's possible that the error indicates that something went wrong.
mappedIn, mappedOut, unmappedIn, errs, e2 := f.storage.SanityCheckMapping(f.currdata.from, f.currdata.to)
@ -2685,11 +2478,7 @@ func (f *fragment) doImportRoaring(ctx context.Context, tx Tx, data []byte, clea
defer span.Finish()
var rowSet map[uint64]int
var wp *io.Writer
if f.storage != nil {
wp = &f.storage.OpWriter
}
err := f.gen.Transaction(wp, func() (err error) {
err := func() (err error) {
var rit roaring.RoaringIterator
rit, err = roaring.NewRoaringIterator(data)
if err != nil {
@ -2698,8 +2487,7 @@ func (f *fragment) doImportRoaring(ctx context.Context, tx Tx, data []byte, clea
_, rowSet, err = tx.ImportRoaringBits(f.index(), f.field(), f.view(), f.shard, rit, clear, true, rowSize)
return err
})
}()
if err != nil {
return nil, false, err
}

6
fragment_internal_test.go
View file

@ -3010,7 +3010,7 @@ func BenchmarkImportRoaringUpdate(b *testing.B) {
b.StopTimer()
var stat os.FileInfo
var statTarget io.Writer
err = f.gen.Transaction(&statTarget, func() error {
err = func() error {
targetFile, ok := statTarget.(*os.File)
if ok {
stat, _ = targetFile.Stat()
@ -3018,7 +3018,7 @@ func BenchmarkImportRoaringUpdate(b *testing.B) {
b.Errorf("couldn't stat file")
}
return nil
})
}()
if err != nil {
b.Errorf("transaction error: %v", err)
}
@ -3486,8 +3486,6 @@ func (f *fragment) Clean(t testing.TB) {
}
}()
errc := f.Close()
// prevent double-closes of generation during testing.
f.gen = nil
if errc != nil {
t.Fatalf("error closing fragment: %v", errc)
}

36
gendebug_test.go
View file

@ -1,36 +0,0 @@
// Copyright 2021 Molecula Corp. All rights reserved.
//
//go:build generationdebug
// +build generationdebug
package pilosa
import (
"errors"
"fmt"
"runtime"
"github.com/molecula/featurebase/v3/testhook"
)
func examineResults() error {
runtime.GC()
stats, results := reportGenerations()
if len(stats) > 0 {
fmt.Printf("generation stats: %s\n", stats)
}
if len(results) == 0 {
return nil
}
if len(results) > 0 {
fmt.Printf("generations:\n")
for _, res := range results {
fmt.Printf(" %s\n", res)
}
}
return errors.New("outstanding generations detected")
}
func init() {
testhook.RegisterPostTestHook(examineResults)
}

450
generation.go
View file

@ -1,450 +0,0 @@
// Copyright 2021 Molecula Corp. All rights reserved.
package pilosa
import (
"fmt"
"io"
"io/ioutil"
"os"
"runtime"
// "runtime/debug"
"sync"
"syscall"
"time"
"github.com/molecula/featurebase/v3/logger"
"github.com/molecula/featurebase/v3/roaring"
"github.com/molecula/featurebase/v3/syswrap"
"github.com/pkg/errors"
)
// generation represents one "generation" of opening a data file.
// This is what determines when it's safe to unmap a data file, if it
// got mapped, and handles closing/reopening files if we need to
// manage file handle availability. It's an interface because this
// lets us write simpler code for specific cases, rather than handling
// the whole matrix of mapped/unmapped, staying open/being reopened,
// etcetera.
//
// You create a generation by calling newGeneration with a file
// path. If it succeeds in opening that path, it calls a provided
// setup function with the data from the generation, and a flag
// indicating whether the data is mmapped. If the setup function
// fails, newGeneration cleans things up and closes. Otherwise,
// it returns a generation.
//
// The generation itself uses runtime.SetFinalizer to clean up when
// the last reference to it goes away. You should store a pointer
// to the generation in any object which is reliant on the generation.
//
// When you anticipate a generation should be done (for instance,
// opening a new generation), the old one gets marked done, which
// stashes a timestamp in it. Later operations can check whether
// the timestamp is a while back, and if so, complain that something
// might be wrong.
//
// In some cases, we don't have enough open file limit to keep every
// file actually open. To address this, use the `Transaction` function,
// which ensures that the file is open, stores a reference to it in
// a provided `*io.Writer`, and then restores the previous value of
// the io.Writer when it's done. For instance, for a bitmap, this might
// be used with `&b.OpWriter`.
//
// newGeneration takes an optional previous generation; it calls
// that generation's Done function after running the provided setup,
// and bumps the generation count.
type generation interface {
// Transaction runs the given transaction with the generation's
// file open. If the **os.File parameter is
// non-nil, the generation's file will be open, and stored
// into that pointer, during the execution of func, after
// which the previous contents are restored. Otherwise
// the file may or may not be open during the operation.
Transaction(*io.Writer, func() error) error
// Done() should be called exactly once, to indicate that a
// generation is expected not to be in use for long -- for instance,
// when a new generation replaces it.
Done()
// Generation count.
Generation() int64
// ID indicates the source -- path and generation number -- that
// this generation represents.
ID() string
// Dead indicates whether this generation is Done.
Dead() bool
// Bytes reports the storage associated with this generation, if any.
// DO NOT USE THIS. Except if you're debugging mmap segfaults.
Bytes() []byte
}
type mmapGeneration struct {
mu sync.Mutex // mutex guards modifiers of generation, not of data
transMu sync.Mutex // guards transactions, specifically
path string
id string
file *os.File
data []byte
generation int64 // generation counter
dead bool // we think this generation is dead
deadSince time.Time // when this generation was marked dead
retries int // for cases where we're retrying
logger logger.Logger
}
func (m *mmapGeneration) Dead() bool {
m.mu.Lock()
defer m.mu.Unlock()
return m.dead
}
func (m *mmapGeneration) ID() string {
return m.id
}
func (m *mmapGeneration) Generation() int64 {
return m.generation
}
// Transaction runs an exclusive call, ensuring that the file is open if
// the *io.Writer parameter is present.
func (m *mmapGeneration) Transaction(fileP *io.Writer, fn func() error) (transactionErr error) {
m.transMu.Lock()
defer m.transMu.Unlock()
// HEY LOOK CAREFULLY AT THIS BIT:
// We can't just defer this unlock. We specifically want to be
// sure to unlock the regular mutex *before* this function is over,
// and if we error out trying to open the file, we want to do it
// even sooner. If we deferred this, the transaction would block
// *everything*, including things like sanity checks against the
// generation being Dead(), but also including the deferred
// re-close-the-file.
m.mu.Lock()
// if we've been asked for a file pointer, we need to ensure that
// our file is open, and that the file pointer to it is stored in
// the requested location, then revert that when we're done.
// if we aren't asked for a file pointer, nothing needs the file
// open.
if m.dead {
elapsed := time.Since(m.deadSince)
m.logger.Warnf("transaction against %s, which has been dead for %v\n", m.id, elapsed)
}
if fileP != nil {
if m.file == nil {
// we ignore the shouldClose response here; if this
// fragment was previously not being kept open, we're
// going to stick with that.
_, err := m.openFile()
if err != nil {
m.mu.Unlock()
return err
}
defer func() {
// report a close error if we have no other error to report
m.mu.Lock()
defer m.mu.Unlock()
err := m.closeFile()
if transactionErr == nil {
transactionErr = err
}
}()
}
var fileStash io.Writer
fileStash, *fileP = *fileP, m.file
defer func() {
*fileP = fileStash
}()
}
// We are done locking the generation itself for now.
m.mu.Unlock()
// wouldPanic := debug.SetPanicOnFault(true)
// defer func() {
// debug.SetPanicOnFault(wouldPanic)
// if r := recover(); r != nil {
// if err, ok := r.(error); ok {
// // special case: if we caught a page fault, we diagnose that directly. sadly,
// // we can't see the actual values that were used to generate this, probably.
// if err.Error() == "runtime error: invalid memory address or nil pointer dereference" {
// if transactionErr == nil {
// transactionErr = errors.New("invalid memory access during transaction")
// } else {
// transactionErr = fmt.Errorf("invalid memory access during transaction, previous error %v", transactionErr)
// }
// return
// }
// }
// if transactionErr == nil {
// transactionErr = fmt.Errorf("panic during transaction: %v", r)
// } else {
// transactionErr = fmt.Errorf("panic during erroring transaction: panic %v, previous error %v", r, transactionErr)
// }
// }
// }()
return fn()
}
func (m *mmapGeneration) Bytes() []byte {
return m.data
}
// Done marks the generation done, and closes its file, but may not unmap it.
// It's still conceptually possible to end up doing a Transaction against a
// done generation, but it's a red flag.
func (m *mmapGeneration) Done() {
if m == nil {
return
}
m.mu.Lock()
defer m.mu.Unlock()
if m.dead {
oops := fmt.Sprintf("generation %s, marked done again at %v, previously marked dead at %v",
m.id, time.Now(), m.deadSince)
panic(oops)
}
m.dead = true
m.deadSince = time.Now()
err := m.closeFile()
if err != nil {
m.logger.Errorf("error closing generation %s: %v", m.id, err)
}
// If we're not debugging, the finalizer won't have been enabled
// previously. Finalizers have non-zero cost, so having them not be
// created until they're needed seems rewarding?
if !generationDebug {
runtime.SetFinalizer(m, generationFinalizer)
}
endGeneration(m.id)
// note, Done() doesn't close the file; only the finalizer actually
// does the shutdown.
}
// Try to close the file if it's currently open.
func (m *mmapGeneration) closeFile() error {
var lastErr error
// report the most serious error encountered, but still close
// file even if something else failed.
if m.file != nil {
if err := m.file.Sync(); err != nil {
lastErr = fmt.Errorf("sync: %s", err)
}
if err := syscall.Flock(int(m.file.Fd()), syscall.LOCK_UN); err != nil {
lastErr = fmt.Errorf("unlock: %s", err)
}
if err := syswrap.CloseFile(m.file); err != nil {
lastErr = fmt.Errorf("close file: %s", err)
}
m.file = nil
}
return lastErr
}
// openFile ensures the file is open and locked, or fails. If it does
// open the file, it will also report the "you need to close this file
// when you're done" flag from syswrap.
func (m *mmapGeneration) openFile() (shouldClose bool, err error) {
if m.file != nil {
return false, nil
}
m.file, shouldClose, err = syswrap.OpenFile(m.path, os.O_RDWR|os.O_CREATE|os.O_APPEND, 0666)
if err != nil {
return false, err
}
// do we actually want this in every openFile? I don't know.
if err := syscall.Flock(int(m.file.Fd()), syscall.LOCK_EX|syscall.LOCK_NB); err != nil {
_ = syswrap.CloseFile(m.file)
m.file = nil
return false, fmt.Errorf("flock: %s", err)
}
return shouldClose, nil
}
func generationFinalizer(m *mmapGeneration) {
m.mu.Lock()
if !m.dead {
m.logger.Infof("finalizing generation %s which isn't dead yet\n",
m.id)
}
m.mu.Unlock()
err := m.closeFile()
if err != nil {
m.logger.Errorf("finalizing generation, closing file: %v\n", err)
}
if m.data != nil {
err := syswrap.Munmap(m.data)
if err != nil {
m.logger.Errorf("finalizing generation, munmap: %v\n", err)
}
m.data = nil
}
finalizeGeneration(m.id)
}
// Cancel closes a generation out entirely. It cancels any finalizer,
// unmaps any data, ends generation tracking, and closes any files.
// It does each of these separately whether or not the others need to be done,
// or succeed. It's used to handle failures from newGeneration; it makes sure
// the generation isn't holding any resources and doesn't need to be cleaned
// up otherwise.
//
// Mostly a helper function because there's several cases where newGeneration
// might fail.
func (m *mmapGeneration) Cancel() {
if m.data != nil {
_ = syswrap.Munmap(m.data)
m.data = nil
}
err := m.closeFile()
if err != nil {
m.logger.Errorf("error cancelling generation %s: %v", m.id, err)
}
runtime.SetFinalizer(m, nil)
m.dead = true
m.deadSince = time.Now()
cancelGeneration(m.id)
}
// newGeneration creates a new generation using the given file path. It
// then calls the provided setup function with the allocated storage, a
// file handle, the new generation, and a flag indicatting whether the storage
// is memory-mapped. If the setup function returns a non-nil error, the
// generation is cleaned up, and newGeneration fails. The setup function
// also returns a boolean indicating whether it used the mapping; if it
// didn't, newGeneration discards the mapping and returns a nil generation.
//
// If generationDebug is enabled, we track the generation even if no mapping
// is actually in use, so we can verify that the tracking is working.
//
// On failure, newGeneration returns nil values for generation and func,
// and an error. On success, the func returned is the close func to use
// when the generation is no longer needed by the caller.
func newGeneration(existing generation, path string, readData bool, setup func([]byte, *os.File, generation, bool) (bool, error), logger logger.Logger) (generation, error) {
m := mmapGeneration{path: path, logger: logger}
if existing != nil {
m.generation = existing.Generation() + 1
m.retries = existing.(*mmapGeneration).retries
// we might keep a previous generation around just for its generation count.
if !existing.Dead() {
defer existing.Done()
}
}
shouldClose, err := m.openFile()
if err != nil {
return nil, err
}
m.id = fmt.Sprintf("%s:%d", m.path, m.generation)
// possibly assign new generation ID if this one's been used, which can
// happen with reopens, especially during testing.
m.id = registerGeneration(m.id)
// if debugging, we always want the finalizer on so we notice if a
// generation is finalized without being closed. for non-debugging
// use, we only need it when the generation is closed.
if generationDebug {
runtime.SetFinalizer(&m, generationFinalizer)
}
// Mmap the underlying file so it can be zero copied.
var mapped bool
var data []byte
fi, err := m.file.Stat()
if err == nil && fi.Size() > 0 {
data, err = syswrap.Mmap(int(m.file.Fd()), 0, int(fi.Size()), syscall.PROT_READ, syscall.MAP_SHARED)
if err == syswrap.ErrMaxMapCountReached {
// I have no idea where/how to display this message.
m.logger.Warnf("maximum number of maps reached, reading file '%s' instead", m.path)
} else if err != nil {
m.Cancel()
return nil, errors.Wrap(err, "mmap failed")
} else {
mapped = true
}
}
if data == nil && readData {
data, err = ioutil.ReadAll(m.file)
if err != nil {
m.Cancel()
return nil, errors.Wrap(err, "failure file readall")
}
}
// if we got here, data's the expected data, so let's try to use it
mappedAny, err := setup(data, m.file, &m, mapped)
// if the setup failed, we unmap data if we previously mapped it,
// and exit. Note that having no data, or having only trivial
// data (like a zero-container Roaring file) isn't "failed".
if err != nil {
m.Cancel()
// Unless, that is, we think the file probably ought to
// be truncated: For instance, if a bitmap has a corrupted
// ops log, we could truncate that part of it and retry.
if err, ok := err.(roaring.FileShouldBeTruncatedError); ok && m.retries < 1 {
m.logger.Infof("file %s read partially, but should-be-truncated at %d bytes\n", m.path, err.SuggestedLength())
// close this generation, then try again. once.
m.retries++
err := os.Truncate(m.path, err.SuggestedLength())
if err != nil {
m.logger.Errorf("truncating file failed [but retrying anyway]: %v\n", err)
}
return newGeneration(&m, path, readData, setup, logger)
}
return nil, err
}
if mapped {
// when generationDebug is on, we want to track this even
// if it's not being used.
if generationDebug || mappedAny {
// Advise the kernel that the mmap is accessed randomly.
// We don't care much about errors with this.
_ = madvise(data, syscall.MADV_RANDOM)
// store the data, so we can unmap it when this generation
// gets finalized.
m.data = data
} else {
// unmap the data and don't stash the pointer in this
// generation. It's not being used. This generation
// doesn't need to exist, yay.
unmapErr := syswrap.Munmap(data)
if unmapErr != nil {
m.logger.Errorf("error unmapping (probably harmless): %v", unmapErr)
}
}
}
// shouldClose comes from underlying syswrap.OpenFile, which checks
// a count of open files to hint at us when we need to start closing
// files to preserve open file descriptor limit.
if shouldClose {
err := m.closeFile()
if err != nil {
m.logger.Errorf("closing file to preserve open files failed: %v\n", err)
}
}
// It's possible that the generation has no actual data to track,
// because nothing's mapped, in which case there won't be any bitmap
// sources following this, just the fragment source. (Bitmaps won't
// be attached to the source unless they're actually mapped to it,
// or generationDebug is true). That's okay. We pay a tiny cost
// for the finalizer, but we also get higher confidence that it really
// does get cleaned up.
return &m, nil
}
// NopGeneration is used in fragment.openStorage() to short-circuit
// generation stuff that only applies to RoaringTx; doesn't apply to RBFTx/BadgerTx/etc.
type NopGeneration struct {
}
func (g *NopGeneration) Transaction(w *io.Writer, f func() error) error {
return f()
}
func (g *NopGeneration) Done() {}
func (g *NopGeneration) Generation() int64 {
return 0
}
func (g *NopGeneration) ID() string {
return "NOP"
}
func (g *NopGeneration) Dead() bool {
return true
}
func (g *NopGeneration) Bytes() (ret []byte) {
return
}

152
generation_debug.go
View file

@ -1,152 +0,0 @@
// Copyright 2021 Molecula Corp. All rights reserved.
//go:build generationdebug
// +build generationdebug
package pilosa
import (
"fmt"
"math/rand"
"runtime"
"runtime/debug"
"sort"
"sync"
"time"
)
const generationDebug = true
type lifespan struct {
from, to, finalized time.Time
stack []byte
}
var knownGenerations map[string]lifespan
var knownGenerationLock sync.Mutex
var timeZero time.Time
var generationDebugVerbose bool
// History reports the finalized/dead/created status of a span which we think
// is in some way in error. It's shared between a couple of places.
func (span *lifespan) History() string {
dead := "not dead"
finalized := "not finalized"
if span.finalized != timeZero {
finalized = fmt.Sprintf("finalized at %v", span.finalized)
}
if span.to != timeZero {
dead = fmt.Sprintf("dead at %v", span.to)
}
return fmt.Sprintf("%s, %s, created at %v at %s", dead, finalized, span.from, span.stack)
}
func (span *lifespan) reportHistory(reason string, id string) string {
return fmt.Sprintf("%s %s: %s", id, reason, span.History())
}
func registerGeneration(id string) string {
knownGenerationLock.Lock()
defer knownGenerationLock.Unlock()
if knownGenerations == nil {
knownGenerations = make(map[string]lifespan)
}
newSpan := lifespan{from: time.Now(), stack: debug.Stack()}
origId := id
// if you have more than 65k of the same file open, maybe you have bigger
// problems than this.
for span, exists := knownGenerations[id]; exists; span, exists = knownGenerations[id] {
suffix := fmt.Sprintf("::%04x", rand.Int63n(65536))
if generationDebugVerbose {
history := span.History()
fmt.Printf("new generation: adding suffix %s, previous %s\n",
suffix, history)
}
id = origId + suffix
}
if generationDebugVerbose {
fmt.Printf("new generation %s\n", id)
}
knownGenerations[id] = newSpan
return id
}
func endGeneration(id string) {
knownGenerationLock.Lock()
defer knownGenerationLock.Unlock()
span, exists := knownGenerations[id]
if !exists {
oops := fmt.Sprintf("ending generation %s: unknown", id)
panic(oops)
}
if span.finalized != timeZero || span.to != timeZero {
panic(span.reportHistory("ending generation", id))
}
span.to = time.Now()
knownGenerations[id] = span
}
// cancelGeneration marks the generation as finalized. In principle it's
// only used in cases where we just started a generation but something
// went wrong. it's not fancier than this because of the weird cases
// where the same generation shows up again, such as when closing and
// reopening an index so we don't know about previous instances of the
// same files.
func cancelGeneration(id string) {
knownGenerationLock.Lock()
defer knownGenerationLock.Unlock()
span, exists := knownGenerations[id]
if exists {
span.finalized = time.Now()
span.to = span.finalized
knownGenerations[id] = span
}
}
func finalizeGeneration(id string) {
knownGenerationLock.Lock()
defer knownGenerationLock.Unlock()
span, exists := knownGenerations[id]
if !exists {
oops := fmt.Sprintf("finalizing generation %s: unknown", id)
panic(oops)
}
if span.finalized != timeZero {
panic(span.reportHistory("finalizing", id))
}
span.finalized = time.Now()
knownGenerations[id] = span
}
func reportGenerations() (stats string, surviving []string) {
runtime.GC()
knownGenerationLock.Lock()
defer knownGenerationLock.Unlock()
times := make([]int64, 0, len(knownGenerations))
for id, span := range knownGenerations {
if span.to == timeZero || span.finalized == timeZero {
surviving = append(surviving, span.reportHistory("surviving", id))
} else {
times = append(times, int64(span.finalized.Sub(span.to)))
}
}
stats = "no recorded finalized spans"
if len(times) > 0 {
sort.Slice(times, func(i, j int) bool { return times[i] < times[j] })
var total int64
for _, d := range times {
total += d
}
var mean, median, p90, p99, worst int64
mean = total / int64(len(times))
median = times[len(times)/2]
p90 = times[(len(times)*9)/10]
p99 = times[(len(times)*99)/100]
worst = times[len(times)-1]
stats = fmt.Sprintf("%d finalized spans. lag: mean %v, median %v, p90 %v, p99 %v, worst %v",
len(times), time.Duration(mean), time.Duration(median), time.Duration(p90), time.Duration(p99), time.Duration(worst))
}
return stats, surviving
}

25
generation_nodebug.go
View file

@ -1,25 +0,0 @@
// Copyright 2021 Molecula Corp. All rights reserved.
//go:build !generationdebug
// +build !generationdebug
package pilosa
const generationDebug = false
func registerGeneration(id string) string {
return id
}
func endGeneration(id string) {
}
func cancelGeneration(id string) {
}
func finalizeGeneration(id string) {
}
//lint:ignore U1000 this is conditional on a build flag, see generation_test.go.
func reportGenerations() []string { //nolint:unused,deadcode
return nil
}

59
generation_test.go
View file

@ -1,59 +0,0 @@
// Copyright 2021 Molecula Corp. All rights reserved.
//
//go:build generationparanoia
// +build generationparanoia
package pilosa
import (
"runtime"
"testing"
"unsafe"
)
func TestGenerationPanic(t *testing.T) {
f := mustOpenFragment("i", "f", viewStandard, 0, "none")
defer f.Clean(t)
for i := 0; i < f.MaxOpN; i++ {
_, _ = f.setBit(0, uint64(i*32))
}
// force snapshot so we get a mmapped row...
_ = f.Snapshot()
_ = f.row(0)
var prevData []byte
if f.gen.(*mmapGeneration).data == nil {
t.Fatalf("generation code didn't create a mapping, apparently?")
}
prevData = f.gen.(*mmapGeneration).data
f.mu.Lock()
_ = defaultSnapshotQueue.Immediate(f)
f.mu.Unlock()
runtime.GC()
for i := 0; i < (f.MaxOpN / 2); i++ {
_, _ = f.setBit(0, uint64(i*32)+23)
}
f.mu.Lock()
defaultSnapshotQueue.Await(f)
f.mu.Unlock()
runtime.GC()
newData := f.gen.(*mmapGeneration).data
if unsafe.Pointer(&prevData[0]) == unsafe.Pointer(&newData[0]) {
t.Fatalf("test can't run usefully, didn't get new data pointer")
}
var wp *io.Writer
if f.storage != nil {
wp = &f.storage.OpWriter
}
err := f.gen.Transaction(wp, func() error {
prevData[0] = 0x3c
return nil
})
if err == nil {
t.Fatalf("expected a panic to get caught, but nothing happened")
}
if err.Error() != "invalid memory access during transaction" {
t.Fatalf("expected \"invalid memory access during transaction\", got %q", err.Error())
}
}

2
holder.go
View file

@ -653,7 +653,7 @@ func (h *Holder) Open() error {
return errors.Wrap(err, "opening index")
}
// Since we don't have createAt stored on disk within the data
// Since we don't have createdAt stored on disk within the data
// directory, we need to populate it from the etcd schema data.
// TODO: we may no longer need the createdAt value stored in memory on
// the index struct; it may only be needed in the schema return value

28
mmap_test.go
View file

@ -2,13 +2,11 @@
package pilosa
import (
"fmt"
"math/rand"
"runtime"
"testing"
"github.com/molecula/featurebase/v3/logger"
"github.com/molecula/featurebase/v3/syswrap"
)
type cv struct {
@ -68,29 +66,3 @@ func forceSnapshotsCheckMapping(t *testing.T) {
}
}
}
// This test should basically never fail, but it might if you were running
// out of available mmaps. Which you can fake up by adding '&& false' to the test
// in newGeneration in generation.go. So this is probably useless but it's
// a failure mode we've been bitten by once...
func TestMmapBehavior(t *testing.T) {
// rbf and lmdb not happy with this test.
roaringOnlyTest(t)
var changed bool
var original uint64
defer func() {
syswrap.SetMaxMapCount(original)
}()
for _, mmapMaxVal := range []uint64{0, 3} {
prev := syswrap.SetMaxMapCount(mmapMaxVal)
if !changed {
original = prev
changed = true
}
t.Run(fmt.Sprintf("maps%d", mmapMaxVal), func(t *testing.T) {
forceSnapshotsCheckMapping(t)
})
}
}

2
roaring/filter.go
View file

@ -579,7 +579,7 @@ func (b *BitmapRowFilterMultiFilter) ConsiderData(key FilterKey, data *Container
// offsets the input bitmap's containers have, it matches them against
// corresponding keys.
type BitmapBitmapFilter struct {
filter *Bitmap // We don't use this while iterating, but in ludicrous edge cases it might be holding a generation we need.
filter *Bitmap // We don't use this while iterating, but in ludicrous edge cases it might be holding a generation we need. TODO @seebs I don't understand why this mentions generations
containers []*Container
nextOffsets []uint64
callback func(uint64) error

7
roaring/generation_debug.go
View file

@ -1,7 +0,0 @@
// Copyright 2021 Molecula Corp. All rights reserved.
//go:build generationdebug
// +build generationdebug
package roaring
const generationDebug = true

7
roaring/generation_nodebug.go
View file

@ -1,7 +0,0 @@
// Copyright 2021 Molecula Corp. All rights reserved.
//go:build !generationdebug
// +build !generationdebug
package roaring
const generationDebug = false

3
roaring/roaring.go
View file

@ -620,8 +620,7 @@ func (b *Bitmap) OffsetRange(offset, start, end uint64) *Bitmap {
}
other.Containers.Put(off+(k-hi0), c.Freeze())
}
// if b.Source != nil && mappedAny {
if b.Source != nil && (generationDebug || mappedAny) {
if b.Source != nil && mappedAny {
other.Source = b.Source
}
return other

49
rrtx_internal_test.go
View file

@ -1,49 +0,0 @@
// Copyright 2021 Molecula Corp. All rights reserved.
package pilosa
import (
"testing"
. "github.com/molecula/featurebase/v3/vprint" // nolint:staticcheck
)
func TestRoaring_HasData(t *testing.T) {
holder := newHolderWithTempPath(t, "roaring")
idx, err := holder.CreateIndex("i", IndexOptions{})
PanicOn(err)
defer idx.Close()
db, err := globalRoaringReg.OpenDBWrapper(idx.path, false, nil)
PanicOn(err)
db.SetHolder(idx.holder)
// HasData should start out false.
hasAnything, err := db.HasData()
PanicOn(err)
if hasAnything {
t.Fatalf("HasData reported existing data on an empty database")
}
// check that HasData sees a committed record.
field, shard := "f", uint64(123)
tx := idx.holder.txf.NewTx(Txo{Write: writable, Index: idx, Shard: shard})
defer tx.Rollback()
f, err := idx.CreateField(field)
PanicOn(err)
_, err = f.SetBit(tx, 1, 1, nil)
PanicOn(err)
PanicOn(tx.Commit())
hasAnything, err = db.HasData()
if err != nil {
t.Fatal(err)
}
if !hasAnything {
t.Fatalf("HasData() reported no data on a database that has 'x' written to it")
}
}