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

rip out generation stuff

Browse source 
it was somewhat difficult to avoid ripping this out without also
touching some of the stuff that supports roaring backend. That's going
soon too, so no worries :)
This commit is contained in:
Matthew Jaffee 2022-01-21 17:00:45 -06:00
parent 1d7a42c7e7
commit 82c75851df
13 changed files with 20 additions and 999 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

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