diff --git a/bolt.go b/bolt.go index e679b1400..12435abef 100644 --- a/bolt.go +++ b/bolt.go @@ -135,8 +135,12 @@ func (r *boltRegistrar) OpenDBWrapper(path string, doAllocZero bool, cfg *storag if !DirExists(path) { PanicOn(os.MkdirAll(dir, 0755)) } + fsyncEnabled := true + if cfg != nil { + fsyncEnabled = cfg.FsyncEnabled + } - db, err := bolt.Open(path, 0666, &bolt.Options{Timeout: 5 * time.Second, InitialMmapSize: TxInitialMmapSize}) + db, err := bolt.Open(path, 0666, &bolt.Options{Timeout: 5 * time.Second, InitialMmapSize: TxInitialMmapSize, NoSync: !fsyncEnabled}) if err != nil { return nil, errors.Wrapf(err, fmt.Sprintf("open bolt path '%v'", path)) } @@ -187,6 +191,7 @@ func (r *boltRegistrar) OpenDBWrapper(path string, doAllocZero bool, cfg *storag openTx: make(map[*BoltTx]bool), DeleteEmptyContainer: true, + fsyncEnabled: cfg.FsyncEnabled, } r.unprotectedRegister(w) @@ -226,7 +231,7 @@ func (w *BoltWrapper) CloseDB() error { func (w *BoltWrapper) OpenDB() error { w.muDb.Lock() defer w.muDb.Unlock() - db, err := bolt.Open(w.path, 0666, &bolt.Options{Timeout: 5 * time.Second, InitialMmapSize: TxInitialMmapSize}) + db, err := bolt.Open(w.path, 0666, &bolt.Options{Timeout: 5 * time.Second, InitialMmapSize: TxInitialMmapSize, NoSync: !w.fsyncEnabled}) if err != nil { return err } @@ -315,6 +320,7 @@ type BoltWrapper struct { doAllocZero bool DeleteEmptyContainer bool + fsyncEnabled bool // for tracking whether our initial config wanted fsync on openTx map[*BoltTx]bool } diff --git a/bolt_test.go b/bolt_test.go index 5c83ae1af..1dc03b7d0 100644 --- a/bolt_test.go +++ b/bolt_test.go @@ -21,6 +21,7 @@ import ( "testing" "github.com/molecula/featurebase/v2/roaring" + "github.com/molecula/featurebase/v2/storage" . "github.com/molecula/featurebase/v2/vprint" // nolint:staticcheck ) @@ -88,7 +89,7 @@ func mustOpenEmptyBoltWrapper(path string) (w *BoltWrapper, cleaner func()) { var err error fn := path PanicOn(os.RemoveAll(fn)) - ww, err := globalBoltReg.OpenDBWrapper(fn, DetectMemAccessPastTx, nil) + ww, err := globalBoltReg.OpenDBWrapper(fn, DetectMemAccessPastTx, &storage.Config{FsyncEnabled: false}) PanicOn(err) w = ww.(*BoltWrapper) diff --git a/boltdb/translate.go b/boltdb/translate.go index 076c77f37..35ef2b08f 100644 --- a/boltdb/translate.go +++ b/boltdb/translate.go @@ -55,8 +55,8 @@ const ( ) // OpenTranslateStore opens and initializes a boltdb translation store. -func OpenTranslateStore(path, index, field string, partitionID, partitionN int) (pilosa.TranslateStore, error) { - s := NewTranslateStore(index, field, partitionID, partitionN) +func OpenTranslateStore(path, index, field string, partitionID, partitionN int, fsyncEnabled bool) (pilosa.TranslateStore, error) { + s := NewTranslateStore(index, field, partitionID, partitionN, fsyncEnabled) s.Path = path if err := s.Open(); err != nil { return nil, err @@ -88,22 +88,24 @@ type TranslateStore struct { once sync.Once closing chan struct{} - readOnly bool - writeNotify chan struct{} + readOnly bool + fsyncEnabled bool + writeNotify chan struct{} // File path to database file. Path string } // NewTranslateStore returns a new instance of TranslateStore. -func NewTranslateStore(index, field string, partitionID, partitionN int) *TranslateStore { +func NewTranslateStore(index, field string, partitionID, partitionN int, fsyncEnabled bool) *TranslateStore { return &TranslateStore{ - index: index, - field: field, - partitionID: partitionID, - partitionN: partitionN, - closing: make(chan struct{}), - writeNotify: make(chan struct{}), + index: index, + field: field, + partitionID: partitionID, + partitionN: partitionN, + closing: make(chan struct{}), + writeNotify: make(chan struct{}), + fsyncEnabled: fsyncEnabled, } } @@ -120,7 +122,7 @@ func (s *TranslateStore) Open() (err error) { if err := os.MkdirAll(filepath.Dir(s.Path), 0777); err != nil { return errors.Wrapf(err, "mkdir %s", filepath.Dir(s.Path)) - } else if s.db, err = bolt.Open(s.Path, 0666, &bolt.Options{Timeout: 1 * time.Second}); err != nil { + } else if s.db, err = bolt.Open(s.Path, 0666, &bolt.Options{Timeout: 1 * time.Second, NoSync: !s.fsyncEnabled}); err != nil { return errors.Wrapf(err, "open file: %s", err) } diff --git a/boltdb/translate_test.go b/boltdb/translate_test.go index a42d68d50..61f2e61c3 100644 --- a/boltdb/translate_test.go +++ b/boltdb/translate_test.go @@ -393,7 +393,7 @@ func MustNewTranslateStore(tb testing.TB) *boltdb.TranslateStore { panic(err) } - s := boltdb.NewTranslateStore("I", "F", 0, topology.DefaultPartitionN) + s := boltdb.NewTranslateStore("I", "F", 0, topology.DefaultPartitionN, false) s.Path = f.Name() return s } diff --git a/cluster_internal_test.go b/cluster_internal_test.go index 8af84efed..cc55ae9aa 100644 --- a/cluster_internal_test.go +++ b/cluster_internal_test.go @@ -155,7 +155,10 @@ func newIndexWithTempPath(tb testing.TB, name string) *Index { if err != nil { panic(err) } - h := NewHolder(path, nil) + cfg := DefaultHolderConfig() + cfg.StorageConfig.FsyncEnabled = false + cfg.RBFConfig.FsyncEnabled = false + h := NewHolder(path, cfg) PanicOn(h.Open()) index, err := h.CreateIndex(name, IndexOptions{}) testhook.Cleanup(tb, func() { diff --git a/dbshard_internal_test.go b/dbshard_internal_test.go index 81d64436c..9482d1504 100644 --- a/dbshard_internal_test.go +++ b/dbshard_internal_test.go @@ -330,6 +330,7 @@ func Test_DBPerShard_GetFieldView2Shards_map_from_RBF(t *testing.T) { cfg := mustHolderConfig() cfg.StorageConfig.Backend = "rbf" + cfg.StorageConfig.FsyncEnabled = false holder := NewHolder(tmpdir, cfg) defer holder.Close() diff --git a/executor_internal_test.go b/executor_internal_test.go index 4aa37971c..b752957ef 100644 --- a/executor_internal_test.go +++ b/executor_internal_test.go @@ -28,7 +28,7 @@ import ( func TestExecutor_TranslateRowsOnBool(t *testing.T) { path, _ := testhook.TempDirInDir(t, *TempDir, "pilosa-executor-") - holder := NewHolder(path, nil) + holder := NewHolder(path, mustHolderConfig()) defer holder.Close() e := &executor{ diff --git a/executor_test.go b/executor_test.go index 414bf47b5..adcf64dbb 100644 --- a/executor_test.go +++ b/executor_test.go @@ -6882,7 +6882,7 @@ func TestMissingKeyRegression(t *testing.T) { c := test.MustRunCluster(t, 1, []server.CommandOption{server.OptCommandServerOptions( pilosa.OptServerStorageConfig(&storage.Config{ Backend: "roaring", - FsyncEnabled: true, + FsyncEnabled: false, }))}) defer c.Close() diff --git a/field.go b/field.go index a6c4b5e75..4edccfb3f 100644 --- a/field.go +++ b/field.go @@ -648,7 +648,7 @@ func (f *Field) writeAvailableShards() { func (f *Field) applyTranslateStore() error { // Instantiate & open translation store. var err error - f.translateStore, err = f.OpenTranslateStore(f.TranslateStorePath(), f.index, f.name, -1, -1) + f.translateStore, err = f.OpenTranslateStore(f.TranslateStorePath(), f.index, f.name, -1, -1, f.holder.cfg.StorageConfig.FsyncEnabled) if err != nil { return errors.Wrap(err, "opening field translate store") } diff --git a/field_internal_test.go b/field_internal_test.go index 88ca4b7d7..080836baa 100644 --- a/field_internal_test.go +++ b/field_internal_test.go @@ -244,6 +244,8 @@ 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) PanicOn(h.Open()) diff --git a/field_test.go b/field_test.go index 38056e8ac..0bc3b60cb 100644 --- a/field_test.go +++ b/field_test.go @@ -159,7 +159,7 @@ func TestField_NameRestriction(t *testing.T) { if err != nil { panic(err) } - field, err := pilosa.NewField(pilosa.NewHolder(path, nil), path, "i", ".meta", pilosa.OptFieldTypeDefault()) + field, err := pilosa.NewField(pilosa.NewHolder(path, mustHolderConfig()), path, "i", ".meta", pilosa.OptFieldTypeDefault()) if field != nil { t.Fatalf("unexpected field name %s", err) } @@ -192,13 +192,13 @@ func TestField_NameValidation(t *testing.T) { panic(err) } for _, name := range validFieldNames { - _, err := pilosa.NewField(pilosa.NewHolder(path, nil), path, "i", name, pilosa.OptFieldTypeDefault()) + _, err := pilosa.NewField(pilosa.NewHolder(path, mustHolderConfig()), path, "i", name, pilosa.OptFieldTypeDefault()) if err != nil { t.Fatalf("unexpected field name: %s %s", name, err) } } for _, name := range invalidFieldNames { - _, err := pilosa.NewField(pilosa.NewHolder(path, nil), path, "i", name, pilosa.OptFieldTypeDefault()) + _, err := pilosa.NewField(pilosa.NewHolder(path, mustHolderConfig()), path, "i", name, pilosa.OptFieldTypeDefault()) if err == nil { t.Fatalf("expected error on field name: %s", name) } diff --git a/fragment_internal_test.go b/fragment_internal_test.go index a7915fcc3..9b4ac461a 100644 --- a/fragment_internal_test.go +++ b/fragment_internal_test.go @@ -3166,7 +3166,7 @@ func BenchmarkImportIntoLargeFragment(b *testing.B) { origF.Close() fi.Close() - h := NewHolder(fi.Name(), nil) + h := NewHolder(fi.Name(), mustHolderConfig()) PanicOn(h.Open()) idx, err := h.CreateIndex("i", IndexOptions{}) PanicOn(err) @@ -5146,7 +5146,7 @@ func TestImportClearRestart(t *testing.T) { PanicOn(tx2.Commit()) - h3 := NewHolder(filepath.Dir(f2.path()), nil) + h3 := NewHolder(filepath.Dir(f2.path()), mustHolderConfig()) testhook.Cleanup(t, func() { h3.Close() }) diff --git a/holder.go b/holder.go index 0daf8416d..dea18e3e7 100644 --- a/holder.go +++ b/holder.go @@ -109,7 +109,7 @@ type Holder struct { OpenTransactionStore OpenTransactionStoreFunc // Func to open the ID allocator. - OpenIDAllocator func(string) (*idAllocator, error) + OpenIDAllocator func(string, bool) (*idAllocator, error) // transactionManager transactionManager *TransactionManager @@ -241,7 +241,7 @@ func DefaultHolderConfig() *HolderConfig { OpenTranslateStore: OpenInMemTranslateStore, OpenTranslateReader: nil, OpenTransactionStore: OpenInMemTransactionStore, - OpenIDAllocator: func(string) (*idAllocator, error) { return &idAllocator{}, nil }, + OpenIDAllocator: func(string, bool) (*idAllocator, error) { return &idAllocator{}, nil }, TranslationSyncer: NopTranslationSyncer, Serializer: GobSerializer, Schemator: disco.InMemSchemator, @@ -623,7 +623,7 @@ func (h *Holder) Open() error { h.transactionManager.Log = h.Logger // Open ID allocator. - h.ida, err = h.OpenIDAllocator(filepath.Join(h.path, "idalloc.db")) + h.ida, err = h.OpenIDAllocator(filepath.Join(h.path, "idalloc.db"), h.cfg.StorageConfig.FsyncEnabled) if err != nil { return errors.Wrap(err, "opening ID allocator") } diff --git a/holder_internal_test.go b/holder_internal_test.go index c87b6bf73..bd35f0b03 100644 --- a/holder_internal_test.go +++ b/holder_internal_test.go @@ -86,6 +86,7 @@ func makeHolder(tb testing.TB, backend string) (*Holder, string, error) { cfg := mustHolderConfig() if backend != "" { cfg.StorageConfig.Backend = backend + cfg.StorageConfig.FsyncEnabled = false } h := NewHolder(path, cfg) return h, path, h.Open() @@ -265,6 +266,8 @@ func mustHolderConfig() *HolderConfig { _ = MustBackendToTxtype(backend) cfg.StorageConfig.Backend = backend } + cfg.StorageConfig.FsyncEnabled = false + cfg.RBFConfig.FsyncEnabled = false cfg.Schemator = disco.InMemSchemator cfg.Sharder = disco.InMemSharder return cfg diff --git a/holder_test.go b/holder_test.go index 6aeeab79d..4807a997e 100644 --- a/holder_test.go +++ b/holder_test.go @@ -25,11 +25,26 @@ import ( "time" "github.com/molecula/featurebase/v2" + "github.com/molecula/featurebase/v2/disco" "github.com/molecula/featurebase/v2/pql" "github.com/molecula/featurebase/v2/test" "github.com/pkg/errors" ) +// 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.FsyncEnabled = false + cfg.RBFConfig.FsyncEnabled = false + cfg.Schemator = disco.InMemSchemator + cfg.Sharder = disco.InMemSharder + return cfg +} + func TestHolder_Open(t *testing.T) { t.Run("ErrIndexPermission", func(t *testing.T) { if os.Geteuid() == 0 { @@ -283,7 +298,7 @@ func TestHolder_HasData(t *testing.T) { // Note that we are intentionally not using test.NewHolder, // because we want to create a Holder object with an invalid path, // rather than creating a valid holder with a temporary path. - h := pilosa.NewHolder("bad-path", nil) + h := pilosa.NewHolder("bad-path", mustHolderConfig()) if ok, err := h.HasData(); ok || err != nil { t.Fatal("expected HasData to return false, no err, but", ok, err) diff --git a/idalloc.go b/idalloc.go index 4c6ea423f..ffa07baf8 100644 --- a/idalloc.go +++ b/idalloc.go @@ -53,18 +53,20 @@ func (k IDAllocKey) String() string { } type idAllocator struct { - db *bolt.DB + db *bolt.DB + fsyncEnabled bool } -type OpenIDAllocatorFunc func(path string) (*idAllocator, error) // whyyyyyyyyy +type OpenIDAllocatorFunc func(path string, enableFsync bool) (*idAllocator, error) // whyyyyyyyyy -func OpenIDAllocator(path string) (*idAllocator, error) { - db, err := bolt.Open(path, 0666, &bolt.Options{Timeout: 1 * time.Second}) +func OpenIDAllocator(path string, enableFsync bool) (*idAllocator, error) { + db, err := bolt.Open(path, 0666, &bolt.Options{Timeout: 1 * time.Second, NoSync: !enableFsync}) if err != nil { return nil, err } - return &idAllocator{db}, nil + return &idAllocator{db: db, fsyncEnabled: enableFsync}, nil } + func (ida *idAllocator) Replace(reader io.Reader) error { newFile := ida.db.Path() + ".bak" liveFile := ida.db.Path() @@ -92,7 +94,7 @@ func (ida *idAllocator) Replace(reader io.Reader) error { } else { _ = os.Remove(liveFile + ".sav") } - db, err := bolt.Open(liveFile, 0666, &bolt.Options{Timeout: 1 * time.Second}) + db, err := bolt.Open(liveFile, 0666, &bolt.Options{Timeout: 1 * time.Second, NoSync: !ida.fsyncEnabled}) ida.db = db return err } diff --git a/index.go b/index.go index 25b7bd5ee..83a4e8e4a 100644 --- a/index.go +++ b/index.go @@ -249,7 +249,7 @@ func (i *Index) open(idx *disco.Index) (err error) { partitionID := partitionID g.Go(func() error { - store, err := i.OpenTranslateStore(i.TranslateStorePath(partitionID), i.name, "", partitionID, i.holder.partitionN) + store, err := i.OpenTranslateStore(i.TranslateStorePath(partitionID), i.name, "", partitionID, i.holder.partitionN, i.holder.cfg.StorageConfig.FsyncEnabled) if err != nil { return errors.Wrapf(err, "opening index translate store: partition=%d", partitionID) } diff --git a/index_internal_test.go b/index_internal_test.go index a8cd51706..0bf4f6909 100644 --- a/index_internal_test.go +++ b/index_internal_test.go @@ -26,7 +26,7 @@ func mustOpenIndex(tb testing.TB, opt IndexOptions) *Index { if err != nil { panic(err) } - h := NewHolder(path, nil) + h := NewHolder(path, mustHolderConfig()) index, err := h.CreateIndex("i", opt) testhook.Cleanup(tb, func() { h.Close() diff --git a/index_test.go b/index_test.go index 149133108..890d481d1 100644 --- a/index_test.go +++ b/index_test.go @@ -261,7 +261,7 @@ func TestIndex_InvalidName(t *testing.T) { if err != nil { panic(err) } - index, err := pilosa.NewIndex(pilosa.NewHolder(path, nil), path, "ABC") + index, err := pilosa.NewIndex(pilosa.NewHolder(path, mustHolderConfig()), path, "ABC") if err == nil { t.Fatalf("should have gotten an error on index name with caps") } diff --git a/internal/clustertests/pause_node_test.go b/internal/clustertests/pause_node_test.go index e9c4e822a..2de330a74 100644 --- a/internal/clustertests/pause_node_test.go +++ b/internal/clustertests/pause_node_test.go @@ -168,7 +168,7 @@ func openTranslateStores(dirPath, index string) (map[int]pilosa.TranslateStore, return nil, err } // open bolt db - ts, err := boltdb.OpenTranslateStore(filePath, index, "", partition, topology.DefaultPartitionN) + ts, err := boltdb.OpenTranslateStore(filePath, index, "", partition, topology.DefaultPartitionN, false) ts.SetReadOnly(true) if err != nil { return nil, err diff --git a/server.go b/server.go index 93d9947a0..fdca4dc39 100644 --- a/server.go +++ b/server.go @@ -339,6 +339,9 @@ func OptServerOpenTranslateReader(fn OpenTranslateReaderFunc) ServerOption { func OptServerStorageConfig(cfg *storage.Config) ServerOption { return func(s *Server) error { s.holderConfig.StorageConfig = cfg + // For historical reasons, RBF's config can ignore the storage config + // in some cases. + s.holderConfig.RBFConfig.FsyncEnabled = s.holderConfig.StorageConfig.FsyncEnabled return nil } } diff --git a/server_internal_test.go b/server_internal_test.go index 7fbddf73e..7a1b7b121 100644 --- a/server_internal_test.go +++ b/server_internal_test.go @@ -19,6 +19,7 @@ import ( "testing" "time" + "github.com/molecula/featurebase/v2/storage" "github.com/molecula/featurebase/v2/testhook" ) @@ -45,8 +46,9 @@ func TestMonitorAntiEntropyZero(t *testing.T) { if err != nil { t.Fatalf("getting temp dir: %v", err) } + cfg := &storage.Config{FsyncEnabled: false, Backend: storage.DefaultBackend} s, err := NewServer(OptServerDataDir(td), - OptServerAntiEntropyInterval(0)) + OptServerAntiEntropyInterval(0), OptServerStorageConfig(cfg)) if err != nil { t.Fatalf("making new server: %v", err) } diff --git a/test/holder.go b/test/holder.go index 079495488..6eb63efac 100644 --- a/test/holder.go +++ b/test/holder.go @@ -38,7 +38,10 @@ func NewHolder(tb testing.TB) *Holder { panic(err) } - h := &Holder{Holder: pilosa.NewHolder(path, nil)} + cfg := pilosa.DefaultHolderConfig() + cfg.StorageConfig.FsyncEnabled = false + cfg.RBFConfig.FsyncEnabled = false + h := &Holder{Holder: pilosa.NewHolder(path, cfg)} return h } diff --git a/test/index.go b/test/index.go index 608bcae1c..46bc7399b 100644 --- a/test/index.go +++ b/test/index.go @@ -33,7 +33,10 @@ func newIndex(tb testing.TB) *Index { if err != nil { panic(err) } - h := pilosa.NewHolder(path, pilosa.DefaultHolderConfig()) + cfg := pilosa.DefaultHolderConfig() + cfg.StorageConfig.FsyncEnabled = false + cfg.RBFConfig.FsyncEnabled = false + h := pilosa.NewHolder(path, cfg) testhook.Cleanup(tb, func() { h.Close() }) diff --git a/translate.go b/translate.go index f449d75be..256dae341 100644 --- a/translate.go +++ b/translate.go @@ -197,7 +197,7 @@ TranslatorSummary{ } // OpenTranslateStoreFunc represents a function for instantiating and opening a TranslateStore. -type OpenTranslateStoreFunc func(path, index, field string, partitionID, partitionN int) (TranslateStore, error) +type OpenTranslateStoreFunc func(path, index, field string, partitionID, partitionN int, fsyncEnabled bool) (TranslateStore, error) // GenerateNextPartitionedID returns the next ID within the same partition. func GenerateNextPartitionedID(index string, prev uint64, partitionID, partitionN int) uint64 { @@ -407,7 +407,7 @@ var _ OpenTranslateStoreFunc = OpenInMemTranslateStore // OpenInMemTranslateStore returns a new instance of InMemTranslateStore. // Implements OpenTranslateStoreFunc. -func OpenInMemTranslateStore(rawurl, index, field string, partitionID, partitionN int) (TranslateStore, error) { +func OpenInMemTranslateStore(rawurl, index, field string, partitionID, partitionN int, fsyncEnabled bool) (TranslateStore, error) { return NewInMemTranslateStore(index, field, partitionID, partitionN), nil } diff --git a/utils_internal_test.go b/utils_internal_test.go index a3a6af2a7..76c3aa66f 100644 --- a/utils_internal_test.go +++ b/utils_internal_test.go @@ -117,7 +117,7 @@ func NewTestClusterWithReplication(tb testing.TB, nNodes, nReplicas, partitionN } // holder - h := NewHolder(path, nil) + h := NewHolder(path, mustHolderConfig()) // cluster availableShardFileFlushDuration.Set(100 * time.Millisecond) diff --git a/view_internal_test.go b/view_internal_test.go index 9258cdd27..e7d2036d8 100644 --- a/view_internal_test.go +++ b/view_internal_test.go @@ -35,7 +35,7 @@ func mustOpenView(tb testing.TB, index, field, name string) *view { CacheSize: DefaultCacheSize, } - h := NewHolder(path, nil) + h := NewHolder(path, mustHolderConfig()) // h needs an *Index so we can call h.Index() and get Index.Txf, in TestView_DeleteFragment cim := &CreateIndexMessage{