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/etcd/embed.go b/etcd/embed.go index 67711d4c1..7da2c9806 100644 --- a/etcd/embed.go +++ b/etcd/embed.go @@ -63,6 +63,7 @@ type Options struct { LClientSocket []*net.TCPListener BootstrapTimeout time.Duration + UnsafeNoFsync bool `toml:"no-fsync"` } var ( @@ -228,6 +229,7 @@ func parseOptions(opt Options) *embed.Config { cfg.InitialClusterToken = opt.ClusterName cfg.BootstrapTimeout = opt.BootstrapTimeout cfg.LCUrls = types.MustNewURLs([]string{opt.LClientURL}) + cfg.UnsafeNoFsync = opt.UnsafeNoFsync if opt.AClientURL != "" { cfg.ACUrls = types.MustNewURLs([]string{opt.AClientURL}) } else { 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/go.mod b/go.mod index e590bdd1b..ff7c1d715 100644 --- a/go.mod +++ b/go.mod @@ -1,6 +1,8 @@ module github.com/molecula/featurebase/v2 -replace go.etcd.io/etcd => github.com/molecula/etcd v0.0.0-20210621160528-2cd93f1df0e7 +replace go.etcd.io/etcd => github.com/molecula/etcd v0.0.0-20210930172242-ad94b354f72c + +replace go.etcd.io/bbolt => github.com/seebs/bbolt v0.0.0-20210930181431-2ea708af0554 require ( github.com/CAFxX/gcnotifier v0.0.0-20190112062741-224a280d589d diff --git a/go.sum b/go.sum index 5f4d1be61..0959b010c 100644 --- a/go.sum +++ b/go.sum @@ -230,8 +230,8 @@ github.com/modern-go/reflect2 v1.0.1 h1:9f412s+6RmYXLWZSEzVVgPGK7C2PphHj5RJrvfx9 github.com/modern-go/reflect2 v1.0.1/go.mod h1:bx2lNnkwVCuqBIxFjflWJWanXIb3RllmbCylyMrvgv0= github.com/molecula/apophenia v0.0.0-20190827192002-68b7a14a478b h1:cZADDaNYM7xn/nklO3g198JerGQjadFuA0ofxBJgK0Y= github.com/molecula/apophenia v0.0.0-20190827192002-68b7a14a478b/go.mod h1:uXd1BiH7xLmgkhVmspdJLENv6uGWrTL/MQX2TN7Yz9s= -github.com/molecula/etcd v0.0.0-20210621160528-2cd93f1df0e7 h1:hufElvtCighE0G2VFJYDGWCY8JlmCWZ1FmXvlf25yUQ= -github.com/molecula/etcd v0.0.0-20210621160528-2cd93f1df0e7/go.mod h1:1X1h4BZ44WjM0LJof1gKKLap1OA4RsicGCDRtACTkLI= +github.com/molecula/etcd v0.0.0-20210930172242-ad94b354f72c h1:YnU+8kIrr/7IDGtIYncawklAs54EWihED4DBIm+kjAA= +github.com/molecula/etcd v0.0.0-20210930172242-ad94b354f72c/go.mod h1:1X1h4BZ44WjM0LJof1gKKLap1OA4RsicGCDRtACTkLI= github.com/mwitkow/go-conntrack v0.0.0-20161129095857-cc309e4a2223 h1:F9x/1yl3T2AeKLr2AMdilSD8+f9bvMnNN8VS5iDtovc= github.com/mwitkow/go-conntrack v0.0.0-20161129095857-cc309e4a2223/go.mod h1:qRWi+5nqEBWmkhHvq77mSJWrCKwh8bxhgT7d/eI7P4U= github.com/niemeyer/pretty v0.0.0-20200227124842-a10e7caefd8e h1:fD57ERR4JtEqsWbfPhv4DMiApHyliiK5xCTNVSPiaAs= @@ -285,6 +285,8 @@ github.com/ryanuber/columnize v0.0.0-20160712163229-9b3edd62028f/go.mod h1:sm1tb github.com/satori/go.uuid v1.2.0 h1:0uYX9dsZ2yD7q2RtLRtPSdGDWzjeM3TbMJP9utgA0ww= github.com/satori/go.uuid v1.2.0/go.mod h1:dA0hQrYB0VpLJoorglMZABFdXlWrHn1NEOzdhQKdks0= github.com/sean-/seed v0.0.0-20170313163322-e2103e2c3529/go.mod h1:DxrIzT+xaE7yg65j358z/aeFdxmN0P9QXhEzd20vsDc= +github.com/seebs/bbolt v0.0.0-20210930181431-2ea708af0554 h1:88K0ffxhVphUHxlqW4ewOaXdnJByH4LcCuvYfv0QI/M= +github.com/seebs/bbolt v0.0.0-20210930181431-2ea708af0554/go.mod h1:qXsaaIqmgQH0T+OPdb99Bf+PKfBBQVAdyD6TY9G8XM4= github.com/shirou/gopsutil/v3 v3.20.11 h1:NeVf1K0cgxsWz+N3671ojRptdgzvp7BXL3KV21R0JnA= github.com/shirou/gopsutil/v3 v3.20.11/go.mod h1:igHnfak0qnw1biGeI2qKQvu0ZkwvEkUcCLlYhZzdr/4= github.com/shurcooL/sanitized_anchor_name v1.0.0/go.mod h1:1NzhyTcUVG4SuEtjjoZeVRXNmyL/1OwPU0+IJeTBvfc= @@ -344,9 +346,6 @@ github.com/zeebo/blake3 v0.1.1 h1:Nbsts7DdKThRHHd+YNlqiGlRqGEF2bE2eXN+xQ1hsEs= github.com/zeebo/blake3 v0.1.1/go.mod h1:G9pM4qQwjRzF1/v7+vabMj/c5mWpGZ2Wzo3Eb4z0pb4= github.com/zeebo/pcg v1.0.0 h1:dt+dx+HvX8g7Un32rY9XWoYnd0NmKmrIzpHF7qiTDj0= github.com/zeebo/pcg v1.0.0/go.mod h1:09F0S9iiKrwn9rlI5yjLkmrug154/YRW6KnnXVDM/l4= -go.etcd.io/bbolt v1.3.2/go.mod h1:IbVyRI1SCnLcuJnV2u8VeU0CEYM7e686BmAb1XKL+uU= -go.etcd.io/bbolt v1.3.5 h1:XAzx9gjCb0Rxj7EoqcClPD1d5ZBxZJk0jbuoPHenBt0= -go.etcd.io/bbolt v1.3.5/go.mod h1:G5EMThwa9y8QZGBClrRx5EY+Yw9kAhnjy3bSjsnlVTQ= go.opencensus.io v0.21.0/go.mod h1:mSImk1erAIZhrmZN+AvHh14ztQfjbGwt4TtuofqLduU= go.opencensus.io v0.22.0/go.mod h1:+kGneAE2xo2IficOXnaByMWTGM9T73dGwxeWcUqIpI8= go.uber.org/atomic v1.3.2/go.mod h1:gD2HeocX3+yG+ygLZcrzQJaqmWj9AIm7n08wl/qW/PE= @@ -443,7 +442,7 @@ golang.org/x/sys v0.0.0-20190624142023-c5567b49c5d0/go.mod h1:h1NjWce9XRLGQEsW7w golang.org/x/sys v0.0.0-20191001151750-bb3f8db39f24/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs= golang.org/x/sys v0.0.0-20191005200804-aed5e4c7ecf9/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs= golang.org/x/sys v0.0.0-20191220142924-d4481acd189f/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs= -golang.org/x/sys v0.0.0-20200202164722-d101bd2416d5/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs= +golang.org/x/sys v0.0.0-20200923182605-d9f96fdee20d/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs= golang.org/x/sys v0.0.0-20200930185726-fdedc70b468f/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs= golang.org/x/sys v0.0.0-20201014080544-cc95f250f6bc/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs= golang.org/x/sys v0.0.0-20201024232916-9f70ab9862d5/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs= 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/idalloc_test.go b/idalloc_test.go index dff559146..b28a56604 100644 --- a/idalloc_test.go +++ b/idalloc_test.go @@ -40,7 +40,7 @@ func TestIDAlloc(t *testing.T) { }() // Open bolt. - db, err := bolt.Open(f.Name(), 0666, &bolt.Options{Timeout: 1 * time.Second}) + db, err := bolt.Open(f.Name(), 0666, &bolt.Options{Timeout: 1 * time.Second, NoSync: true}) if err != nil { t.Errorf("opening bolt: %v", err) return 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/rbf/db.go b/rbf/db.go index 92efffd72..f1623ded8 100644 --- a/rbf/db.go +++ b/rbf/db.go @@ -274,7 +274,7 @@ func (db *DB) Close() (err error) { // Close writer handler. if db.file != nil { - err = db.file.Sync() + err = db.fsync(db.file) if err != nil { return } 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/cluster.go b/test/cluster.go index 126245a46..137aa6c83 100644 --- a/test/cluster.go +++ b/test/cluster.go @@ -607,7 +607,7 @@ func prependTestServerOpts(opts []server.CommandOption) []server.CommandOption { pilosa.OptServerNodeDownRetries(5, 100*time.Millisecond), pilosa.OptServerStorageConfig(&storage.Config{ Backend: pilosa.CurrentBackendOrDefault(), - FsyncEnabled: true, + FsyncEnabled: false, }), ), } diff --git a/test/disco.go b/test/disco.go index 7aebd8f89..d00c06a31 100644 --- a/test/disco.go +++ b/test/disco.go @@ -111,6 +111,7 @@ func GetPortsGenConfigs(tb testing.TB, nodes []*Command) error { LPeerSocket: []*net.TCPListener{peerListener}, LClientSocket: []*net.TCPListener{clientListener}, BootstrapTimeout: 50 * time.Millisecond, + UnsafeNoFsync: true, } peerUrls[i] = fmt.Sprintf("%s=%s", name, peerURL) } @@ -144,14 +145,16 @@ func GenPortsConfig(tb testing.TB, ports []Ports) []*server.Config { BindGRPC: fmt.Sprintf(":%d", ports[i].Grpc), GRPCListener: ports[i].LsnG, Etcd: etcd.Options{ - Dir: discoDir, - LClientURL: lClientURL, - AClientURL: lClientURL, - LPeerURL: lPeerURL, - APeerURL: lPeerURL, - HeartbeatTTL: 5, - LPeerSocket: []*net.TCPListener{lsnP}, - LClientSocket: []*net.TCPListener{lsnC}, + Dir: discoDir, + LClientURL: lClientURL, + AClientURL: lClientURL, + LPeerURL: lPeerURL, + APeerURL: lPeerURL, + HeartbeatTTL: 5, + LPeerSocket: []*net.TCPListener{lsnP}, + LClientSocket: []*net.TCPListener{lsnC}, + BootstrapTimeout: 50 * time.Millisecond, + UnsafeNoFsync: true, }, } cfgs[i].Cluster.Name = clusterName 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{