diff --git a/Makefile b/Makefile index 2b65141db..37ed97f17 100644 --- a/Makefile +++ b/Makefile @@ -79,9 +79,6 @@ testvsub-race: cd ..; \ done -tour: - ./tournament.sh - bench: $(GO) test ./... -bench=. -run=NoneZ -timeout=127m $(TESTFLAGS) @@ -349,14 +346,5 @@ install-gometalinter: GO111MODULE=off gometalinter --install GO111MODULE=off $(GO) get github.com/remyoudompheng/go-misc/deadcode -test-txstore-rbf: - PILOSA_STORAGE_BACKEND=rbf $(MAKE) testv-race - -# WARNING: This feature is no longer being tested regularly in CI. The test is -# very slow and very expensive, and we're not sure it actually provides useful -# information now. -test-txstore-rbf_bolt: - PILOSA_STORAGE_BACKEND=rbf_bolt $(MAKE) testv-race - test-external-lookup: $(GO) test . -tags='$(BUILD_TAGS) $(TEST_TAGS)' $(TESTFLAGS) -run ^TestExternalLookup$$ -externalLookupDSN $(EXTERNAL_LOOKUP_DSN) diff --git a/ctl/server.go b/ctl/server.go index 98391f761..1482c3314 100644 --- a/ctl/server.go +++ b/ctl/server.go @@ -75,13 +75,7 @@ func BuildServerFlags(cmd *cobra.Command, srv *server.Command) { flags.IntVar(&srv.Config.Profile.BlockRate, "profile.block-rate", srv.Config.Profile.BlockRate, "Sampling rate for goroutine blocking profiler. One sample per ns.") flags.IntVar(&srv.Config.Profile.MutexFraction, "profile.mutex-fraction", srv.Config.Profile.MutexFraction, "Sampling fraction for mutex contention profiling. Sample 1/ of events.") - // Storage - // Note: the default for --storage.backend must be kept "" empty string. - // Otherwise we cannot detect and honor the PILOSA_STORAGE_BACKEND env var - // over-ride. - // TODO: the comment above was carried over from the PILOSA_TXSRC flag, but - // we should confirm that this still applies. - flags.StringVar(&srv.Config.Storage.Backend, "storage.backend", storage.DefaultBackend, fmt.Sprintf("transaction/storage to use: one of roaring or rbf. The default is: %v. The env var PILOSA_STORAGE_BACKEND is over-ridden by --storage.backend option on the command line.", storage.DefaultBackend)) + flags.StringVar(&srv.Config.Storage.Backend, "storage.backend", storage.DefaultBackend, fmt.Sprintf("transaction/storage to use: 'rbf' is only supported value.", storage.DefaultBackend)) flags.BoolVar(&srv.Config.Storage.FsyncEnabled, "storage.fsync", true, "enable fsync fully safe flush-to-disk") // RBF specific flags. See pilosa/rbf/cfg/cfg.go for definitions. diff --git a/dbshard.go b/dbshard.go index ee738458f..7e7387c68 100644 --- a/dbshard.go +++ b/dbshard.go @@ -67,9 +67,8 @@ type DBShard struct { Shard uint64 Open bool - typ txtype - styp string - hasRoaring bool // if either of the types is roaringTxn + typ txtype + styp string W DBWrapper ParentDBIndex *DBIndex @@ -131,8 +130,7 @@ type DBPerShard struct { // Easily see how many we have. Flatmap map[flatkey]*DBShard - typ txtype - hasRoaring bool + typ txtype txf *TxFactory holder *Holder @@ -269,11 +267,6 @@ func (txf *TxFactory) NewDBPerShard(typ txtype, holderDir string, holder *Holder vprint.PanicOn("must have holder.cfg.RBFConfig and holder.cfg.StorageConfig set here") } - hasRoaring := false - if typ == roaringTxn { - hasRoaring = true - } - d = &DBPerShard{ typ: typ, HolderDir: holderDir, @@ -281,7 +274,6 @@ func (txf *TxFactory) NewDBPerShard(typ txtype, holderDir string, holder *Holder dbh: NewDBHolder(), Flatmap: make(map[flatkey]*DBShard), txf: txf, - hasRoaring: hasRoaring, index2shards: newIndex2Shards(), StorageConfig: holder.cfg.StorageConfig, RBFConfig: holder.cfg.RBFConfig, @@ -407,10 +399,7 @@ func (per *DBPerShard) unprotectedGetDBShard(index string, shard uint64, idx *In } dbs, ok = dbi.Shard[shard] if dbs != nil && dbs.closed { - // roaring txn are nil/fake anyway. Don't freak out. - if per.typ != roaringTxn { - vprint.PanicOn(fmt.Sprintf("cannot retain closed dbs across holder ReOpen dbs='%p'; per.typ='%v'", dbs, per.typ)) - } + vprint.PanicOn(fmt.Sprintf("cannot retain closed dbs across holder ReOpen dbs='%p'; per.typ='%v'", dbs, per.typ)) } if !ok { dbs = &DBShard{ @@ -421,7 +410,6 @@ func (per *DBPerShard) unprotectedGetDBShard(index string, shard uint64, idx *In HolderPath: per.HolderDir, idx: idx, per: per, - hasRoaring: per.hasRoaring, } dbs.styp = per.typ.String() dbi.Shard[shard] = dbs @@ -430,8 +418,6 @@ func (per *DBPerShard) unprotectedGetDBShard(index string, shard uint64, idx *In if !dbs.Open { var registry DBRegistry switch dbs.typ { - case roaringTxn: - registry = globalRoaringReg case rbfTxn: registry = globalRbfDBReg registry.(*rbfDBRegistrar).SetRBFConfig(per.RBFConfig) @@ -470,8 +456,6 @@ func (f *TxFactory) GetShardsForIndex(idx *Index, roaringViewPath string, requir return f.dbPerShard.TypedDBPerShardGetShardsForIndex(f.typ, idx, roaringViewPath, requireData) } -// if roaringViewPath is "" then for ty == roaringTxn we go to disk to discover -// all the view paths under idx for type ty. // requireData means open the database file and verify that at least one key is set. // The returned sliceOfShards should not be modified. We will cache it for subsequent // queries. @@ -485,14 +469,6 @@ func (per *DBPerShard) TypedDBPerShardGetShardsForIndex(ty txtype, idx *Index, r per.Mu.Lock() defer per.Mu.Unlock() - if ty == roaringTxn && roaringViewPath != "" { - shardMap, err := roaringMapOfShards(roaringViewPath) - if err != nil { - return nil, err - } - return shardMap, nil - } - i2ss := per.index2shards ss, ok := i2ss[idx.name] @@ -507,27 +483,6 @@ func (per *DBPerShard) TypedDBPerShardGetShardsForIndex(ty txtype, idx *Index, r // Upon return, cache the setOfShards value and reuse it next time - if ty == roaringTxn { - // INVAR: roaringViewPath == "", because the other case is - // handled above. - fields := idx.Fields() - for _, field := range fields { - for _, view := range field.views() { - shardMap, err := roaringMapOfShards(view.path) - if err != nil { - return nil, - errors.Wrap(err, fmt.Sprintf( - "TypedDBPerShardGetLocalShardsForIndex roaringTxn view.path='%v'", view.path)) - } - for shard := range shardMap { - setOfShards.add(shard) - } - } - } - return setOfShards.CloneMaybe(), nil - } - // INVAR: not-roaring. - path := per.prefixForType(idx, ty) ignoreEmpty := false @@ -730,8 +685,6 @@ func (per *DBPerShard) GetFieldView2ShardsMapForIndex(idx *Index) (vs *FieldView ty := per.typ switch ty { - case roaringTxn: - return roaringGetFieldView2Shards(idx) default: vs = NewFieldView2Shards() diff --git a/dbshard_internal_test.go b/dbshard_internal_test.go index ceb63ab89..38f752425 100644 --- a/dbshard_internal_test.go +++ b/dbshard_internal_test.go @@ -71,7 +71,7 @@ func Test_DBPerShard_GetShardsForIndex_LocalOnly(t *testing.T) { v2s.addViewShardSet(txkey.FieldView{Field: field, View: "standard"}, stdShardSet) } - for _, src := range []string{"roaring", "rbf"} { + for _, src := range []string{"rbf"} { cfg := mustHolderConfig() cfg.StorageConfig.Backend = src holder := NewHolder(tmpdir, cfg) @@ -82,7 +82,6 @@ func Test_DBPerShard_GetShardsForIndex_LocalOnly(t *testing.T) { idx, err = NewIndex(holder, filepath.Join(tmpdir, index), index) PanicOn(err) } - estd := "rick/fields/_exists/views/standard" std := "rick/fields/f/views/standard" shards, err := holder.txf.GetShardsForIndex(idx, tmpdir+sep+std, false) @@ -93,65 +92,23 @@ func Test_DBPerShard_GetShardsForIndex_LocalOnly(t *testing.T) { panic(fmt.Sprintf("missing shard=%v from shards='%#v'", shard, shards)) } } - if src == "roaring" { - // check estd too - shards, err = holder.txf.GetShardsForIndex(idx, tmpdir+sep+estd, false) + for _, shard := range []uint64{93, 223, 221, 215, 219, 217} { + tx := idx.holder.txf.NewTx(Txo{Write: !writable, Index: idx, Shard: shard}) + fvs, err := tx.GetSortedFieldViewList(idx, shard) PanicOn(err) - for _, shard := range []uint64{93, 223, 221, 215, 219, 217} { - if !shards[shard] { - panic(fmt.Sprintf("missing shard=%v from shards='%#v'", shard, shards)) - } + // expect these same two field/views for all 6 shards + expect0 := txkey.FieldView{Field: "_exists", View: "standard"} + expect1 := txkey.FieldView{Field: "f", View: "standard"} + if len(fvs) != 2 { + panic(fmt.Sprintf("fvs should be len 2, got '%#v' (%s)", fvs, src)) } - - // check GetSortedFieldViewList() and roaringGetFieldView2Shards() - vs, err := roaringGetFieldView2Shards(idx) - PanicOn(err) - - for _, shard := range []uint64{93, 223, 221, 215, 219, 217} { - tx := idx.holder.txf.NewTx(Txo{Write: !writable, Index: idx, Shard: shard}) - fvs, err := tx.GetSortedFieldViewList(idx, shard) - PanicOn(err) - // expect these same two field/views for all 6 shards - expect0 := txkey.FieldView{Field: "_exists", View: "standard"} - expect1 := txkey.FieldView{Field: "f", View: "standard"} - if len(fvs) != 2 { - panic(fmt.Sprintf("fvs should be len 2, got '%#v' (%s)", fvs, src)) - } - if fvs[0] != expect0 { - panic(fmt.Sprintf("expected fvs[0]='%#v', but got '%#v'", expect0, fvs[0])) - } - if fvs[1] != expect1 { - panic(fmt.Sprintf("expected fvs[1]='%#v', but got '%#v'", expect1, fvs[1])) - } - - for _, fv := range fvs { - if !vs.has(fv.Field, fv.View, shard) { - panic(fmt.Sprintf("vs did not contain fv='%#v' for shard %v", fv, shard)) - } - } - tx.Rollback() + if fvs[0] != expect0 { + panic(fmt.Sprintf("expected fvs[0]='%#v', but got '%#v'", expect0, fvs[0])) } - } else { - // non-roaring: rbf - - for _, shard := range []uint64{93, 223, 221, 215, 219, 217} { - tx := idx.holder.txf.NewTx(Txo{Write: !writable, Index: idx, Shard: shard}) - fvs, err := tx.GetSortedFieldViewList(idx, shard) - PanicOn(err) - // expect these same two field/views for all 6 shards - expect0 := txkey.FieldView{Field: "_exists", View: "standard"} - expect1 := txkey.FieldView{Field: "f", View: "standard"} - if len(fvs) != 2 { - panic(fmt.Sprintf("fvs should be len 2, got '%#v' (%s)", fvs, src)) - } - if fvs[0] != expect0 { - panic(fmt.Sprintf("expected fvs[0]='%#v', but got '%#v'", expect0, fvs[0])) - } - if fvs[1] != expect1 { - panic(fmt.Sprintf("expected fvs[1]='%#v', but got '%#v'", expect1, fvs[1])) - } - tx.Rollback() + if fvs[1] != expect1 { + panic(fmt.Sprintf("expected fvs[1]='%#v', but got '%#v'", expect1, fvs[1])) } + tx.Rollback() } holder.Close() } diff --git a/executor_test.go b/executor_test.go index 1c534ac52..d22db1c9b 100644 --- a/executor_test.go +++ b/executor_test.go @@ -33,7 +33,6 @@ import ( "github.com/molecula/featurebase/v3/pql" "github.com/molecula/featurebase/v3/proto" "github.com/molecula/featurebase/v3/server" - "github.com/molecula/featurebase/v3/storage" "github.com/molecula/featurebase/v3/test" "github.com/molecula/featurebase/v3/testhook" . "github.com/molecula/featurebase/v3/vprint" // nolint:staticcheck @@ -1277,15 +1276,6 @@ func TestExecutor_Execute_Count(t *testing.T) { } -func roaringOnlyTest(t *testing.T) { - src := pilosa.CurrentBackend() - if src == pilosa.RoaringTxn || (storage.DefaultBackend == pilosa.RoaringTxn && src == "") { - // okay to run, we are under roaring only - } else { - t.Skip("skip for everything but roaring") - } -} - // Ensure a set query can be executed. func TestExecutor_Execute_Set(t *testing.T) { t.Run("RowIDColumnID", func(t *testing.T) { diff --git a/field_internal_test.go b/field_internal_test.go index da1ba40bd..98dea321b 100644 --- a/field_internal_test.go +++ b/field_internal_test.go @@ -247,7 +247,6 @@ func NewTestField(t testing.TB, opts FieldOption) *TestField { } cfg := DefaultHolderConfig() - cfg.StorageConfig.Backend = CurrentBackendOrDefault() cfg.StorageConfig.FsyncEnabled = false cfg.RBFConfig.FsyncEnabled = false h := NewHolder(path, cfg) diff --git a/fragment.go b/fragment.go index 093f6c763..2997264ba 100644 --- a/fragment.go +++ b/fragment.go @@ -3,7 +3,6 @@ package pilosa import ( "archive/tar" - "bufio" "bytes" "container/heap" "context" @@ -16,7 +15,6 @@ import ( "math/bits" "os" "path/filepath" - "runtime/debug" "sort" "strconv" "strings" @@ -59,18 +57,12 @@ const ( // width of roaring containers is 2^16 containerWidth = 1 << 16 - // snapshotExt is the file extension used for an in-process snapshot. - snapshotExt = ".snapshotting" - // cacheExt is the file extension for persisted cache ids. cacheExt = ".cache" // HashBlockSize is the number of rows in a merkle hash block. HashBlockSize = 100 - // defaultFragmentMaxOpN is the default value for Fragment.MaxOpN. - defaultFragmentMaxOpN = 10000 - // Row ids used for boolean fields. falseRowID = uint64(0) trueRowID = uint64(1) @@ -132,22 +124,11 @@ type fragment struct { // idx cached to avoid repeatedly looking it up everywhere. idx *Index - // parent holder, used to find snapshot queue, etc. + // parent holder holder *Holder - // debugging tool: addresses of current and previous maps - prevdata, currdata struct{ from, to uintptr } - // File-backed storage - flags byte // user-defined flags passed to roaring - 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 - snapshotPending bool // set to true when requesting a snapshot, set to false after snapshot completes - snapshotCond sync.Cond - snapshotErr error // error yielded by the last snapshot operation - snapshotStamp time.Time // timestamp of last snapshot - open bool // is this fragment actually open? + storage *roaring.Bitmap // Cache for row counts. CacheType string // passed in by field @@ -164,11 +145,6 @@ type fragment struct { // Cached checksums for each block. checksums map[int][]byte - // Number of operations performed before performing a snapshot. - // This limits the size of fragments on the heap and flushes them to disk - // so that they can be mmapped and heap utilization can be kept low. - MaxOpN int - // Logger used for out-of-band log entries. Logger logger.Logger @@ -194,18 +170,15 @@ func newFragment(holder *Holder, spec fragSpec, shard uint64, flags byte) *fragm fieldstr: spec.fieldstr, fld: spec.field, shard: shard, - flags: flags, idx: idx, CacheType: DefaultCacheType, CacheSize: DefaultCacheSize, holder: holder, - MaxOpN: defaultFragmentMaxOpN, stats: stats.NopStatsClient, } - f.snapshotCond = sync.Cond{L: &f.mu} return f } @@ -259,12 +232,6 @@ func (f *fragment) Open() error { defer f.mu.Unlock() if err := func() error { - // Initialize storage in a function so we can close if anything goes wrong. - f.holder.Logger.Debugf("open storage for index/field/view/fragment: %s/%s/%s/%d", f.index(), f.field(), f.view(), f.shard) - if err := f.openStorage(true); err != nil { - return errors.Wrap(err, "opening storage") - } - // Fill cache with rows persisted to disk. f.holder.Logger.Debugf("open cache for index/field/view/fragment: %s/%s/%s/%d", f.index(), f.field(), f.view(), f.shard) if err := f.openCache(); err != nil { @@ -278,24 +245,12 @@ func (f *fragment) Open() error { f.close() return err } - f.open = true _ = testhook.Opened(f.holder.Auditor, f, nil) f.holder.Logger.Debugf("successfully opened index/field/view/fragment: %s/%s/%s/%d", f.index(), f.field(), f.view(), f.shard) return nil } -// 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.currdata = struct{ from, to uintptr }{} - f.prevdata = f.currdata - return nil // openStorage becomes a noop under RBF, Badger, etc. - } - - return nil -} - // openCache initializes the cache from row ids persisted to disk. func (f *fragment) openCache() error { // Determine cache type from field name. @@ -351,12 +306,6 @@ func (f *fragment) Close() error { defer func() { _ = testhook.Closed(f.holder.Auditor, f, nil) }() - for f.snapshotPending { - f.snapshotCond.Wait() - } - // Note: snapshots won't progress on a closed fragment, so we - // wait until after a possible pending snapshot to close. - f.open = false return f.close() } @@ -367,28 +316,12 @@ func (f *fragment) close() error { return errors.Wrap(err, "flushing cache") } - // Close underlying storage. - if err := f.closeStorage(); err != nil { - f.holder.Logger.Errorf("fragment: error closing storage: err=%s, path=%s", err, f.path()) - return errors.Wrap(err, "closing storage") - } - // Remove checksums. f.checksums = nil return nil } -// 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 - - return nil -} - // mutexCheck checks for any entries in fragment which violate the mutex // property of having only one value set for a given column ID. func (f *fragment) mutexCheck(tx Tx, details bool, limit int) (map[uint64][]uint64, error) { @@ -511,9 +444,6 @@ func (f *fragment) unprotectedSetBit(tx Tx, rowID, columnID uint64) (changed boo // Invalidate block checksum. delete(f.checksums, int(rowID/HashBlockSize)) - // Increment number of operations until snapshot is required. - f.incrementOpN(1) - // If we're using a cache, update it. Otherwise skip the // possibly-expensive count operation. if f.CacheType != CacheTypeNone { @@ -563,9 +493,6 @@ func (f *fragment) unprotectedClearBit(tx Tx, rowID, columnID uint64) (changed b // Invalidate block checksum. delete(f.checksums, int(rowID/HashBlockSize)) - // Increment number of operations until snapshot is required. - f.incrementOpN(1) - // If we're using a cache, update it. Otherwise skip the // possibly-expensive count operation. if f.CacheType != CacheTypeNone { @@ -632,8 +559,6 @@ func (f *fragment) unprotectedSetRow(tx Tx, row *Row, rowID uint64) (changed boo } } - // Snapshot storage. - f.holder.SnapshotQueue.Enqueue(f) f.stats.Count("setRow", 1, 1.0) return changed, nil @@ -672,9 +597,6 @@ func (f *fragment) unprotectedClearRow(tx Tx, rowID uint64) (changed bool, err e // Clear the row in cache. f.cache.Add(rowID, 0) - // Snapshot storage. - f.holder.SnapshotQueue.Enqueue(f) - return changed, nil } @@ -1929,7 +1851,7 @@ func (f *fragment) mergeBlock(tx Tx, id int, data []pairSet) (sets, clears []pai return sets[1:], clears[1:], err } -// bulkImport bulk imports a set of bits and then snapshots the storage. +// bulkImport bulk imports a set of bits. // The cache is updated to reflect the new data. func (f *fragment) bulkImport(tx Tx, rowIDs, columnIDs []uint64, options *ImportOptions) error { // Verify that there are an equal number of row ids and column ids. @@ -2142,68 +2064,48 @@ func (p parallelSlices) Swap(i, j int) { // snapshot of the fragment or just do in-memory updates while appending // operations to the op log. func (f *fragment) importPositions(tx Tx, set, clear []uint64, rowSet map[uint64]struct{}) error { - //tx.AddN() - doFunc := func() error { - if len(set) > 0 { - f.stats.Count(MetricImportingN, int64(len(set)), 1) + if len(set) > 0 { + f.stats.Count(MetricImportingN, int64(len(set)), 1) - // TODO benchmark Add/RemoveN behavior with sorted/unsorted positions - changedN, err := tx.Add(f.index(), f.field(), f.view(), f.shard, set...) - if err != nil { - return errors.Wrap(err, "adding positions") - } - f.stats.Count(MetricImportedN, int64(changedN), 1) - f.incrementOpN(changedN) + // TODO benchmark Add/RemoveN behavior with sorted/unsorted positions + changedN, err := tx.Add(f.index(), f.field(), f.view(), f.shard, set...) + if err != nil { + return errors.Wrap(err, "adding positions") } + f.stats.Count(MetricImportedN, int64(changedN), 1) + } - if len(clear) > 0 { - f.stats.Count(MetricClearingN, int64(len(clear)), 1) - changedN, err := tx.Remove(f.index(), f.field(), f.view(), f.shard, clear...) - if err != nil { - return errors.Wrap(err, "clearing positions") - } - f.stats.Count(MetricClearedN, int64(changedN), 1) - f.incrementOpN(changedN) + if len(clear) > 0 { + f.stats.Count(MetricClearingN, int64(len(clear)), 1) + changedN, err := tx.Remove(f.index(), f.field(), f.view(), f.shard, clear...) + if err != nil { + return errors.Wrap(err, "clearing positions") } + f.stats.Count(MetricClearedN, int64(changedN), 1) + } - // Update cache counts for all affected rows. - for rowID := range rowSet { - // Invalidate block checksum. - delete(f.checksums, int(rowID/HashBlockSize)) - - if f.CacheType != CacheTypeNone { - start := rowID * ShardWidth - end := (rowID + 1) * ShardWidth - - n, err := tx.CountRange(f.index(), f.field(), f.view(), f.shard, start, end) - if err != nil { - return errors.Wrap(err, "CountRange") - } - - f.cache.BulkAdd(rowID, n) - } - } + // Update cache counts for all affected rows. + for rowID := range rowSet { + // Invalidate block checksum. + delete(f.checksums, int(rowID/HashBlockSize)) if f.CacheType != CacheTypeNone { - f.cache.Invalidate() - } - return nil - } - 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) - if errs != 0 { - f.holder.Logger.Errorf("transaction failed on %s. storage has %d mapped in range, %d mapped out of range, %d unmapped in range, %d errors total, last %v", - f.path(), mappedIn, mappedOut, unmappedIn, errs, e2) - if f.prevdata.from != f.currdata.from { - mappedIn, mappedOut, unmappedIn, errs, e2 = f.storage.SanityCheckMapping(f.prevdata.from, f.prevdata.to) - f.holder.Logger.Errorf("with previous map, storage would have %d mapped in range, %d mapped out of range, %d unmapped in range, %d errors total, last %v", - mappedIn, mappedOut, unmappedIn, errs, e2) + start := rowID * ShardWidth + end := (rowID + 1) * ShardWidth + + n, err := tx.CountRange(f.index(), f.field(), f.view(), f.shard, start, end) + if err != nil { + return errors.Wrap(err, "CountRange") } + + f.cache.BulkAdd(rowID, n) } } - return err + + if f.CacheType != CacheTypeNone { + f.cache.Invalidate() + } + return nil } // sliceDifference removes everything from original that's found in remove, @@ -2503,125 +2405,6 @@ func (f *fragment) importRoaringOverwrite(ctx context.Context, tx Tx, data []byt return f.importRoaring(ctx, tx, data, false) } -// incrementOpN increase the operation count by one. -// If the count exceeds the maximum allowed then a snapshot is performed. -func (f *fragment) incrementOpN(changed int) { - if changed <= 0 { - return - } - // don't count opN or ops if our index doesn't want snapshots - if !f.idx.NeedsSnapshot() { - return - } - f.opN += changed - f.ops++ - if f.opN > f.MaxOpN { - f.holder.SnapshotQueue.Enqueue(f) - } -} - -// Snapshot writes the storage bitmap to disk and reopens it. This may -// coexist with existing background-queue snapshotting; it does not remove -// things from the queue. You probably don't want to do this; use -// the snapshotQueue's Enqueue/Await. -func (f *fragment) Snapshot() error { - f.mu.Lock() - defer f.mu.Unlock() - return f.snapshot() -} - -func track(start time.Time, message string, stats stats.StatsClient, logger logger.Logger) { - elapsed := time.Since(start) - logger.Debugf("%s took %s", message, elapsed) - stats.Timing(MetricSnapshotDurationSeconds, elapsed, 1.0) -} - -// snapshot does the actual snapshot operation. it does not check or care -// about f.snapshotPending. -func (f *fragment) snapshot() (err error) { - if !f.idx.NeedsSnapshot() { - return nil - } - if !f.open { - return errors.New("snapshot request on closed fragment") - } - wouldPanic := debug.SetPanicOnFault(true) - defer func() { - debug.SetPanicOnFault(wouldPanic) - if r := recover(); r != nil { - if e2, ok := r.(error); ok { - err = e2 - // 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 e2.Error() == "runtime error: invalid memory address or nil pointer dereference" { - mappedIn, mappedOut, unmappedIn, errs, _ := f.storage.SanityCheckMapping(f.currdata.from, f.currdata.to) - f.holder.Logger.Errorf("transaction failed on %s. storage has %d mapped in range, %d mapped out of range, %d unmapped in range, %d errors total", - f.path(), mappedIn, mappedOut, unmappedIn, errs) - } - } else { - err = fmt.Errorf("non-error PanicOn: %v", r) - } - } - }() - _, err = unprotectedWriteToFragment(f, f.storage) - if err == nil { - f.snapshotStamp = time.Now() - } - return err -} - -// unprotectedWriteToFragment writes the fragment f with bm as the data. It is unprotected, and -// f.mu must be locked when calling it. -func unprotectedWriteToFragment(f *fragment, bm *roaring.Bitmap) (n int64, err error) { // nolint: interfacer - completeMessage := fmt.Sprintf("fragment: snapshot complete %s/%s/%s/%d", f.index(), f.field(), f.view(), f.shard) - start := time.Now() - defer track(start, completeMessage, f.stats, f.holder.Logger) - - // Create a temporary file to snapshot to. - snapshotPath := f.path() + snapshotExt - file, err := os.Create(snapshotPath) - if err != nil { - return n, fmt.Errorf("create snapshot file: %s", err) - } - // No deferred close, because we want to close it sooner than the - // end of this function. - - // Write storage to snapshot. - bw := bufio.NewWriter(file) - if n, err = bm.WriteTo(bw); err != nil { - file.Close() - return n, fmt.Errorf("snapshot write to: %s", err) - } - - if err := bw.Flush(); err != nil { - file.Close() - return n, fmt.Errorf("flush: %s", err) - } - - // we close the file here so we don't still have it open when trying - // to open it in a moment. - file.Close() - - // Move snapshot to data file location. - if err := os.Rename(snapshotPath, f.path()); err != nil { - return n, fmt.Errorf("rename snapshot: %s", err) - } - - // if we reloaded from the file, we'd end up with this bitmap - // as our storage. so... let's use this bitmap. as our storage. - f.storage = bm - - // Reopen storage. - if err := f.openStorage(false); err != nil { - return n, fmt.Errorf("open storage: %s", err) - } - - // Reset operation count. - f.opN = 0 - - return n, nil -} - // RecalculateCache rebuilds the cache regardless of invalidate time delay. func (f *fragment) RecalculateCache() { f.mu.Lock() diff --git a/fragment_internal_test.go b/fragment_internal_test.go index f8b1ec526..8953e113d 100644 --- a/fragment_internal_test.go +++ b/fragment_internal_test.go @@ -8,13 +8,9 @@ import ( "fmt" "io" "io/ioutil" - "math" "math/rand" "os" - "path/filepath" "reflect" - "runtime" - "runtime/debug" "sort" "strconv" "strings" @@ -25,7 +21,6 @@ import ( "github.com/davecgh/go-spew/spew" "github.com/molecula/featurebase/v3/pql" "github.com/molecula/featurebase/v3/roaring" - "github.com/molecula/featurebase/v3/storage" "github.com/molecula/featurebase/v3/testhook" . "github.com/molecula/featurebase/v3/vprint" // nolint:staticcheck "github.com/pkg/errors" @@ -1025,50 +1020,44 @@ func BenchmarkFragment_ImportValue(b *testing.B) { // // We test a variety of combinations of the number of separate updates(imports), // the number of bits in the import, the number of rows in the fragment (which -// is a pretty good proxy for fragment size on disk), and the MaxOpN on the -// fragment which controls how many set bits occur before a snapshot is done. If -// the number of bits in a given import is greater than MaxOpN, bulkImport will -// always go through the standard snapshotting import path. +// is a pretty good proxy for fragment size on disk). func BenchmarkFragment_RepeatedSmallImports(b *testing.B) { for _, numUpdates := range []int{100} { for _, bitsPerUpdate := range []int{100, 1000} { for _, numRows := range []int{1000, 100000, 1000000} { - for _, opN := range []int{1, 5000, 50000} { - b.Run(fmt.Sprintf("Rows%dUpdates%dBits%dOpN%d", numRows, numUpdates, bitsPerUpdate, opN), func(b *testing.B) { - for a := 0; a < b.N; a++ { - b.StopTimer() - // build the update data set all at once - this will get applied - // to a fragment in numUpdates batches - updateRows := make([]uint64, numUpdates*bitsPerUpdate) - updateCols := make([]uint64, numUpdates*bitsPerUpdate) - for i := 0; i < numUpdates*bitsPerUpdate; i++ { - updateRows[i] = uint64(rand.Int63n(int64(numRows))) // row id - updateCols[i] = uint64(rand.Int63n(ShardWidth)) // column id - } - f, idx, tx := mustOpenFragment(b, "i", "f", viewStandard, 0, "") - _ = idx - f.MaxOpN = opN - defer f.Clean(b) - - err := f.importRoaringT(tx, getZipfRowsSliceRoaring(uint64(numRows), 1, 0, ShardWidth), false) - if err != nil { - b.Fatalf("importing base data for benchmark: %v", err) - } - b.StartTimer() - for i := 0; i < numUpdates; i++ { - err := f.bulkImportStandard(tx, - updateRows[bitsPerUpdate*i:bitsPerUpdate*(i+1)], - updateRows[bitsPerUpdate*i:bitsPerUpdate*(i+1)], - &ImportOptions{}, - ) - if err != nil { - b.Fatalf("doing small bulk import: %v", err) - } - } - tx.Rollback() // don't exhaust the Tx space under b.N iterations. + b.Run(fmt.Sprintf("Rows%dUpdates%dBits%d", numRows, numUpdates, bitsPerUpdate), func(b *testing.B) { + for a := 0; a < b.N; a++ { + b.StopTimer() + // build the update data set all at once - this will get applied + // to a fragment in numUpdates batches + updateRows := make([]uint64, numUpdates*bitsPerUpdate) + updateCols := make([]uint64, numUpdates*bitsPerUpdate) + for i := 0; i < numUpdates*bitsPerUpdate; i++ { + updateRows[i] = uint64(rand.Int63n(int64(numRows))) // row id + updateCols[i] = uint64(rand.Int63n(ShardWidth)) // column id } - }) - } + f, idx, tx := mustOpenFragment(b, "i", "f", viewStandard, 0, "") + _ = idx + defer f.Clean(b) + + err := f.importRoaringT(tx, getZipfRowsSliceRoaring(uint64(numRows), 1, 0, ShardWidth), false) + if err != nil { + b.Fatalf("importing base data for benchmark: %v", err) + } + b.StartTimer() + for i := 0; i < numUpdates; i++ { + err := f.bulkImportStandard(tx, + updateRows[bitsPerUpdate*i:bitsPerUpdate*(i+1)], + updateRows[bitsPerUpdate*i:bitsPerUpdate*(i+1)], + &ImportOptions{}, + ) + if err != nil { + b.Fatalf("doing small bulk import: %v", err) + } + } + tx.Rollback() // don't exhaust the Tx space under b.N iterations. + } + }) } } } @@ -1078,33 +1067,30 @@ func BenchmarkFragment_RepeatedSmallImportsRoaring(b *testing.B) { for _, numUpdates := range []int{100} { for _, bitsPerUpdate := range []uint64{100, 1000} { for _, numRows := range []uint64{1000, 100000, 1000000} { - for _, opN := range []int{1, 5000, 50000} { - b.Run(fmt.Sprintf("Rows%dUpdates%dBits%dOpN%d", numRows, numUpdates, bitsPerUpdate, opN), func(b *testing.B) { - for a := 0; a < b.N; a++ { - b.StopTimer() - // build the update data set all at once - this will get applied - // to a fragment in numUpdates batches - f, idx, tx := mustOpenFragment(b, "i", "f", viewStandard, 0, "") - _ = idx - f.MaxOpN = opN - defer f.Clean(b) + b.Run(fmt.Sprintf("Rows%dUpdates%dBits%d", numRows, numUpdates, bitsPerUpdate), func(b *testing.B) { + for a := 0; a < b.N; a++ { + b.StopTimer() + // build the update data set all at once - this will get applied + // to a fragment in numUpdates batches + f, idx, tx := mustOpenFragment(b, "i", "f", viewStandard, 0, "") + _ = idx + defer f.Clean(b) - err := f.importRoaringT(tx, getZipfRowsSliceRoaring(numRows, 1, 0, ShardWidth), false) + err := f.importRoaringT(tx, getZipfRowsSliceRoaring(numRows, 1, 0, ShardWidth), false) + if err != nil { + b.Fatalf("importing base data for benchmark: %v", err) + } + for i := 0; i < numUpdates; i++ { + data := getUpdataRoaring(numRows, bitsPerUpdate, int64(i)) + b.StartTimer() + err := f.importRoaringT(tx, data, false) + b.StopTimer() if err != nil { - b.Fatalf("importing base data for benchmark: %v", err) - } - for i := 0; i < numUpdates; i++ { - data := getUpdataRoaring(numRows, bitsPerUpdate, int64(i)) - b.StartTimer() - err := f.importRoaringT(tx, data, false) - b.StopTimer() - if err != nil { - b.Fatalf("doing small roaring import: %v", err) - } + b.Fatalf("doing small roaring import: %v", err) } } - }) - } + } + }) } } } @@ -1131,70 +1117,34 @@ func BenchmarkFragment_RepeatedSmallValueImports(b *testing.B) { updateVals[i] = int64(rand.Int63n(1 << 21)) } - for _, opN := range []int{1, 5000, 50000} { - b.Run(fmt.Sprintf("Updates%dVals%dOpN%d", numUpdates, valsPerUpdate, opN), func(b *testing.B) { - for i := 0; i < b.N; i++ { - b.StopTimer() - f, _, tx := mustOpenBSIFragment(b, "i", "f", viewBSIGroupPrefix+"foo", 0) - f.MaxOpN = opN + b.Run(fmt.Sprintf("Updates%dVals%d", numUpdates, valsPerUpdate), func(b *testing.B) { + for i := 0; i < b.N; i++ { + b.StopTimer() + f, _, tx := mustOpenBSIFragment(b, "i", "f", viewBSIGroupPrefix+"foo", 0) - err := f.importValue(tx, initialCols, initialVals, 21, false) - if err != nil { - b.Fatalf("initial value import: %v", err) - } - b.StartTimer() - for j := 0; j < numUpdates; j++ { - err := f.importValue(tx, - updateCols[valsPerUpdate*j:valsPerUpdate*(j+1)], - updateVals[valsPerUpdate*j:valsPerUpdate*(j+1)], - 21, - false, - ) - if err != nil { - b.Fatalf("importing values: %v", err) - } - } - tx.Rollback() // don't exhaust the Tx over the b.N iterations. + err := f.importValue(tx, initialCols, initialVals, 21, false) + if err != nil { + b.Fatalf("initial value import: %v", err) } - }) - } - + b.StartTimer() + for j := 0; j < numUpdates; j++ { + err := f.importValue(tx, + updateCols[valsPerUpdate*j:valsPerUpdate*(j+1)], + updateVals[valsPerUpdate*j:valsPerUpdate*(j+1)], + 21, + false, + ) + if err != nil { + b.Fatalf("importing values: %v", err) + } + } + tx.Rollback() // don't exhaust the Tx over the b.N iterations. + } + }) } } } -// Ensure a fragment can snapshot correctly. -func TestFragment_Snapshot(t *testing.T) { - f, idx, tx := mustOpenFragment(t, "i", "f", viewStandard, 0, "") - defer f.Clean(t) - - // Set and then clear bits on the fragment. - if _, err := f.setBit(tx, 1000, 1); err != nil { - t.Fatal(err) - } else if _, err := f.setBit(tx, 1000, 2); err != nil { - t.Fatal(err) - } else if _, err := f.clearBit(tx, 1000, 1); err != nil { - t.Fatal(err) - } - PanicOn(tx.Commit()) - tx = idx.holder.txf.NewTx(Txo{Write: !writable, Index: idx, Fragment: f, Shard: f.shard}) - defer tx.Rollback() - - // Snapshot bitmap and verify data. - if err := f.Snapshot(); err != nil { - t.Fatal(err) - } else if n := f.mustRow(tx, 1000).Count(); n != 1 { - t.Fatalf("unexpected count: %d", n) - } - - // Close and reopen the fragment & verify the data. - if err := f.Reopen(); err != nil { - t.Fatal(err) - } else if n := f.mustRow(tx, 1000).Count(); n != 1 { - t.Fatalf("unexpected count (reopen): %d", n) - } -} - // Ensure a fragment can iterate over all bits in order. func TestFragment_ForEachBit(t *testing.T) { f, idx, tx := mustOpenFragment(t, "i", "f", viewStandard, 0, "") @@ -1577,91 +1527,8 @@ func TestFragment_LRUCache_Persistence(t *testing.T) { } } -// Ensure a fragment's cache can be persisted between restarts. -func TestFragment_RankCache_Persistence(t *testing.T) { - roaringOnlyTest(t) - - index := mustOpenIndex(t, IndexOptions{}) - defer index.Close() - - // Create field. - field, err := index.CreateFieldIfNotExists("f", OptFieldTypeSet(CacheTypeRanked, DefaultCacheSize)) - if err != nil { - t.Fatal(err) - } - - // Create view. - view, err := field.createViewIfNotExists(viewStandard) - if err != nil { - t.Fatal(err) - } - - // Create fragment. - f, err := view.CreateFragmentIfNotExists(0) - if err != nil { - t.Fatal(err) - } - - // Obtain transaction. - tx := index.holder.txf.NewTx(Txo{Write: writable, Index: index, Fragment: f, Shard: f.shard}) - defer tx.Rollback() - - // Set bits on the fragment. - for i := uint64(0); i < 1000; i++ { - if _, err := f.setBit(tx, i, 0); err != nil { - t.Fatal(err) - } - } - - PanicOn(tx.Commit()) - tx = index.holder.txf.NewTx(Txo{Write: !writable, Index: index, Fragment: f, Shard: f.shard}) - defer tx.Rollback() - - // Verify correct cache type and size. - if cache, ok := f.cache.(*rankCache); !ok { - t.Fatalf("unexpected cache: %T", f.cache) - } else if cache.Len() != 1000 { - t.Fatalf("unexpected cache len: %d", cache.Len()) - } - - // Reopen the index. - if err := index.reopen(); err != nil { - t.Fatal(err) - } - - // Re-fetch fragment. - f = index.Field("f").view(viewStandard).Fragment(0) - - // Re-verify correct cache type and size. - if cache, ok := f.cache.(*rankCache); !ok { - t.Fatalf("unexpected cache: %T", f.cache) - } else if cache.Len() != 1000 { - t.Fatalf("unexpected cache len: %d", cache.Len()) - } -} - -func roaringOnlyTest(t *testing.T) { - src := CurrentBackend() - if src == RoaringTxn || (storage.DefaultBackend == RoaringTxn && src == "") { - // okay to run, we are under roaring only - } else { - t.Skip("skip for everything but roaring") - } -} - -func roaringOnlyBenchmark(b *testing.B) { - src := CurrentBackend() - if src == RoaringTxn || (storage.DefaultBackend == RoaringTxn && src == "") { - // okay to run, we are under roaring only - } else { - b.Skip("skip for everything but roaring") - } -} - // Ensure a fragment can be copied to another fragment. func TestFragment_WriteTo_ReadFrom(t *testing.T) { - // roaringOnlyTest(t) - f0, _, tx := mustOpenFragment(t, "i", "f", viewStandard, 0, "") defer f0.Clean(t) @@ -1752,7 +1619,6 @@ func BenchmarkFragment_Blocks(b *testing.B) { func BenchmarkFragment_IntersectionCount(b *testing.B) { f, idx, tx := mustOpenFragment(b, "i", "f", viewStandard, 0, "") defer f.Clean(b) - f.MaxOpN = math.MaxInt32 // Generate some intersecting data. for i := 0; i < 10000; i += 2 { @@ -1770,11 +1636,6 @@ func BenchmarkFragment_IntersectionCount(b *testing.B) { tx = idx.holder.txf.NewTx(Txo{Write: !writable, Index: idx, Fragment: f, Shard: f.shard}) defer tx.Rollback() - // Snapshot to disk before benchmarking. - if err := f.Snapshot(); err != nil { - b.Fatal(err) - } - // Start benchmark b.ResetTimer() for i := 0; i < b.N; i++ { @@ -1834,35 +1695,6 @@ func TestFragment_Zero_Tanimoto(t *testing.T) { } } -func TestFragment_Snapshot_Run(t *testing.T) { - roaringOnlyTest(t) - - f, idx, tx := mustOpenFragment(t, "i", "f", viewStandard, 0, "") - _ = idx - defer f.Clean(t) - - // Set bits on the fragment. - for i := uint64(1); i < 3; i++ { - if _, err := f.setBit(tx, 1000, i); err != nil { - t.Fatal(err) - } - } - - // Snapshot bitmap and verify data. - if err := f.Snapshot(); err != nil { - t.Fatal(err) - } else if n := f.mustRow(tx, 1000).Count(); n != 2 { - t.Fatalf("unexpected count: %d", n) - } - - // Close and reopen the fragment & verify the data. - if err := f.Reopen(); err != nil { - t.Fatal(err) - } else if n := f.mustRow(tx, 1000).Count(); n != 2 { - t.Fatalf("unexpected count (reopen): %d", n) - } -} - // Ensure a fragment can set mutually exclusive values. func TestFragment_SetMutex(t *testing.T) { f, _, tx := mustOpenMutexFragment(t, "i", "f", viewStandard, 0, "") @@ -2684,77 +2516,6 @@ func makeTestFragSpec(path, index, field, view0 string) fragSpec { } } -func BenchmarkFragment_Snapshot(b *testing.B) { - if *FragmentPath == "" { - b.Skip("no fragment specified") - } - - b.ReportAllocs() - // Open the fragment specified by the path. - f := newFragment(newTestHolder(b), makeTestFragSpec(*FragmentPath, "i", "f", viewStandard), 0, 0) - if err := f.Open(); err != nil { - b.Fatal(err) - } - defer f.Clean(b) - b.ResetTimer() - - // Reset timer and execute benchmark. - b.ResetTimer() - b.ReportAllocs() - for i := 0; i < b.N; i++ { - err := f.Snapshot() - if err != nil { - b.Fatalf("unexpected count (reopen): %s", err) - } - } -} - -func BenchmarkFragment_FullSnapshot(b *testing.B) { - f, idx, tx := mustOpenFragment(b, "i", "f", viewStandard, 0, "") - _ = idx - tx.Rollback() - defer f.Clean(b) - - // Generate some intersecting data. - maxX := ShardWidth / 2 - sz := maxX - rows := make([]uint64, sz) - cols := make([]uint64, sz) - - options := &ImportOptions{} - max := 0 - for row := 0; row < 100; row++ { - val := 1 - i := 0 - for col := 0; col < ShardWidth/2; col++ { - rows[i] = uint64(row) - cols[i] = uint64(val) - val += 2 - i++ - } - - tx := idx.holder.txf.NewTx(Txo{Write: !writable, Index: idx, Fragment: f, Shard: f.shard}) - defer tx.Rollback() - - if err := f.bulkImport(tx, rows, cols, options); err != nil { - b.Fatalf("Error Building Sample: %s", err) - } - tx.Rollback() - if row > max { - max = row - } - } - - b.ResetTimer() - b.ReportAllocs() - - for i := 0; i < b.N; i++ { - if err := f.Snapshot(); err != nil { - b.Fatal(err) - } - } -} - func BenchmarkFragment_Import(b *testing.B) { b.StopTimer() maxX := ShardWidth * 5 * 2 @@ -2813,11 +2574,6 @@ func BenchmarkImportRoaring(b *testing.B) { err := f.importRoaringT(tx, data, false) if err != nil { - // we don't actually particularly - // care whether this succeeds, - // but if it's happening we want - // it to be done. - _ = f.holder.SnapshotQueue.Await(f) f.Clean(b) b.Fatalf("import error: %v", err) } @@ -2859,9 +2615,6 @@ func BenchmarkImportRoaringConcurrent(b *testing.B) { defer txs[j].Rollback() err := frags[j].importRoaringT(txs[j], data[j], false) - // error unimportant if it happened, but we want - // any snapshots to have finished. - _ = frags[j].holder.SnapshotQueue.Await(frags[j]) return err }) } @@ -2879,68 +2632,6 @@ func BenchmarkImportRoaringConcurrent(b *testing.B) { } } } -func BenchmarkImportRoaringUpdateConcurrent(b *testing.B) { - roaringOnlyBenchmark(b) - if testing.Short() { - b.SkipNow() - } - for _, numRows := range rowCases { - for _, numCols := range colCases { - data := getZipfRowsSliceRoaring(numRows, 1, 0, ShardWidth) - updata := getUpdataRoaring(numRows, numCols, 1) - for _, concurrency := range concurrencyCases { - for _, cacheType := range cacheCases { - b.Run(fmt.Sprintf("Rows%dCols%dConcurrency%dCache_%s", numRows, numCols, concurrency, cacheType), func(b *testing.B) { - b.StopTimer() - frags := make([]*fragment, concurrency) - txs := make([]Tx, concurrency) - for i := 0; i < b.N; i++ { - for j := 0; j < concurrency; j++ { - frags[j], _, txs[j] = mustOpenFragment(b, "i", "f", viewStandard, uint64(j), cacheType) - - // the cost of actually doing the op log for the large initial data set - // is excessive. force storage into snapshotted state, then use import - // to generate an op log and/or snapshot. - // note: skipped for rbf, bolt, lmdb, above. - _, _, err := frags[j].storage.ImportRoaringBits(data, false, false, 0) - if err != nil { - b.Fatalf("importing roaring: %v", err) - } - err = frags[j].holder.SnapshotQueue.Immediate(frags[j]) - if err != nil { - b.Fatalf("snapshot after import: %v", err) - } - } - eg := errgroup.Group{} - b.StartTimer() - for j := 0; j < concurrency; j++ { - j := j - eg.Go(func() error { - defer txs[j].Rollback() - - err := frags[j].importRoaringT(txs[j], updata, false) - err2 := frags[j].holder.SnapshotQueue.Await(frags[j]) - if err == nil { - err = err2 - } - return err - }) - } - err := eg.Wait() - if err != nil { - b.Errorf("importing fragment: %v", err) - } - b.StopTimer() - for j := 0; j < concurrency; j++ { - frags[j].Clean(b) - } - } - }) - } - } - } - } -} func BenchmarkImportStandard(b *testing.B) { for _, cacheType := range cacheCases { @@ -2984,29 +2675,18 @@ func BenchmarkImportRoaringUpdate(b *testing.B) { f, idx, tx := mustOpenFragment(b, "i", fmt.Sprintf("r%dc%dcache_%s", numRows, numCols, cacheType), viewStandard, 0, cacheType) _ = idx - // the cost of actually doing the op log for the large initial data set - // is excessive. force storage into snapshotted state, then use import - // to generate an op log and/or snapshot. itr, err := roaring.NewRoaringIterator(data) PanicOn(err) _, _, err = tx.ImportRoaringBits(f.index(), f.field(), f.view(), f.shard, itr, false, false, 0) if err != nil { b.Errorf("import error: %v", err) } - err = f.holder.SnapshotQueue.Immediate(f) - if err != nil { - b.Errorf("snapshot after import error: %v", err) - } b.StartTimer() err = f.importRoaringT(tx, updata, false) if err != nil { f.Clean(b) b.Errorf("import error: %v", err) } - err = f.holder.SnapshotQueue.Await(f) - if err != nil { - b.Errorf("snapshot after import error: %v", err) - } b.StopTimer() var stat os.FileInfo var statTarget io.Writer @@ -3160,9 +2840,6 @@ func BenchmarkImportRoaringIntoLargeFragment(b *testing.B) { //nf, idx, tx := mustOpenFragmentFlags(index, field, view string, shard uint64, cacheType string, flags byte) idx := fragTestMustOpenIndex("i", th, IndexOptions{}) - if th.NeedsSnapshot() { - th.SnapshotQueue = newSnapshotQueue(1, 1, nil) - } // XXX TODO: newFragment is using the wrong path here, we should fix that someday. f := newFragment(th, makeTestFragSpec(fi.Name(), "i", "f", viewStandard), 0, 0) defer f.Clean(b) @@ -3433,31 +3110,6 @@ func BenchmarkFileWrite(b *testing.B) { } -///////////////////////////////////////////////////////////////////// - -// not called under Tx stores b/c f.idx.NeedsSnapshot() in Clean() avoids it. -func (f *fragment) sanityCheck(t testing.TB) { - newBM := roaring.NewFileBitmap() - file, err := os.Open(f.path()) - if err != nil { - t.Fatalf("sanityCheck couldn't open file %s: %v", f.path(), err) - } - defer file.Close() - data, err := ioutil.ReadAll(file) - if err != nil { - t.Fatalf("sanityCheck couldn't read fragment %s: %v", f.path(), err) - } - err = newBM.UnmarshalBinary(data) - if err != nil { - t.Fatalf("sanityCheck couldn't unmarshal fragment %s: %v", f.path(), err) - } - // Refactor fragment.storage - // note: not called for rbf, see above. - if equal, reason := newBM.BitwiseEqual(f.storage); !equal { - t.Fatalf("fragment %s: unmarshalled bitmap different: %v", f.path(), reason) - } -} - // Clean used to delete fragments, but doesn't anymore -- deleting is // handled by the testhook.TempDir when appropriate. // TODO(jaffee): this can likely go away entirely... it was doing snapshot/source/generation stuff that it no longer needs to. @@ -3490,7 +3142,7 @@ func newTestHolder(tb testing.TB) *Holder { testhook.Cleanup(tb, func() { h.Close() }) - //h.SnapshotQueue = newSnapshotQueue(1, 1, nil) + return h } @@ -3524,9 +3176,6 @@ func mustOpenFragmentFlags(tb testing.TB, index, field, view string, shard uint6 th := newTestHolder(tb) idx := fragTestMustOpenIndex(index, th, IndexOptions{}) - if th.NeedsSnapshot() { - th.SnapshotQueue = newSnapshotQueue(1, 1, nil) - } fragDir := fmt.Sprintf("%v/%v/views/%v/fragments/", idx.path, field, view) PanicOn(os.MkdirAll(fragDir, 0777)) @@ -4230,75 +3879,6 @@ func TestFragmentRowIterator_WithTxCommit(t *testing.T) { }) } -func TestUnionInPlaceMapped(t *testing.T) { - roaringOnlyTest(t) - - f, _, _ := mustOpenFragment(t, "i", "f", "v", 0, CacheTypeNone) - // note: clean has to be deferred first, because it has to run with - // the lock *not* held, because it is sometimes so it has to grab the - // lock... - defer f.Clean(t) - - f.mu.Lock() - defer f.mu.Unlock() - r0 := rand.New(rand.NewSource(2)) - r1 := rand.New(rand.NewSource(1)) - data0 := randPositions(1000000, r0) - setBM0 := roaring.NewBitmap() - setBM0.OpWriter = nil - _, err := setBM0.Add(data0...) - if err != nil { - t.Fatalf("adding bits: %v", err) - } - count0 := setBM0.Count() - - data1 := randPositions(1000000, r1) - setBM1 := roaring.NewBitmap() - setBM1.OpWriter = nil - _, err = setBM1.Add(data1...) - if err != nil { - t.Fatalf("adding bits: %v", err) - } - count1 := setBM1.Count() - - // now we write setBM0 into f.storage. - _, err = unprotectedWriteToFragment(f, setBM0) - if err != nil { - t.Fatalf("trying to flush fragment to disk: %v", err) - } - countF := f.storage.Count() - - f.storage.UnionInPlace(setBM1) - countUnion := f.storage.Count() - - // UnionInPlace produces no ops log, we have to make it snapshot, to - // ensure that the on-disk representation is correct. Note, UIP is - // not used for things that are modifying real fragments, usually; - // it's used only in computation of things that usually don't go to - // disk, which is why we handle this specially in testing and not - // generically. - err = f.holder.SnapshotQueue.Immediate(f) - if err != nil { - t.Fatalf("snapshot after union-in-place: %v", err) - } - - if count0 != countF { - t.Fatalf("writing bitmap to storage changed count: %d => %d", count0, countF) - } - min := count0 - if count1 > min { - min = count1 - } - max := count0 + count1 - // We don't know how many bits we should have, because of overlap, - // but it should be between the size of the largest bitmap and the - // sum of the bitmaps. - if countUnion < min || countUnion > max { - t.Fatalf("union of sets with cardinality %d and %d should be between %d and %d, got %d", - count0, count1, min, max, countUnion) - } -} - func randPositions(n int, r *rand.Rand) []uint64 { ret := make([]uint64, n) for i := 0; i < n; i++ { @@ -4927,218 +4507,12 @@ func TestFragmentBSISigned(t *testing.T) { }) } -func TestImportClearRestart(t *testing.T) { - roaringOnlyTest(t) - - tests := []struct { - rows []uint64 - cols []uint64 - }{ - { - rows: []uint64{1}, - cols: []uint64{1}, - }, - { - rows: []uint64{1, 2, 3, 4, 5, 6, 7, 8, 9, 1}, - cols: []uint64{1, 2, 3, 4, 5, 6, 7, 8, 9, 500000}, - }, - { - rows: []uint64{0, 0, 0, 0, 0, 0, 0, 0, 0, 0}, - cols: []uint64{0, 65535, 65536, 131071, 131072, 196607, 196608, 262143, 262144, 1000000}, - }, - { - rows: []uint64{1, 2, 20, 200, 2000, 200000}, - cols: []uint64{1, 1, 1, 1, 1, 1}, - }, - } - for i, test := range tests { - for _, maxOpN := range []int{0, 10000} { - t.Run(fmt.Sprintf("%dMaxOpN%d", i, maxOpN), func(t *testing.T) { - testrows, testcols := make([]uint64, len(test.rows)), make([]uint64, len(test.rows)) - copy(testrows, test.rows) - copy(testcols, test.cols) - exp := make(map[uint64]map[uint64]struct{}) // row num to cols - if len(testrows) != len(testcols) { - t.Fatalf("bad test spec-need same number of rows/cols, %d/%d", len(testrows), len(testcols)) - } - // set up expected data - expOpN := 0 - for i := range testrows { - row, col := testrows[i], testcols[i] - cols, ok := exp[row] - if !ok { - exp[row] = make(map[uint64]struct{}) - cols = exp[row] - } - if _, ok = cols[col]; !ok { - expOpN++ - cols[col] = struct{}{} - } - } - - f, idx, tx := mustOpenFragment(t, "i", "f", viewStandard, 0, "") - _ = idx - f.MaxOpN = maxOpN - - err := f.bulkImport(tx, testrows, testcols, &ImportOptions{}) - if err != nil { - t.Fatalf("initial small import: %v", err) - } - if idx.holder.txf.TxType() == RoaringTxn { - if expOpN <= maxOpN && f.opN != expOpN { - t.Errorf("unexpected opN - %d is not %d", f.opN, expOpN) - } - } - check(t, tx, f, exp) - - err = f.Close() - if err != nil { - t.Fatalf("closing fragment: %v", err) - } - PanicOn(tx.Commit()) - - err = f.Open() - tx = idx.holder.txf.NewTx(Txo{Write: writable, Index: idx, Fragment: f, Shard: f.shard}) - defer tx.Rollback() - if err != nil { - t.Fatalf("reopening fragment: %v", err) - } - - if idx.holder.txf.TxType() == RoaringTxn { - if expOpN <= maxOpN && f.opN != expOpN { - t.Errorf("unexpected opN after close/open %d is not %d", f.opN, expOpN) - } - } - - check(t, tx, f, exp) - - h := newTestHolder(t) - idx2, err := h.CreateIndex("i", IndexOptions{}) - _ = idx2 - PanicOn(err) - - // OVERWRITING the f.path with a new fragment - f2 := newFragment(h, makeTestFragSpec(f.path(), "i", "f", viewStandard), 0, 0) - f2.MaxOpN = maxOpN - f2.CacheType = f.CacheType - - PanicOn(tx.Commit()) // match the f.closeStorage which overlaps the f2 creation. - - tx2 := idx.holder.txf.NewTx(Txo{Write: writable, Index: idx, Fragment: f2, Shard: f2.shard}) - defer tx2.Rollback() - - err = f.Close() - if err != nil { - t.Fatalf("closing storage: %v", err) - } - - err = f2.Open() - if err != nil { - t.Fatalf("opening new fragment: %v", err) - } - - if idx.holder.txf.TxType() == RoaringTxn { - if expOpN <= maxOpN && f2.opN != expOpN { - t.Errorf("unexpected opN after close/open %d is not %d", f2.opN, expOpN) - } - } - - check(t, tx2, f2, exp) - - copy(testrows, test.rows) - copy(testcols, test.cols) - err = f2.bulkImport(tx2, testrows, testcols, &ImportOptions{Clear: true}) - if err != nil { - t.Fatalf("clearing imported data: %v", err) - } - - // clear exp, but leave rows in so we re-query them in `check` - for row := range exp { - exp[row] = nil - } - - check(t, tx2, f2, exp) - - PanicOn(tx2.Commit()) - - h3 := NewHolder(filepath.Dir(f2.path()), mustHolderConfig()) - testhook.Cleanup(t, func() { - h3.Close() - }) - - idx3, err := h3.CreateIndex("i", IndexOptions{}) - _ = idx3 - PanicOn(err) - - f3 := newFragment(h3, makeTestFragSpec(f2.path(), "i", "f", viewStandard), 0, 0) - f3.MaxOpN = maxOpN - f3.CacheType = f.CacheType - - tx3 := idx.holder.txf.NewTx(Txo{Write: writable, Index: idx, Fragment: f3, Shard: f3.shard}) - defer tx3.Rollback() - - err = f2.Close() - if err != nil { - t.Fatalf("f2 closing storage: %v", err) - } - - err = f3.Open() - if err != nil { - t.Fatalf("opening f3: %v", err) - } - defer f3.Clean(t) - - check(t, tx3, f3, exp) - - }) - - } - } -} - -func check(t *testing.T, tx Tx, f *fragment, exp map[uint64]map[uint64]struct{}) { - - for rowID, colsExp := range exp { - colsAct := f.mustRow(tx, rowID).Columns() - if len(colsAct) != len(colsExp) { - t.Errorf("row %d len mismatch got: %d exp:%d", rowID, len(colsAct), len(colsExp)) - } - for _, colAct := range colsAct { - if _, ok := colsExp[colAct]; !ok { - t.Errorf("extra column: %d", colAct) - } - } - for colExp := range colsExp { - found := false - for _, colAct := range colsAct { - if colExp == colAct { - found = true - break - } - } - if !found { - t.Errorf("expected %d, but not found", colExp) - } - } - } - -} - func TestImportValueConcurrent(t *testing.T) { f, idx, tx := mustOpenBSIFragment(t, "i", "f", viewBSIGroupPrefix+"foo", 0) defer f.Clean(t) // we will be making a new Tx each time, so we can rollback the default provided one. tx.Rollback() - ty := idx.holder.txf.TxTyp() - switch ty { - case roaringTxn: - t.Skip(fmt.Sprintf("skipping TestImportValueConcurrent under " + - "roaring because the lack of transactional consistency " + - "from Roaring-per-file will create false comparison " + - "failures.")) - } - eg := &errgroup.Group{} for i := 0; i < 4; i++ { i := i @@ -5178,38 +4552,29 @@ func TestImportMultipleValues(t *testing.T) { } for i, test := range tests { - for _, maxOpN := range []int{0, 10000} { // test small/large write - t.Run(fmt.Sprintf("%dLowOpN", i), func(t *testing.T) { - f, _, tx := mustOpenBSIFragment(t, "i", "f", viewBSIGroupPrefix+"foo", 0) - f.MaxOpN = maxOpN - defer f.Clean(t) + t.Run(fmt.Sprintf("%d", i), func(t *testing.T) { + f, _, tx := mustOpenBSIFragment(t, "i", "f", viewBSIGroupPrefix+"foo", 0) + defer f.Clean(t) - err := f.importValue(tx, test.cols, test.vals, test.depth, false) + err := f.importValue(tx, test.cols, test.vals, test.depth, false) + if err != nil { + t.Fatalf("importing values: %v", err) + } + + for i := range test.checkCols { + cc, cv := test.checkCols[i], test.checkVals[i] + n, exists, err := f.value(tx, cc, test.depth) if err != nil { - t.Fatalf("importing values: %v", err) + t.Fatalf("getting value: %v", err) } - - // probably too slow, would hit disk alot: - //PanicOn(tx.Commit()) - //tx = idx.holder.txf.NewTx(Txo{Write: !writable, Index: idx, Fragment: f, Shard:f.shard, ShardSet:true}) - //defer tx.Rollback() - - for i := range test.checkCols { - cc, cv := test.checkCols[i], test.checkVals[i] - n, exists, err := f.value(tx, cc, test.depth) - if err != nil { - t.Fatalf("getting value: %v", err) - } - if !exists { - t.Errorf("column %d should exist", cc) - } - if n != cv { - t.Errorf("wrong value: %d is not %d", n, cv) - } + if !exists { + t.Errorf("column %d should exist", cc) } - }) - - } + if n != cv { + t.Errorf("wrong value: %d is not %d", n, cv) + } + } + }) } } @@ -5241,35 +4606,32 @@ func TestImportValueRowCache(t *testing.T) { } for i, test := range tests { - for _, maxOpN := range []int{1, 10000} { - t.Run(fmt.Sprintf("%dMaxOpN%d", i, maxOpN), func(t *testing.T) { - f, _, tx := mustOpenBSIFragment(t, "i", "f", viewBSIGroupPrefix+"foo", 0) - f.MaxOpN = maxOpN - defer f.Clean(t) + t.Run(fmt.Sprintf("%d", i), func(t *testing.T) { + f, _, tx := mustOpenBSIFragment(t, "i", "f", viewBSIGroupPrefix+"foo", 0) + defer f.Clean(t) - // First import (tc1) - if err := f.importValue(tx, test.tc1.cols, test.tc1.vals, test.tc1.depth, false); err != nil { - t.Fatalf("importing values: %v", err) - } + // First import (tc1) + if err := f.importValue(tx, test.tc1.cols, test.tc1.vals, test.tc1.depth, false); err != nil { + t.Fatalf("importing values: %v", err) + } - if r, err := f.rangeOp(tx, pql.GT, test.tc1.depth, 0); err != nil { - t.Error("getting range of values") - } else if !reflect.DeepEqual(r.Columns(), test.tc1.checkCols) { - t.Errorf("wrong column values. expected: %v, but got: %v", test.tc1.checkCols, r.Columns()) - } + if r, err := f.rangeOp(tx, pql.GT, test.tc1.depth, 0); err != nil { + t.Error("getting range of values") + } else if !reflect.DeepEqual(r.Columns(), test.tc1.checkCols) { + t.Errorf("wrong column values. expected: %v, but got: %v", test.tc1.checkCols, r.Columns()) + } - // Second import (tc2) - if err := f.importValue(tx, test.tc2.cols, test.tc2.vals, test.tc2.depth, false); err != nil { - t.Fatalf("importing values: %v", err) - } + // Second import (tc2) + if err := f.importValue(tx, test.tc2.cols, test.tc2.vals, test.tc2.depth, false); err != nil { + t.Fatalf("importing values: %v", err) + } - if r, err := f.rangeOp(tx, pql.GT, test.tc2.depth, 0); err != nil { - t.Error("getting range of values") - } else if !reflect.DeepEqual(r.Columns(), test.tc2.checkCols) { - t.Errorf("wrong column values. expected: %v, but got: %v", test.tc2.checkCols, r.Columns()) - } - }) - } + if r, err := f.rangeOp(tx, pql.GT, test.tc2.depth, 0); err != nil { + t.Error("getting range of values") + } else if !reflect.DeepEqual(r.Columns(), test.tc2.checkCols) { + t.Errorf("wrong column values. expected: %v, but got: %v", test.tc2.checkCols, r.Columns()) + } + }) } } @@ -5312,64 +4674,6 @@ func TestFragmentConcurrentReadWrite(t *testing.T) { t.Logf("%d", acc) } -func TestRemapCache(t *testing.T) { - f, _, tx := mustOpenFragment(t, "i", "f", viewStandard, 0, "") - defer f.Close() - index, field, view, shard := f.index(), f.field(), f.view(), f.shard - - // request a PanicOn that doesn't kill the program on fault - wouldFault := debug.SetPanicOnFault(true) - defer func() { - debug.SetPanicOnFault(wouldFault) - 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" { - t.Fatalf("segfault trapped during remap test (expected failure mode)") - } - } - t.Fatalf("unexpected PanicOn: %v", r) - } - }() - - // create a container - _, err := tx.Add(index, field, view, shard, 65537) - if err != nil { - t.Fatalf("storage add: %v", err) - } - // cause the container to be mapped - err = f.Snapshot() - if err != nil { - t.Fatalf("storage snapshot: %v", err) - } - // freeze the row - _ = f.mustRow(tx, 0) - // add a bit that isn't in that container, so that container doesn't - // change - _, err = tx.Add(index, field, view, shard, 2) - if err != nil { - t.Fatalf("storage add: %v", err) - } - // make the original container be the most recent, thus cached, container - _, err = f.bit(tx, 0, 65537) - if err != nil { - t.Fatalf("storage bit check: %v", err) - } - // force snapshot, remapping the containers - err = f.Snapshot() - if err != nil { - t.Fatalf("storage snapshot: %v", err) - } - // get rid of the old mapping - runtime.GC() - // try to read that container again - _, err = f.bit(tx, 0, 65537) - if err != nil { - t.Fatalf("storage bit check: %v", err) - } -} - func TestFragment_Bug_Q2DoubleDelete(t *testing.T) { f, idx, tx := mustOpenFragment(t, "i", "f", viewStandard, 0, "") _ = idx @@ -5987,47 +5291,3 @@ func TestSliceDifference(t *testing.T) { compareSlices(t, name, tc.expected, result) } } - -func TestBitmapGrowth(t *testing.T) { - roaringOnlyTest(t) - f, _, tx := mustOpenFragment(t, "i", "f", viewBSIGroupPrefix+"foo", 0, "") - path := f.path() - defer f.Clean(t) - const values = 500 - cols := make([]uint64, values) - vals := make([]int64, values) - for i := range cols { - cols[i] = uint64(rand.Int63n(65536)) - vals[i] = rand.Int63n(24) - } - err := f.importValue(tx, cols, vals, 7, false) - if err != nil { - t.Fatalf("importing values: %v", err) - } - info, err := os.Stat(path) - if err != nil { - t.Fatalf("statting %s: %v", path, err) - } - prevSize := info.Size() - prevOpN := f.opN - err = f.importValue(tx, cols, vals, 7, false) - if err != nil { - t.Fatalf("importing values: %v", err) - } - info, err = os.Stat(path) - if err != nil { - t.Fatalf("statting %s: %v", path, err) - } - deltaSize := info.Size() - prevSize - deltaOpN := f.opN - prevOpN - // This is somewhat arbitrary, but the issue tested for was that - // opN would grow by 0 or 1 with multiple KB of actual ops written. - // If deltaOpN is at least 20, we'll probably see snapshots happening - // at least occasionally, and if deltaSize is under 1024, the writes - // are probably going to be small enough that the regular backlog of - // snapshotting catches them anyway. - if deltaSize > 1024 && deltaOpN < 20 { - t.Fatalf("bitmap grew by %d bytes but OpN only grew by %d", - deltaSize, deltaOpN) - } -} diff --git a/holder.go b/holder.go index cfbd50801..1c4d3c940 100644 --- a/holder.go +++ b/holder.go @@ -85,8 +85,7 @@ type Holder struct { // The interval at which the cached row ids are persisted to disk. cacheFlushInterval time.Duration - Logger logger.Logger - SnapshotQueue SnapshotQueue + Logger logger.Logger // Instantiates new translation stores OpenTranslateStore OpenTranslateStoreFunc @@ -271,8 +270,6 @@ func NewHolder(path string, cfg *HolderConfig) *Holder { Logger: cfg.Logger, Opts: HolderOpts{StorageBackend: cfg.StorageConfig.Backend}, - SnapshotQueue: defaultSnapshotQueue, - Auditor: NewAuditor(), path: path, @@ -734,16 +731,15 @@ func (h *Holder) maybeSpool(msg Message) bool { return true } -// Activate runs the background tasks relevant to keeping a holder in a stable -// state, such as scanning it for needed snapshots, or flushing caches. This -// is separate from opening because, while a server would nearly always want -// to do this, other use cases (like consistency checks of a data directory) +// Activate runs the background tasks relevant to keeping a holder in +// a stable state, such as flushing caches. This is separate from +// opening because, while a server would nearly always want to do +// this, other use cases (like consistency checks of a data directory) // need to avoid it even getting started. func (h *Holder) Activate() { // Periodically flush cache. - h.wg.Add(2) + h.wg.Add(1) go func() { defer h.wg.Done(); h.monitorCacheFlush() }() - go func() { defer h.wg.Done(); h.SnapshotQueue.ScanHolder(h, h.closing) }() } // checkForeignIndex is a check before applying a foreign @@ -791,7 +787,6 @@ func (h *Holder) Close() error { // Notify goroutines of closing and wait for completion. close(h.closing) h.wg.Wait() - for _, index := range h.Indexes() { if err := index.Close(); err != nil { return errors.Wrap(err, "closing index") @@ -809,10 +804,6 @@ func (h *Holder) Close() error { h.opened.mu.Lock() h.opened.ch = make(chan struct{}) h.opened.mu.Unlock() - if h.SnapshotQueue != nil { - h.SnapshotQueue.Stop() - h.SnapshotQueue = nil - } if h.lookupDB != nil { err := h.lookupDB.Close() @@ -827,13 +818,6 @@ func (h *Holder) Close() error { return nil } -func (h *Holder) NeedsSnapshot() bool { - h.mu.RLock() - defer h.mu.RUnlock() - - return h.txf.NeedsSnapshot() -} - // HasData returns true if Holder contains at least one index. // This is used to determine if the rebalancing of data is necessary // when a node joins the cluster. diff --git a/holder_internal_test.go b/holder_internal_test.go index b1455aa5e..a762408b8 100644 --- a/holder_internal_test.go +++ b/holder_internal_test.go @@ -174,17 +174,9 @@ func TestHolderOperatorCancel(t *testing.T) { } } -// mustHolderConfig is meant to help minimize the number of places in the code -// where we're reading the PILOSA_STORAGE_BACKEND environment variable for -// testing purposes. Ideally we would handle this differently, but this is a -// first attempt at improving things. Note: the actual os.Getenv() call was -// moved to the CurrentBackend() function. +// mustHolderConfig sets up a default holder config for tests. func mustHolderConfig() *HolderConfig { cfg := DefaultHolderConfig() - if backend := CurrentBackend(); backend != "" { - _ = MustBackendToTxtype(backend) - cfg.StorageConfig.Backend = backend - } cfg.StorageConfig.FsyncEnabled = false cfg.RBFConfig.FsyncEnabled = false cfg.Schemator = disco.InMemSchemator diff --git a/holder_test.go b/holder_test.go index cff3cf7da..1485f59fc 100644 --- a/holder_test.go +++ b/holder_test.go @@ -5,13 +5,12 @@ import ( "context" "math" "os" - "path/filepath" "reflect" "strings" "testing" "time" - "github.com/molecula/featurebase/v3" + pilosa "github.com/molecula/featurebase/v3" "github.com/molecula/featurebase/v3/disco" "github.com/molecula/featurebase/v3/pql" "github.com/molecula/featurebase/v3/test" @@ -21,10 +20,7 @@ import ( // mustHolderConfig provides a default test-friendly holder config. func mustHolderConfig() *pilosa.HolderConfig { cfg := pilosa.DefaultHolderConfig() - if backend := pilosa.CurrentBackend(); backend != "" { - _ = pilosa.MustBackendToTxtype(backend) - cfg.StorageConfig.Backend = backend - } + cfg.StorageConfig.Backend = "rbf" cfg.StorageConfig.FsyncEnabled = false cfg.RBFConfig.FsyncEnabled = false cfg.Schemator = disco.InMemSchemator @@ -55,109 +51,6 @@ func TestHolder_Open(t *testing.T) { t.Fatalf("unexpected error: %v", err) } }) - t.Run("ErrFragmentStoragePermission", func(t *testing.T) { - roaringOnlyTest(t) - - if os.Geteuid() == 0 { - t.Skip("Skipping permissions test since user is root.") - } - h := test.MustOpenHolder(t) - defer h.Close() - - var idx *pilosa.Index - var err error - if idx, err = h.CreateIndex("foo", pilosa.IndexOptions{}); err != nil { - t.Fatal(err) - } - - var shard uint64 - tx := idx.Txf().NewTx(pilosa.Txo{Write: writable, Index: idx, Shard: shard}) - defer tx.Rollback() - - if field, err := idx.CreateField("bar", pilosa.OptFieldTypeDefault()); err != nil { - t.Fatal(err) - } else if _, err := field.SetBit(tx, 0, 0, nil); err != nil { - t.Fatal(err) - } else if err := tx.Commit(); err != nil { - t.Fatal(err) - } else if err := h.Holder.Close(); err != nil { - t.Fatal(err) - } else if err := os.Chmod(filepath.Join(h.Path(), "foo", "bar", "views", "standard", "fragments", "0"), 0000); err != nil { - t.Fatal(err) - } - defer func() { - _ = os.Chmod(filepath.Join(h.Path(), "foo", "bar", "views", "standard", "fragments", "0"), 0644) - }() - if err := h.Reopen(); err == nil || !strings.Contains(err.Error(), "permission denied") { - t.Fatalf("unexpected error: %s", err) - } - }) - t.Run("ErrFragmentStorageCorrupt", func(t *testing.T) { - roaringOnlyTest(t) - - h := test.MustOpenHolder(t) - defer h.Close() - - var idx *pilosa.Index - var err error - if idx, err = h.CreateIndex("foo", pilosa.IndexOptions{}); err != nil { - t.Fatal(err) - } - - var shard uint64 - tx := idx.Txf().NewTx(pilosa.Txo{Write: writable, Index: idx, Shard: shard}) - if err != nil { - t.Fatal(err) - } - defer tx.Rollback() - - if field, err := idx.CreateField("bar", pilosa.OptFieldTypeDefault()); err != nil { - t.Fatal(err) - } else if _, err := field.SetBit(tx, 0, 0, nil); err != nil { - t.Fatal(err) - } else if err := tx.Commit(); err != nil { - t.Fatal(err) - } else if err := h.Holder.Close(); err != nil { - t.Fatal(err) - } else if err := os.Truncate(filepath.Join(h.Path(), "foo", "bar", "views", "standard", "fragments", "0"), 2); err != nil { - t.Fatal(err) - } - - if err := h.Reopen(); err == nil || !strings.Contains(err.Error(), "open fragment: shard=0, err=opening storage: unmarshal storage") { - t.Fatalf("unexpected error: %s", err) - } - }) - t.Run("ErrFragmentStorageRecoverable", func(t *testing.T) { - roaringOnlyTest(t) - - h := test.MustOpenHolder(t) - defer h.Close() - - idx, err := h.CreateIndex("foo", pilosa.IndexOptions{}) - if err != nil { - t.Fatal(err) - } - var shard uint64 - tx := idx.Txf().NewTx(pilosa.Txo{Write: writable, Index: idx, Shard: shard}) - defer tx.Rollback() - - if field, err := idx.CreateField("bar", pilosa.OptFieldTypeDefault()); err != nil { - t.Fatal(err) - } else if _, err := field.SetBit(tx, 0, 0, nil); err != nil { - t.Fatal(err) - } else if err := tx.Commit(); err != nil { - t.Fatal(err) - } else if err := h.Holder.Close(); err != nil { - t.Fatal(err) - } else if err := os.Truncate(filepath.Join(h.IndexesPath(), "foo", "bar", "views", "standard", "fragments", "0"), 20); err != nil { - t.Fatal(err) - } - - if err := h.Reopen(); err != nil { - t.Fatalf("unexpected error: %s", err) - } - }) - t.Run("ForeignIndex", func(t *testing.T) { t.Run("ErrForeignIndexNotFound", func(t *testing.T) { h := test.MustOpenHolder(t) diff --git a/http/handler.go b/http/handler.go index 700b4dc7a..785e0e7f3 100644 --- a/http/handler.go +++ b/http/handler.go @@ -37,6 +37,7 @@ import ( "github.com/molecula/featurebase/v3/logger" "github.com/molecula/featurebase/v3/pql" "github.com/molecula/featurebase/v3/rbf" + "github.com/molecula/featurebase/v3/storage" "github.com/molecula/featurebase/v3/topology" "github.com/molecula/featurebase/v3/tracing" "github.com/pkg/errors" @@ -1087,7 +1088,7 @@ func (h *Handler) handlePostQuery(w http.ResponseWriter, r *http.Request) { req, ok := qreq.(*pilosa.QueryRequest) if DoPerQueryProfiling { - backend := pilosa.CurrentBackend() + backend := storage.DefaultBackend reqHash := hash(req.Query) qlen := len(req.Query) diff --git a/index.go b/index.go index 864e71346..ba3901d67 100644 --- a/index.go +++ b/index.go @@ -93,10 +93,6 @@ func (i *Index) NewTx(txo Txo) Tx { return i.holder.txf.NewTx(txo) } -func (i *Index) NeedsSnapshot() bool { - return i.holder.txf.NeedsSnapshot() -} - // CreatedAt is an timestamp for a specific version of an index. func (i *Index) CreatedAt() int64 { i.mu.RLock() diff --git a/mmap_test.go b/mmap_test.go deleted file mode 100644 index d1905eb9e..000000000 --- a/mmap_test.go +++ /dev/null @@ -1,68 +0,0 @@ -// Copyright 2021 Molecula Corp. All rights reserved. -package pilosa - -import ( - "math/rand" - "runtime" - "testing" - - "github.com/molecula/featurebase/v3/logger" -) - -type cv struct { - cols []uint64 - vals []int64 -} - -func forceSnapshotsCheckMapping(t *testing.T) { - depth := uint64(6) - f, idx, tx := mustOpenBSIFragment(t, "i", "f", viewStandard, 0) - tx.Rollback() - f.Logger = logger.NewLogfLogger(t) - defer f.Clean(t) - - tx = idx.holder.txf.NewTx(Txo{Write: writable, Index: idx, Fragment: f, Shard: f.shard}) - defer tx.Rollback() - - for i := 0; i < f.MaxOpN; i++ { - _, _ = f.setBit(tx, 0, uint64(32*i)) - } - // force snapshot so we get a mmapped row... - err := f.Snapshot() - if err != nil { - t.Fatalf("initial snapshot error: %v", err) - } - - values := make([]cv, 1024) - for i := range values { - cols := make([]uint64, 128) - vals := make([]int64, 128) - for j := range cols { - // pick values in the first 16 cols of each of the 16 - // shards in a default shardwidth, so each set will - // probably change some values from the previous one. - cols[j] = uint64(((rand.Int63n(16) & int64(i>>2)) << 16) + rand.Int63n(16)) - vals[j] = int64(rand.Int63n(1 << depth)) - } - values[i] = cv{cols, vals} - } - - // modify the original bitmap, until it causes a snapshot, which - // then invalidates the other map... - for i := 0; i < 32; i++ { - cv := values[i%len(values)] - // periodically force gc, so if we have a small pool of maps - // we'll go in and out of mapping mode - if i%5 == 0 { - runtime.GC() - } - err := f.importValue(tx, cv.cols, cv.vals, depth, (i%3 == 1)) - if err != nil { - t.Fatalf("importValue[%d]: %v", i, err) - } - err = f.Snapshot() - if err != nil { - t.Fatalf("snapshot[%d]: %v", i, err) - } - } -} diff --git a/pilosa.go b/pilosa.go index 354c435ff..9cf4f715f 100644 --- a/pilosa.go +++ b/pilosa.go @@ -2,13 +2,11 @@ package pilosa import ( - "os" "regexp" "time" "github.com/molecula/featurebase/v3/disco" pnet "github.com/molecula/featurebase/v3/net" - "github.com/molecula/featurebase/v3/storage" "github.com/pkg/errors" ) @@ -157,20 +155,3 @@ func AddressWithDefaults(addr string) (*pnet.URI, error) { } return pnet.NewURIFromAddress(addr) } - -// CurrentBackend is one step in an attempt to centralize (and either minimize -// or completely remove), the calls to environment variables throughout the -// tests. Ideally we could get rid of this and rely completely on the -// configuration parameters. -func CurrentBackend() string { - return os.Getenv("PILOSA_STORAGE_BACKEND") -} - -// CurrentBackendOrDefault tries the environment variable first, but falls back -// to the default backend if the environment variable is empty. -func CurrentBackendOrDefault() string { - if backend := os.Getenv("PILOSA_STORAGE_BACKEND"); backend != "" { - return backend - } - return storage.DefaultBackend -} diff --git a/pprof.go b/pprof.go index 07f87b1b2..400b48af7 100644 --- a/pprof.go +++ b/pprof.go @@ -19,10 +19,7 @@ import ( // commented out—in holder.go. func CPUProfileForDur(dur time.Duration, outpath string) { // per-query pprof output: - backend := CurrentBackend() - if backend == "" { - backend = storage.DefaultBackend - } + backend := storage.DefaultBackend path := outpath + "." + backend f, err := os.Create(path) vprint.PanicOn(err) @@ -45,10 +42,7 @@ func CPUProfileForDur(dur time.Duration, outpath string) { // commented out—in holder.go. func MemProfileForDur(dur time.Duration, outpath string) { // per-query pprof output: - backend := CurrentBackend() - if backend == "" { - backend = storage.DefaultBackend - } + backend := storage.DefaultBackend path := outpath + "." + backend f, err := os.Create(path) vprint.PanicOn(err) diff --git a/server.go b/server.go index 076142a9a..66332c9fb 100644 --- a/server.go +++ b/server.go @@ -64,11 +64,10 @@ type Server struct { // nolint: maligned schemator disco.Schemator // External - systemInfo SystemInfo - gcNotifier GCNotifier - logger logger.Logger - queryLogger logger.Logger - snapshotQueue SnapshotQueue + systemInfo SystemInfo + gcNotifier GCNotifier + logger logger.Logger + queryLogger logger.Logger nodeID string uri pnet.URI @@ -544,13 +543,6 @@ func (s *Server) UpAndDown() error { func (s *Server) Open() error { s.logger.Infof("open server. PID %v", os.Getpid()) - if s.holder.NeedsSnapshot() { - // Start background monitoring. - s.snapshotQueue = newSnapshotQueue(10, 2, s.logger) - } else { - s.snapshotQueue = defaultSnapshotQueue //TODO (twg) rethink this - } - // Log startup err := s.holder.logStartup() if err != nil { @@ -612,7 +604,6 @@ func (s *Server) Open() error { return errors.Wrap(err, "opening Holder") } // bring up the background tasks for the holder. - s.holder.SnapshotQueue = s.snapshotQueue s.holder.Activate() // if we joined existing cluster then broadcast "resize on add" message if initState == disco.InitialClusterStateExisting { @@ -743,11 +734,6 @@ func (s *Server) Close() error { if s.holder != nil { errh = s.holder.Close() } - if s.snapshotQueue != nil { - s.holder.SnapshotQueue = nil - s.snapshotQueue.Stop() - s.snapshotQueue = nil - } // prefer to return holder error over cluster // error. This order is somewhat arbitrary. It would be better if we had diff --git a/server_internal_test.go b/server_internal_test.go index 9859e950d..da6d57578 100644 --- a/server_internal_test.go +++ b/server_internal_test.go @@ -2,7 +2,6 @@ package pilosa import ( - "runtime" "testing" "time" @@ -10,23 +9,6 @@ import ( "github.com/molecula/featurebase/v3/testhook" ) -// Ensure the file handle count is working -func TestCountOpenFiles(t *testing.T) { - roaringOnlyTest(t) - - // Windows is not supported yet - if runtime.GOOS == "windows" { - t.Skip("Skipping unsupported countOpenFiles test on Windows.") - } - count, err := countOpenFiles() - if err != nil { - t.Errorf("countOpenFiles failed: %s", err) - } - if count == 0 { - t.Error("countOpenFiles returned invalid value 0.") - } -} - func TestMonitorAntiEntropyZero(t *testing.T) { td, err := testhook.TempDirInDir(t, *TempDir, "") diff --git a/snapshotqueue.go b/snapshotqueue.go deleted file mode 100644 index ef0bda24c..000000000 --- a/snapshotqueue.go +++ /dev/null @@ -1,495 +0,0 @@ -// Copyright 2021 Molecula Corp. All rights reserved. -package pilosa - -import ( - "context" - "fmt" - "io" - "math/bits" - "os" - "sync" - "sync/atomic" - "time" - - "github.com/molecula/featurebase/v3/logger" - "github.com/molecula/featurebase/v3/testhook" - "github.com/pkg/errors" -) - -// snapshotQueue is a thing which can handle enqueuing snapshots. A snapshot -// queue distinguishes between high-priority requests, which get satisfied -// by the next available worker, and regular requests, which get enqueued -// if there's space in the queue, and otherwise dropped. There's also a -// separate background task to scan a holder for fragments which may need -// snapshots, but which is processed only when the queue is empty, and only -// slowly. "Await" awaits an existing snapshot if one is already enqueued. -// "Immediate" tries to do one right away. (If one's already enqueued, this -// can leave it in the queue, which will ignore anything that shows up with -// the request flag cleared.) -// -// Await, Enqueue, and Immediate should be called only with the fragment lock -// held. -// -// If you create a queue, it should get stopped at some point. The -// atomicSnapshotQueue implementation used as defaultSnapshotQueue has -// a Start function which will tell you whether it actually started a -// queue. This logic exists because in a normal server case, you probably -// want the queue to be shut down as part of server shutdown, but if you're -// running cluster tests, you probably want to start and shop the queue as -// part of the test, not stop it when any server terminates. -// -// It's less likely to be desireable to start/stop individual queues, -// because fragments use the defaultSnapshotQueue anyway. This design -// needs revisiting. -type SnapshotQueue interface { - Immediate(*fragment) error - Enqueue(*fragment) - Await(*fragment) error - ScanHolder(*Holder, chan struct{}) - Stop() -} - -// queuelessSnapshotQueue isn't a snapshot queue, but it satisfies the -// interface. -type queuelessSnapshotQueue struct{} - -func (q *queuelessSnapshotQueue) Enqueue(f *fragment) { - // We don't actually try to enqueue the snapshot; it breaks things - // if a snapshot gets caused during a transaction. -} - -func (q *queuelessSnapshotQueue) Await(f *fragment) error { - return nil -} - -func (q *queuelessSnapshotQueue) Immediate(f *fragment) error { - return f.snapshot() -} - -func (q *queuelessSnapshotQueue) ScanHolder(h *Holder, done chan struct{}) { -} - -func (q *queuelessSnapshotQueue) Stop() { -} - -var defaultSnapshotQueue = &queuelessSnapshotQueue{} - -// newSnapshotQueue makes a new snapshot queue, of depth N, with -// w worker threads. -func newSnapshotQueue(n int, w int, l logger.Logger) SnapshotQueue { - ctx, cancel := context.WithCancel(context.Background()) - sq := &prioritySnapshotQueue{ - normal: make(chan snapshotRequest, n), - urgent: make(chan snapshotRequest), - background: make(chan snapshotRequest), - ctx: ctx, - cancel: cancel, - maxOpN: 10000, - logger: l, - } - if sq.logger == nil { - sq.logger = logger.NewStandardLogger(os.Stderr) - } - _ = testhook.Opened(NewAuditor(), sq, nil) - sq.spawnWorkers(w) - return sq -} - -type snapshotRequest struct { - frag *fragment - when time.Time -} - -// prioritySnapshotQueue gives preference to "immediate" requests, and -// dispreference to "background" requests from ScanHolder. It timestamps -// requests, so it can discard a request if the most recent snapshot is -// newer than the request. The snapshotPending flag in the fragment is -// used to track that a given fragment thinks it has been successfully -// enqueued. Background requests are not considered enqueued, since -// they'll never get processed if there's anything else. In normal workloads, -// immediate/urgent snapshots should be rare, but we'll happily drop -// most requests on the floor; the scanner should pick them up once things -// are quiet. -type prioritySnapshotQueue struct { - logger logger.Logger - urgent chan snapshotRequest - normal chan snapshotRequest - background chan snapshotRequest - ctx context.Context - cancel context.CancelFunc - mu sync.RWMutex - scanWG, workerWG sync.WaitGroup - maxOpN int - observedOpN [16]uint32 - stats struct { - enqueued uint32 - skipped uint32 - } - stopped bool -} - -func (sq *prioritySnapshotQueue) spawnWorkers(w int) { - sq.mu.Lock() - defer sq.mu.Unlock() - if sq.ctx.Err() != nil { - sq.logger.Infof("prioritySnapshotQueue worker: already done") - return - } - sq.workerWG.Add(w) - for i := 0; i < w; i++ { - go sq.worker(sq.ctx, sq.urgent, sq.normal, sq.background) - } -} - -func (sq *prioritySnapshotQueue) worker(ctx context.Context, urgent, normal, background chan snapshotRequest) { - defer sq.workerWG.Done() - done := ctx.Done() - ok := true - var req snapshotRequest - for ok { - req.frag = nil - select { - case _, ok = <-done: - case req, ok = <-urgent: - default: - select { - case _, ok = <-done: - case req, ok = <-urgent: - case req, ok = <-normal: - default: - select { - case _, ok = <-done: - case req, ok = <-urgent: - case req, ok = <-normal: - case req, ok = <-background: - } - } - } - if req.frag != nil { - sq.process(req) - } - } -} - -// process actually runs a fragment. it will do this if either the fragment -// has a pending snapshot, or the force flag is set. -func (sq *prioritySnapshotQueue) process(req snapshotRequest) { - f := req.frag - f.mu.Lock() - defer f.mu.Unlock() - if f.snapshotStamp.Before(req.when) { - f.snapshotErr = f.snapshot() - if f.snapshotErr != nil { - fmt.Printf("ERROR: snapshot error: %v\n", f.snapshotErr) - sq.logger.Errorf("snapshot error: %v", f.snapshotErr) - } - f.snapshotPending = false - f.snapshotCond.Broadcast() - } -} - -// Stop shuts down the snapshot queue. It first marks it as done, causing -// the background scanner(s), if any, to shut down, then waits for them, then -// closes and nils the queues. The background scanner has to get stopped -// because otherwise it might try to write to those closed queues. -func (sq *prioritySnapshotQueue) Stop() { - sq.mu.Lock() - defer sq.mu.Unlock() - if sq.stopped { - return - } - sq.stopped = true - sq.cancel() - // scanners need to be done before we close the other channels. - sq.scanWG.Wait() - close(sq.normal) - sq.normal = nil - close(sq.urgent) - sq.urgent = nil - close(sq.background) - sq.background = nil - _ = testhook.Closed(NewAuditor(), sq, nil) - enqueued := atomic.LoadUint32(&sq.stats.enqueued) - skipped := atomic.LoadUint32(&sq.stats.skipped) - if skipped > 0 || enqueued > 1 { - sq.logger.Infof("snapshot queue: enqueued %d, skipped %d\n", sq.stats.enqueued, sq.stats.skipped) - } -} - -// Enqueue tries to add a fragment to the queue, if the fragment is not already -// enqueued. You should hold a lock on the fragment when calling this. -func (sq *prioritySnapshotQueue) Enqueue(f *fragment) { - if f.snapshotPending { - return - } - sq.observeOpN(uint32(f.opN)) - sq.mu.RLock() - defer sq.mu.RUnlock() - if sq.normal == nil { - sq.logger.Infof("requested snapshot after snapshot queue was closed") - return - } - // we have to set this before enqueing, because it's - // otherwise possible that we're at the head of the queue, - // and the recipient gets the fragment before we execute the - // line after the send. - f.snapshotPending = true - // try to enqueue snapshot - select { - case sq.normal <- snapshotRequest{frag: f, when: time.Now()}: - atomic.AddUint32(&sq.stats.enqueued, 1) - return - default: - atomic.AddUint32(&sq.stats.skipped, 1) - f.snapshotPending = false - return - } -} - -// Await returns when f is not pending a snapshot. Call with the fragment lock -// held. Await waits on a condition variable inside f, associated with the -// fragment's lock, so this does not conflict with the lock being used for -// snapshots. -// -// Note that workers don't stop just because the queue's been stopped; only -// the background scanner is stopped. So an Await shouldn't block forever -// even if the queue gets shut down. If you're reading this, possibly that -// analysis is incorrect. -func (sq *prioritySnapshotQueue) Await(f *fragment) (err error) { - for f.snapshotPending { - f.snapshotCond.Wait() - } - err, f.snapshotErr = f.snapshotErr, nil - return err -} - -// Immediate forces an immediate snapshot of the given fragment. Call with -// the fragment locked. If the queue is already closing, the fragment does -// not get snapshotted. -func (sq *prioritySnapshotQueue) Immediate(f *fragment) error { - sq.mu.RLock() - // no deferred unlock, because we want to unlock this before calling Await. - // Not because that needs this lock, but because once we're that far, we - // *don't* need this lock anymore so someone else should have it. - if sq.urgent == nil { - sq.mu.RUnlock() - sq.logger.Errorf("requested immediate snapshot after snapshot queue was closed") - return errors.New("requested immediate snapshot after snapshot queue was closed") - } - f.snapshotPending = true - sq.observeOpN(uint32(f.opN)) - req := snapshotRequest{frag: f, when: time.Now()} - // if the fragment was already in the work queue, it's *possible* - // that the only available worker just picked it off the queue, and - // is now waiting on getting the fragment's lock, so it can run - // a snapshot. So we let go of the lock on the fragment, send the - // request, then request the fragment lock again, because Await will - // be sleeping on the condition variable associated with the lock, - // which means it needs to hold the lock so it can let it go during - // the wait... No, really, this made sense. - f.mu.Unlock() - sq.urgent <- req - sq.mu.RUnlock() - f.mu.Lock() - return sq.Await(f) -} - -// ScanHolder spawns a goroutine which iterates through the holder's -// indexes/fields/views/fragments, looking for fragments which have OpN -// high enough to justify a snapshot but don't seem to have one pending. -// It then dumps these in the low priority background queue. -func (sq *prioritySnapshotQueue) ScanHolder(h *Holder, done chan struct{}) { - sq.mu.Lock() - sq.scanWG.Add(1) - go sq.scanHolderWorker(h, sq.background, done) - sq.mu.Unlock() -} - -// observeOpN reports that a given value of opN was "observed", meaning, -// we encountered a fragment which had that value. This happens for every -// enqueue/immediate, including enqueue attempts which fail to actually -// enter the queue, and it also happens for fragments noticed by the background -// scan but which don't have high enough opN to trigger a snapshot. -func (sq *prioritySnapshotQueue) observeOpN(n uint32) { - // aka "log2(n) + 1", or 0 for n==0 - pow2 := 32 - bits.LeadingZeros32(n) - // 15 == 16384. Our usual fragment maxOpN is 10k, so most fragments - // should end up in the 8k-16k bucket, rather than the 16k+ bucket, - // unless we've got a lot of ingests with large batches going on, - // in which case the 16k bucket will win. - if pow2 > 15 { - pow2 = 15 - } - // store in inverse order so the lowest slot in the array is the - // highest cardinality - atomic.AddUint32(&sq.observedOpN[15-pow2], 1) -} - -// computeMaxOpN tries to pick a reasonable new maxOpN for the background -// scan to use. On a quiet system, we want to gradually lower opN, picking -// the fragments with the highest opN values first, because those offer the -// largest benefit. So, whenever we check a fragment in the background, if we -// *don't* snapshot it, we'll "observe" its OpN value, and then we pick a -// value which picks up at least 1/4 of them. -// -// If there's ingest activity, the Immediate and Enqueue operations will -// "observe" the OpN of fragments submitted to them. This can drive OpN back -// up, if those fragments frequently have very high opN values, which reflects -// the fact that we have enough of that activity that we don't need the -// background scanner adding more. -// -// If we have enough ingest activity that the background scanner never actually -// gets to submit work, we'll rarely get here, because the background scanner -// will block until there's no snapshots pending for the normal workload. -// When we do, we'll probably pick a MaxOpN which is dominated by the ingest -// workload's opN values. So for instance, if everything coming in from the -// ingest workload has 10k or more items, because that's the default fragment -// maxOpN, that will probably set the background snapshot queue value to 8k. -func (sq *prioritySnapshotQueue) computeMaxOpN() { - sq.logger.Debugf("observedOpN by power of 2: %d\n", sq.observedOpN[:]) - total := uint32(0) - for i := range sq.observedOpN { - total += atomic.LoadUint32(&sq.observedOpN[i]) - } - target := (total / 4) + 1 - subTotal := uint32(0) - for i := range sq.observedOpN { - v := atomic.LoadUint32(&sq.observedOpN[i]) - subTotal += v - if subTotal >= target { - prevMaxOpN := sq.maxOpN - sq.maxOpN = (1 << (15 - uint(i))) / 2 - if sq.maxOpN > 0 { - sq.maxOpN-- - } - if prevMaxOpN != sq.maxOpN { - sq.logger.Infof("background scan: %d/%d fragments considered have opN %d or higher\n", - subTotal, total, sq.maxOpN) - } - break - } - } - // It's conceptually possible that we'll miss a couple of observations - // here but that's not really important. This is all pretty approximate. - for i := range sq.observedOpN { - atomic.StoreUint32(&sq.observedOpN[i], 0) - } -} - -// prioritySnapshotQueueScanner is the data type that implements HolderOperator -// and represents a single scan of a holder, with a given maxOpN. -type prioritySnapshotQueueScanner struct { - HolderFilterAll - HolderProcessNone - sq *prioritySnapshotQueue - holder *Holder - queue chan snapshotRequest - ctx context.Context - maxOpN int - seen, hits, counter int -} - -func (s *prioritySnapshotQueueScanner) ProcessFragment(f *fragment) error { - if f == nil { - return nil - } - s.seen++ - // we can't defer this reasonably, because otherwise we'll keep - // the fragment locked forever if we end up trying to send it - // to the queue, but the workers are busy on other fragments. - f.mu.Lock() - open := f.open - snapshotPending, opN := f.snapshotPending, f.opN - f.mu.Unlock() - - // a pending snapshot is one that is either in the normal or - // immediate queue, or is trying to get into the normal queue - // and about to fail, but either way, it already got observed - // there, so we don't need to observe it here. A closed fragment - // doesn't matter to us -- it should be a transient state that - // happens during a shutdown, or shouldn't happen, but we don't - // care about it. - if snapshotPending || !open { - return nil - } - if opN <= s.maxOpN { - // observe the value but don't do a snapshot - s.sq.observeOpN(uint32(opN)) - s.counter++ - if s.counter == 1000 { - select { - case <-time.After(1 * time.Second): - case <-s.ctx.Done(): - return io.EOF - } - s.counter = 0 - } - return nil - } - // we don't observe values when we decide to trigger a snapshot, - // because those values will be changing anyway. we could also - // observe them as zero, but that's also sort of wrong. - s.hits++ - select { - case s.queue <- snapshotRequest{frag: f, when: time.Now()}: - s.sq.logger.Debugf("found fragment needing snapshot: %s\n", f.path()) - case <-s.ctx.Done(): - return io.EOF - } - return nil - -} - -func contextMergedWithStructChan(ctx context.Context, ch chan struct{}) (context.Context, context.CancelFunc) { - canCancel, cancel := context.WithCancel(ctx) - go func() { - select { - case <-ctx.Done(): - cancel() - case <-ch: - cancel() - case <-canCancel.Done(): - // don't need to cancel, but do need to exit this - // function - } - }() - return canCancel, cancel -} - -// scanHolderWorker is a background task that scans a holder looking for -// fragments which need snapshots taken. It's the cleanup task for snapshots -// that would have been requested by Enqueue, but the queue was full. -func (sq *prioritySnapshotQueue) scanHolderWorker(h *Holder, background chan snapshotRequest, done chan struct{}) { - defer sq.scanWG.Done() - ctx, cancel := contextMergedWithStructChan(sq.ctx, done) - defer cancel() - scanner := &prioritySnapshotQueueScanner{ - sq: sq, - holder: h, - queue: background, - ctx: sq.ctx, - maxOpN: sq.maxOpN, - } - for { - err := h.Process(ctx, scanner) - if err != nil { - return - } - - if scanner.hits > 0 { - sq.logger.Infof("background scan: %d/%d fragments needed snapshots\n", scanner.hits, scanner.seen) - scanner.hits = 0 - } else { - sq.logger.Debugf("background scan: no fragments needed snapshots, waiting\n") - // No reason to be active if we're not finding anything. - select { - case <-time.After(60 * time.Second): - case <-ctx.Done(): - return - } - } - scanner.seen = 0 - sq.computeMaxOpN() - scanner.maxOpN = sq.maxOpN - } -} diff --git a/stattx.go b/stattx.go index 8b34ec582..4780f70bc 100644 --- a/stattx.go +++ b/stattx.go @@ -12,6 +12,7 @@ import ( "github.com/molecula/featurebase/v3/debugstats" "github.com/molecula/featurebase/v3/roaring" txkey "github.com/molecula/featurebase/v3/short_txkey" + "github.com/molecula/featurebase/v3/storage" "github.com/molecula/featurebase/v3/vprint" ) @@ -56,7 +57,7 @@ func (w *callStats) reset() { } func (c *callStats) report() (r string) { - backend := CurrentBackend() + backend := storage.DefaultBackend r = fmt.Sprintf("callStats: (%v)\n", backend) c.mu.Lock() defer c.mu.Unlock() diff --git a/storage/config.go b/storage/config.go index f1307943e..efb44d9d7 100644 --- a/storage/config.go +++ b/storage/config.go @@ -3,9 +3,7 @@ package storage // public strings that pilosa/server/config.go can reference const ( - RoaringBackend string = "roaring" - RBFBackend string = "rbf" - BoltBackend string = "bolt" + RBFBackend string = "rbf" ) // DefaultBackend is set here. pilosa/server/config.go references it diff --git a/test/cluster.go b/test/cluster.go index 5aea1147f..d66da2c7b 100644 --- a/test/cluster.go +++ b/test/cluster.go @@ -593,7 +593,7 @@ func prependTestServerOpts(opts []server.CommandOption) []server.CommandOption { pilosa.OptServerOpenTranslateStore(pilosa.OpenInMemTranslateStore), pilosa.OptServerNodeDownRetries(5, 100*time.Millisecond), pilosa.OptServerStorageConfig(&storage.Config{ - Backend: pilosa.CurrentBackendOrDefault(), + Backend: storage.DefaultBackend, FsyncEnabled: false, }), ), diff --git a/tournament.sh b/tournament.sh deleted file mode 100755 index 1729ef167..000000000 --- a/tournament.sh +++ /dev/null @@ -1,13 +0,0 @@ -#!/bin/bash - -## tournament.sh runs a sequence of duels between greens and blues. -## Each test run changes the PILOSA_STORAGE_BACKEND and runs either -## one or two backends through the rigors of make testv-race. -## logs are saved to the tourna.log.${i} files. - -for i in rbf roaring bolt rbf_roaring roaring_rbf roaring_bolt; do - echo "$(date) starting ${i}, output to tourna.log.${i}" - echo "***=== ${i} ====================*** $(date)" &> tourna.log.${i} - PILOSA_STORAGE_BACKEND=${i} make testv-race 2>&1 > tourna.log.${i} -done - diff --git a/tx_test.go b/tx_test.go index 6f1815c8e..95c82f4af 100644 --- a/tx_test.go +++ b/tx_test.go @@ -4,13 +4,11 @@ package pilosa_test import ( "context" "fmt" - "strings" "testing" pilosa "github.com/molecula/featurebase/v3" "github.com/molecula/featurebase/v3/http" "github.com/molecula/featurebase/v3/server" - "github.com/molecula/featurebase/v3/storage" "github.com/molecula/featurebase/v3/test" . "github.com/molecula/featurebase/v3/vprint" // nolint:staticcheck ) @@ -47,15 +45,7 @@ func queryBalances(m0api *pilosa.API, acctOwnerID uint64, fldAcct0, fldAcct1, in return } -func skipForRoaring(t *testing.T) { - src := pilosa.CurrentBackend() - if (storage.DefaultBackend == pilosa.RoaringTxn) || strings.Contains(src, "roaring") { - t.Skip("skip if roaring pseudo-txn involved -- won't show transactional rollback") - } -} - func TestAPI_ImportAtomicRecord(t *testing.T) { - skipForRoaring(t) c := test.MustRunCluster(t, 1, []server.CommandOption{ server.OptCommandServerOptions( diff --git a/txfactory.go b/txfactory.go index c16208532..8b404b5cb 100644 --- a/txfactory.go +++ b/txfactory.go @@ -17,8 +17,7 @@ import ( // public strings that pilosa/server/config.go can reference const ( - RoaringTxn string = "roaring" - RBFTxn string = "rbf" + RBFTxn string = "rbf" ) // DetectMemAccessPastTx true helps us catch places in api and executor @@ -377,9 +376,8 @@ type TxFactory struct { type txtype int const ( - noneTxn txtype = 0 - roaringTxn txtype = 1 // these don't really have any transactions - rbfTxn txtype = 2 + noneTxn txtype = 0 + rbfTxn txtype = 2 ) // DirectoryName just returns a string version of the transaction type. We @@ -388,8 +386,6 @@ const ( // replaced/removed) during that refactor. func (ty txtype) DirectoryName() string { switch ty { - case roaringTxn: - return "roaring" case rbfTxn: return "rbf" } @@ -397,18 +393,12 @@ func (ty txtype) DirectoryName() string { return "" } -func (txf *TxFactory) NeedsSnapshot() (b bool) { - return txf.typ == roaringTxn -} - func MustBackendToTxtype(backend string) (typ txtype) { if strings.Contains(backend, "_") { panic("blue-green comparisons removed") } switch backend { - case RoaringTxn: // "roaring" - return roaringTxn case RBFTxn: // "rbf" return rbfTxn } @@ -839,8 +829,6 @@ func (ty txtype) String() string { switch ty { case noneTxn: return "noneTxn" - case roaringTxn: - return "roaring" case rbfTxn: return "rbf" } @@ -946,16 +934,10 @@ func anyGlobalDBWrappersStillOpen() bool { return false } -func (f *TxFactory) hasRoaring() bool { - return f.typ == roaringTxn -} - func (f *TxFactory) hasRBF() bool { return f.typ == rbfTxn } -var _ = (&TxFactory{}).hasRoaring // happy linter - func (f *TxFactory) GetDBShardPath(index string, shard uint64, idx *Index, ty txtype, write bool) (shardPath string, err error) { dbs, err := f.dbPerShard.GetDBShard(index, shard, idx) if err != nil { diff --git a/txfactory_internal_test.go b/txfactory_internal_test.go index 46f8c918b..b32887605 100644 --- a/txfactory_internal_test.go +++ b/txfactory_internal_test.go @@ -8,8 +8,8 @@ import ( func Test_TxFactory_verifyStringConstantsMatch(t *testing.T) { // txtype.String() method MUST return strings that match // our const definitions at the top of txfactory.go. - check := []txtype{roaringTxn, rbfTxn} - expect := []string{RoaringTxn, RBFTxn} + check := []txtype{rbfTxn} + expect := []string{RBFTxn} for i, chk := range check { obs := chk.String() if obs != expect[i] {