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

Merge pull request #1720 from molecula/fsync

Browse source 
Fsync
This commit is contained in:
seebs 2021-10-01 10:58:13 -05:00 • committed by GitHub
commit 8ed922d30e
No known key found for this signature in database
GPG key ID: 4AEE18F83AFDEB23
34 changed files with 116 additions and 64 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

10
bolt.go
View file

@ -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
}

3
bolt_test.go
View file

@ -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)

26
boltdb/translate.go
View file

@ -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)
}

2
boltdb/translate_test.go
View file

@ -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
}

5
cluster_internal_test.go
View file

@ -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() {

1
dbshard_internal_test.go
View file

@ -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()

2
etcd/embed.go
View file

@ -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 {

2
executor_internal_test.go
View file

@ -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{

2
executor_test.go
View file

@ -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()

2
field.go
View file

@ -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")
}

2
field_internal_test.go
View file

@ -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())

6
field_test.go
View file

@ -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)
}

4
fragment_internal_test.go
View file

@ -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()
})

4
go.mod
View file

@ -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

11
go.sum
View file

@ -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=

6
holder.go
View file

@ -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")
}

3
holder_internal_test.go
View file

@ -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

17
holder_test.go
View file

@ -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)

14
idalloc.go
View file

@ -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
}

2
idalloc_test.go
View file

@ -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

2
index.go
View file

@ -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)
}

2
index_internal_test.go
View file

@ -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()

2
index_test.go
View file

@ -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")
}

2
internal/clustertests/pause_node_test.go
View file

@ -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

2
rbf/db.go
View file

@ -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
}

3
server.go
View file

@ -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
}
}

4
server_internal_test.go
View file

@ -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)
}

2
test/cluster.go
View file

@ -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,
}),
),
}

19
test/disco.go
View file

@ -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

5
test/holder.go
View file

@ -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
}

5
test/index.go
View file

@ -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()
})

4
translate.go
View file

@ -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
}

2
utils_internal_test.go
View file

@ -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)

2
view_internal_test.go
View file

@ -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{