From 3ad8ec7f0f4ddff3ff7739f93bb4eb6c084d85e8 Mon Sep 17 00:00:00 2001 From: J Date: Mon, 12 Oct 2020 12:00:15 -0500 Subject: [PATCH] rbf: add runtime options to DB struct - FsyncEnabled and DoAllocZero moved to DB struct. - deletes unused xrbrsupport.go and cmd/convert - fixes #941 --- cmd/convert/main.go | 43 ------------------- rbf.go | 8 +--- rbf/cursorx.go | 9 +--- rbf/db.go | 31 +++++++------- rbf/rbf.go | 13 ++---- rbf/tx.go | 6 +-- rbf/wal.go | 10 +++-- rbf/wal_test.go | 9 ++-- xrbrsupport.go | 100 -------------------------------------------- 9 files changed, 38 insertions(+), 191 deletions(-) delete mode 100644 cmd/convert/main.go delete mode 100644 xrbrsupport.go diff --git a/cmd/convert/main.go b/cmd/convert/main.go deleted file mode 100644 index 817bc2eb2..000000000 --- a/cmd/convert/main.go +++ /dev/null @@ -1,43 +0,0 @@ -// Copyright 2017 Pilosa Corp. -// -// Licensed under the Apache License, Version 2.0 (the "License"); -// you may not use this file except in compliance with the License. -// You may obtain a copy of the License at -// -// http://www.apache.org/licenses/LICENSE-2.0 -// -// Unless required by applicable law or agreed to in writing, software -// distributed under the License is distributed on an "AS IS" BASIS, -// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. -// See the License for the specific language governing permissions and -// limitations under the License. - -package main - -import ( - "log" - "os" - - "github.com/pilosa/pilosa/v2" - "github.com/pilosa/pilosa/v2/rbf" -) - -func main() { - - if len(os.Args) != 3 { - log.Fatal("USAGE convert srcPath destPath") - - } - holder := pilosa.NewHolder(os.Args[1], nil) - err := holder.Open() - - if err != nil { - log.Fatal(err) - - } - c := &pilosa.RBFConverter{ - Dbs: make(map[string]*rbf.DB), - Base: os.Args[2], - } - holder.ConvertToRBF(c) -} diff --git a/rbf.go b/rbf.go index 36eebf64d..cb764bc29 100644 --- a/rbf.go +++ b/rbf.go @@ -153,12 +153,8 @@ func (r *rbfDBRegistrar) OpenDBWrapper(path0 string, doAllocZero bool) (DBWrappe // creates the effect of having only one DB open per pilosa node. return w, nil } - var db *rbf.DB - if doAllocZero { - db = rbf.NewDBWithAllocZero(path) - } else { - db = rbf.NewDB(path) - } + db := rbf.NewDB(path) + db.DoAllocZero = doAllocZero w = &RbfDBWrapper{ reg: r, diff --git a/rbf/cursorx.go b/rbf/cursorx.go index 26d0c002d..64c006134 100644 --- a/rbf/cursorx.go +++ b/rbf/cursorx.go @@ -28,13 +28,6 @@ import ( // directly, but only a copy. const EnableRowCache = true -// DoAllocZero means we copy mmap read data and -// wipe it afterwards to catch retention of data -// past Tx.Rollback which was a big problem. -// This should be set by NewDBWithAllocZero and never changed -// afterwards in order to avoid a data race. -var DoAllocZero bool - //probably should just implement the container interface // but for now i'll do it func (c *Cursor) Rows() ([]uint64, error) { @@ -158,7 +151,7 @@ func toContainer(l leafCell, tx *Tx) (c *roaring.Container) { orig := l.Data var cpMaybe []byte var mapped bool - if EnableRowCache || DoAllocZero { + if EnableRowCache || tx.db.DoAllocZero { // make a copy, otherwise the rowCache will see corrupted data // or mmapped data that may disappear. cpMaybe = make([]byte, len(orig)) diff --git a/rbf/db.go b/rbf/db.go index f60b12081..b34d732f2 100644 --- a/rbf/db.go +++ b/rbf/db.go @@ -38,6 +38,8 @@ const ( MaxWALSegmentFileSize = 10 * (1 << 20) ) +// DB options like MaxSize, FsyncEnabled, DoAllocZero +// can be set before calling DB.Open(). type DB struct { data []byte // mmap data file *os.File // file descriptor @@ -58,24 +60,23 @@ type DB struct { // The maximum allowed database size. Required by mmap. MaxSize int64 -} -// NewDBWithAllocZero sets DoAllocZero true and -// returns a new instance of DB. We set it here -// to avoid a data race afterwards. -func NewDBWithAllocZero(path string) *DB { - DoAllocZero = true - return NewDB(path) + // Set before calling db.Open() + FsyncEnabled bool + + // for mmap correctness testing. + DoAllocZero bool } // NewDB returns a new instance of DB. func NewDB(path string) *DB { db := &DB{ - txs: make(map[*Tx]struct{}), - pageMap: immutable.NewMap(&uint32Hasher{}), - wcache: make([]byte, MaxWALSegmentFileSize+PageSize), - Path: path, - MaxSize: DefaultMaxSize, + txs: make(map[*Tx]struct{}), + pageMap: immutable.NewMap(&uint32Hasher{}), + wcache: make([]byte, MaxWALSegmentFileSize+PageSize), + Path: path, + MaxSize: DefaultMaxSize, + FsyncEnabled: true, } return db } @@ -169,7 +170,7 @@ func (db *DB) openWALSegments() error { continue } - segment := NewWALSegment(filepath.Join(db.WALPath(), fi.Name())) + segment := db.NewWALSegment(filepath.Join(db.WALPath(), fi.Name())) if err := segment.Open(); err != nil { _ = db.closeWALSegments() return err @@ -180,7 +181,7 @@ func (db *DB) openWALSegments() error { // Truncate everything after the last successful meta page. if walID, err := findLastWALMetaPage(db.segments); err != nil { return err - } else if db.segments, err = truncateWALAfter(db.segments, walID); err != nil { + } else if db.segments, err = db.truncateWALAfter(db.segments, walID); err != nil { return err } @@ -306,7 +307,7 @@ func (db *DB) checkpoint(exclusive bool, mu sync.Locker) error { } // Ensure WAL pages are fully copied & synced to DB file. - if err := fsync(db.file); err != nil { + if err := db.fsync(db.file); err != nil { return fmt.Errorf("db file sync: %w", err) } diff --git a/rbf/rbf.go b/rbf/rbf.go index b504e4fbc..d1f2fe4ca 100644 --- a/rbf/rbf.go +++ b/rbf/rbf.go @@ -91,11 +91,6 @@ var ( // Debug is just a temporary flag used for debugging. var Debug bool -// Testing constants. -const ( - SyncEnabled = true -) - // Magic32 returns the magic bytes as a big endian encoded uint32. func Magic32() uint32 { return binary.BigEndian.Uint32([]byte(Magic)) @@ -655,7 +650,7 @@ func RowValues(b []uint64) []uint64 { // } // truncate truncates the file at path to sz bytes. File must exist. -func truncate(path string, sz int64) error { +func (db *DB) truncate(path string, sz int64) error { f, err := os.OpenFile(path, os.O_WRONLY, 0666) if err != nil { return fmt.Errorf("open file: %w", err) @@ -664,14 +659,14 @@ func truncate(path string, sz int64) error { if err := f.Truncate(sz); err != nil { return fmt.Errorf("truncate: %w", err) - } else if err := fsync(f); err != nil { + } else if err := db.fsync(f); err != nil { return fmt.Errorf("sync: %w", err) } return f.Close() } -func fsync(f *os.File) error { - if !SyncEnabled { +func (db *DB) fsync(f *os.File) error { + if !db.FsyncEnabled { return nil } return f.Sync() diff --git a/rbf/tx.go b/rbf/tx.go index c2647b20f..153e38380 100644 --- a/rbf/tx.go +++ b/rbf/tx.go @@ -122,7 +122,7 @@ func (tx *Tx) Rollback() { // TODO(bbj): Invalidate DB if rollback fails. Possibly attempt reopen? if tx.dirty { - if _, err := truncateWALAfter(tx.segments, tx.walID); err != nil { + if _, err := tx.db.truncateWALAfter(tx.segments, tx.walID); err != nil { panicOn(err) } tx.segments = nil @@ -1610,7 +1610,7 @@ func (tx *Tx) flushWALWriter() error { // Flush cache to writer. if _, err := w.WriteAt(tx.wcache, int64(segment.PageN)*PageSize); err != nil { return fmt.Errorf("write wal segment: %w", err) - } else if err := fsync(w); err != nil { + } else if err := tx.db.fsync(w); err != nil { return fmt.Errorf("sync wal segment: %w", err) } else if err := w.Close(); err != nil { return fmt.Errorf("close wal segment: %w", err) @@ -1686,7 +1686,7 @@ func (tx *Tx) ensureWritableWALSegment() error { } // Create new segment file. - s := NewWALSegment(filepath.Join(tx.db.WALPath(), FormatWALSegmentPath(base))) + s := tx.db.NewWALSegment(filepath.Join(tx.db.WALPath(), FormatWALSegmentPath(base))) if err := s.Open(); err != nil { return fmt.Errorf("add wal segment: %w", err) } diff --git a/rbf/wal.go b/rbf/wal.go index 14e64a107..f6787cfca 100644 --- a/rbf/wal.go +++ b/rbf/wal.go @@ -26,6 +26,7 @@ import ( // WALSegment represents a single file in the WAL. type WALSegment struct { + db *DB Path string // path to file MinWALID int64 // base WALID; calculated from path PageN int // number of written pages @@ -34,8 +35,9 @@ type WALSegment struct { } // NewWALSegment returns a new instance of WALSegment for a given path. -func NewWALSegment(path string) WALSegment { +func (db *DB) NewWALSegment(path string) WALSegment { return WALSegment{ + db: db, Path: path, } } @@ -74,7 +76,7 @@ func (s *WALSegment) Open() (err error) { s.PageN = int(sz / PageSize) if sz%PageSize != 0 { sz = int64(s.PageN * PageSize) - if err := truncate(s.Path, sz); err != nil { + if err := s.db.truncate(s.Path, sz); err != nil { return fmt.Errorf("truncate wal file: %w", err) } } @@ -212,7 +214,7 @@ func findLastWALMetaPage(segments []WALSegment) (walID int64, err error) { } // truncateWALAfter removes all pages in the WAL after walID. -func truncateWALAfter(segments []WALSegment, walID int64) ([]WALSegment, error) { +func (db *DB) truncateWALAfter(segments []WALSegment, walID int64) ([]WALSegment, error) { var newSegments []WALSegment for i := range segments { @@ -229,7 +231,7 @@ func truncateWALAfter(segments []WALSegment, walID int64) ([]WALSegment, error) newSegment := *segment newSegment.PageN = int((walID - newSegment.MinWALID) + 1) - if err := truncate(newSegment.Path, int64(newSegment.PageN)*PageSize); err != nil { + if err := db.truncate(newSegment.Path, int64(newSegment.PageN)*PageSize); err != nil { return segments, err } newSegments = append(newSegments, newSegment) diff --git a/rbf/wal_test.go b/rbf/wal_test.go index 59463d435..257001da4 100644 --- a/rbf/wal_test.go +++ b/rbf/wal_test.go @@ -28,7 +28,10 @@ import ( func TestWALSegment_Open(t *testing.T) { t.Run("OK", func(t *testing.T) { - s := MustOpenWALSegment(t, 10) + db := MustOpenDB(t) + defer MustCloseDB(t, db) + + s := MustOpenWALSegment(t, db, 10) defer MustCloseWALSegment(t, s) if got, want := s.MinWALID, int64(10); got != want { t.Fatalf("Base()=%d, want %d", got, want) @@ -153,7 +156,7 @@ func benchmarkWALSegment_WriteWALPage(b *testing.B, flushSize int) { */ // MustOpenWALSegment opens a WAL segment in a temporary path. Fails on error. -func MustOpenWALSegment(tb testing.TB, walID int64) rbf.WALSegment { +func MustOpenWALSegment(tb testing.TB, db *rbf.DB, walID int64) rbf.WALSegment { tb.Helper() dir, err := ioutil.TempDir("", "") @@ -165,7 +168,7 @@ func MustOpenWALSegment(tb testing.TB, walID int64) rbf.WALSegment { tb.Fatal(err) } - s := rbf.NewWALSegment(path) + s := db.NewWALSegment(path) if err := s.Open(); err != nil { tb.Fatal(err) } diff --git a/xrbrsupport.go b/xrbrsupport.go deleted file mode 100644 index 70720582a..000000000 --- a/xrbrsupport.go +++ /dev/null @@ -1,100 +0,0 @@ -// Copyright 2017 Pilosa Corp. -// -// Licensed under the Apache License, Version 2.0 (the "License"); -// you may not use this file except in compliance with the License. -// You may obtain a copy of the License at -// -// http://www.apache.org/licenses/LICENSE-2.0 -// -// Unless required by applicable law or agreed to in writing, software -// distributed under the License is distributed on an "AS IS" BASIS, -// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. -// See the License for the specific language governing permissions and -// limitations under the License. -package pilosa - -import ( - "fmt" - - "github.com/pilosa/pilosa/v2/rbf" - "github.com/pilosa/pilosa/v2/roaring" -) - -type Converter interface { - Convert(index, field, view string, shard uint64, rb *roaring.Bitmap) error - Shutdown() -} - -type RBFConverter struct { - Dbs map[string]*rbf.DB - Base string -} - -func (rbc *RBFConverter) GetOrCreateDB(index string, shard uint64) (*rbf.DB, error) { - key := fmt.Sprintf("%s/%d", index, shard) - db, found := rbc.Dbs[key] - if found { - return db, nil - } - path := rbc.Base + "/" + key - db = rbf.NewDB(path) - err := db.Open() - if err != nil { - return nil, err - } - rbc.Dbs[key] = db - return db, nil -} -func (rbc *RBFConverter) Shutdown() { - for key, db := range rbc.Dbs { - fmt.Println("Shutdown", key) - db.Close() - - } - -} -func (rbc *RBFConverter) Convert(index, field, view string, shard uint64, rb *roaring.Bitmap) error { - fmt.Println("CONVERT", index, field, view, shard) - db, err := rbc.GetOrCreateDB(index, shard) - if err != nil { - return err - } - tx, err := db.Begin(writable) - if err != nil { - return err - } - defer tx.Rollback() - - name := fmt.Sprintf("%s/%s", field, view) - err = tx.CreateBitmap(name) - if err != nil { - return err - } - _, err = tx.AddRoaring(name, rb) - if err != nil { - return err - } - return tx.Commit() -} - -func (h *Holder) ConvertToRBF(c Converter) { - /* - for idxname, idx := range h.indexes { - for fieldName, field := range idx.fields { - for _, view := range field.views() { - for shard, fragment := range view.fragments { - panic("NEED bitmap from storage") - junk := roaring.NewBitmap() - err := c.Convert(idxname, fieldName, view.name, shard, junk) - if err != nil { - fmt.Println("ERR", err, fragment.shard) //just added shard for compile - } - } - } - - } - - } - c.Shutdown() - */ -}