From 062bd5c8c72d01738ef78a5daab1925e0c32f2db Mon Sep 17 00:00:00 2001 From: "Jason E. Aten" Date: Tue, 27 Oct 2020 20:53:17 +0000 Subject: [PATCH] pilosa: write lock the fragment when rowcache used - only allocate the rowcache if it is in use (avoid allocation per fragment) - when the rowcache is use, fragment.go intRowIterator must write lock the fragment because the f.rowCache will be updated. - eliminate unused bitmapCache interface to keep the linter happy. - fixes #1035 --- cache.go | 14 ++++----- fragment.go | 65 +++++++++++++++++++++++++++++---------- fragment_internal_test.go | 1 + txfactory.go | 12 ++++++++ 4 files changed, 69 insertions(+), 23 deletions(-) diff --git a/cache.go b/cache.go index b1208f7ea..b03e4951c 100644 --- a/cache.go +++ b/cache.go @@ -607,13 +607,7 @@ func (p uint64Slice) merge(other []uint64) []uint64 { return ret } -// bitmapCache provides an interface for caching full bitmaps. -type bitmapCache interface { - Fetch(id uint64) (*Row, bool) - Add(id uint64, b *Row) -} - -// simpleCache implements BitmapCache +// simpleCache implements a bitmap Rowcache. // it is meant to be a short-lived cache for cases where writes are continuing to access // the same row within a short time frame (i.e. good for write-heavy loads) // A read-heavy use case would cause the cache to get bigger, potentially causing the @@ -628,6 +622,12 @@ func (s *simpleCache) Fetch(id uint64) (*Row, bool) { return m, ok } +func newSimpleCache() *simpleCache { + return &simpleCache{ + cache: make(map[uint64]*Row), + } +} + // Add adds the bitmap to the cache, keyed on the id. A nil row means // deleting the row from the cache. func (s *simpleCache) Add(id uint64, b *Row) { diff --git a/fragment.go b/fragment.go index 33d0f3bfd..734d8157d 100644 --- a/fragment.go +++ b/fragment.go @@ -144,7 +144,7 @@ type fragment struct { CacheSize uint32 // Cache containing full rows (not just counts). - rowCache bitmapCache + rowCache *simpleCache // Cached checksums for each block. checksums map[int][]byte @@ -396,9 +396,12 @@ func (f *fragment) inspectStorage(data []byte, file *os.File, newGen generation, // (remapping an existing bitmap to match a new backing store). func (f *fragment) openStorage(unmarshalData bool) error { + useRowCache := f.idx.Txf().UseRowCache() if !f.idx.NeedsSnapshot() { f.gen = &NopGeneration{} - f.rowCache = &simpleCache{make(map[uint64]*Row)} + if useRowCache { + f.rowCache = newSimpleCache() + } f.currdata = struct{ from, to uintptr }{} f.prevdata = f.currdata return nil // openStorage becomes a noop under RBF, Badger, etc. @@ -412,7 +415,9 @@ func (f *fragment) openStorage(unmarshalData bool) error { // unmarshal this data in order to have any. unmarshalData = true } - f.rowCache = &simpleCache{make(map[uint64]*Row)} + if useRowCache { + f.rowCache = newSimpleCache() + } var storageOp func([]byte, *os.File, generation, bool) (bool, error) if f.holder.Opts.Inspect { // note that this will unmarshal even if we already have @@ -567,6 +572,9 @@ func (f *fragment) unprotectedRow(tx Tx, rowID uint64) (*Row, error) { useRowCache := tx.UseRowCache() if useRowCache { + if f.rowCache == nil { + f.rowCache = newSimpleCache() + } r, ok := f.rowCache.Fetch(rowID) if ok && r != nil { return r, nil @@ -693,7 +701,9 @@ func (f *fragment) unprotectedSetBit(tx Tx, rowID, columnID uint64) (changed boo } // Drop the rowCache entry; it's wrong, and we don't want to force // a new copy if no one's reading it. - f.rowCache.Add(rowID, nil) + if tx.UseRowCache() && f.rowCache != nil { + f.rowCache.Add(rowID, nil) + } f.stats.Count(MetricSetBit, 1, 1.0) @@ -756,7 +766,9 @@ func (f *fragment) unprotectedClearBit(tx Tx, rowID, columnID uint64) (changed b } // Drop the rowCache entry; it's wrong, and we don't want to force // a new copy if no one's reading it. - f.rowCache.Add(rowID, nil) + if tx.UseRowCache() && f.rowCache != nil { + f.rowCache.Add(rowID, nil) + } f.stats.Count(MetricClearBit, 1, 1.0) @@ -823,7 +835,9 @@ func (f *fragment) unprotectedSetRow(tx Tx, row *Row, rowID uint64) (changed boo } // invalidate rowCache for this row. - f.rowCache.Add(rowID, nil) + if tx.UseRowCache() && f.rowCache != nil { + f.rowCache.Add(rowID, nil) + } // Snapshot storage. f.holder.SnapshotQueue.Enqueue(f) @@ -872,7 +886,9 @@ func (f *fragment) unprotectedClearRow(tx Tx, rowID uint64) (changed bool, err e // Clear the row in cache. f.cache.Add(rowID, 0) - f.rowCache.Add(rowID, nil) + if tx.UseRowCache() && f.rowCache != nil { + f.rowCache.Add(rowID, nil) + } // Snapshot storage. f.holder.SnapshotQueue.Enqueue(f) @@ -2293,6 +2309,8 @@ func (f *fragment) importPositions(tx Tx, set, clear []uint64, rowSet map[uint64 if f.storage != nil { wp = &f.storage.OpWriter } + useRowCache := tx.UseRowCache() + doFunc := func() error { if len(set) > 0 { f.stats.Count(MetricImportingN, int64(len(set)), 1) @@ -2334,8 +2352,9 @@ func (f *fragment) importPositions(tx Tx, set, clear []uint64, rowSet map[uint64 f.cache.BulkAdd(rowID, n) } - - f.rowCache.Add(rowID, nil) + if useRowCache && f.rowCache != nil { + f.rowCache.Add(rowID, nil) + } } if f.CacheType != CacheTypeNone { @@ -2465,8 +2484,10 @@ func (f *fragment) importValueSmallWrite(tx Tx, columnIDs []uint64, values []int return errors.Wrap(err, "importing positions") } - // Reset the rowCache. - f.rowCache = &simpleCache{make(map[uint64]*Row)} + if tx.UseRowCache() { + // Reset the rowCache. + f.rowCache = newSimpleCache() + } return nil } @@ -2515,8 +2536,10 @@ func (f *fragment) importValue(tx Tx, columnIDs []uint64, values []int64, bitDep f.opN += totalChanges f.ops++ - // Reset the rowCache. - f.rowCache = &simpleCache{make(map[uint64]*Row)} + if tx.UseRowCache() { + // Reset the rowCache. + f.rowCache = newSimpleCache() + } // in theory, this should probably have been queued anyway, but if enough // of the bits matched existing bits, we'll be under our opN estimate, and @@ -2541,6 +2564,8 @@ func (f *fragment) importRoaring(ctx context.Context, tx Tx, data []byte, clear func (f *fragment) unprotectedImportRoaring(ctx context.Context, tx Tx, data []byte, clear bool) error { rowSize := uint64(1 << shardVsContainerExponent) span, ctx := tracing.StartSpanFromContext(ctx, "importRoaring.ImportRoaringBits") + + useRowCache := tx.UseRowCache() var changed int var rowSet map[uint64]int var wp *io.Writer @@ -2570,7 +2595,9 @@ func (f *fragment) unprotectedImportRoaring(ctx context.Context, tx Tx, data []b if changes == 0 { continue } - f.rowCache.Add(rowID, nil) + if useRowCache && f.rowCache != nil { + f.rowCache.Add(rowID, nil) + } if updateCache { anyChanged = true if changes < 0 { @@ -3208,8 +3235,14 @@ func (f *fragment) intRowIterator(tx Tx, wrap bool, filters ...rowFilter) (rowIt // accumulator [column ID] -> [int value] acc := make(map[uint64]int64) - f.mu.RLock() - defer f.mu.RUnlock() + if tx.UseRowCache() { + // needs a write lock since it will update the f.rowCache + f.mu.Lock() + defer f.mu.Unlock() + } else { + f.mu.RLock() + defer f.mu.RUnlock() + } if err := f.foreachRow(tx, filters, func(rid uint64) error { // skip exist(0) and sign(1) rows if rid == bsiExistsBit || rid == bsiSignBit { diff --git a/fragment_internal_test.go b/fragment_internal_test.go index addead151..f96ecafaa 100644 --- a/fragment_internal_test.go +++ b/fragment_internal_test.go @@ -432,6 +432,7 @@ func TestFragment_SetValue(t *testing.T) { f, idx, tx := mustOpenFragment(t, "i", "f", viewStandard, 0, "") _ = idx defer f.Clean(t) + defer tx.Rollback() // Set value. if changed, err := f.setValue(tx, 100, 10, 20); err != nil { diff --git a/txfactory.go b/txfactory.go index 0d75ecbc0..c0ba21f0d 100644 --- a/txfactory.go +++ b/txfactory.go @@ -27,6 +27,7 @@ import ( "text/tabwriter" "github.com/pilosa/pilosa/v2/hash" + "github.com/pilosa/pilosa/v2/rbf" "github.com/pilosa/pilosa/v2/roaring" "github.com/pilosa/pilosa/v2/txkey" "github.com/pkg/errors" @@ -522,6 +523,8 @@ func NewTxFactory(txsrc string, holderDir string, holder *Holder) (f *TxFactory, if len(types) == 2 { f.blueGreenReg = newBlueGreenReg(types) f.isBlueGreen = true + // blue-green can never use the rowCache. + rbf.SetRowcacheOn(false) } f.dbPerShard = f.NewDBPerShard(types, holderDir, holder) @@ -538,6 +541,15 @@ func (f *TxFactory) Open() error { return f.dbPerShard.LoadExistingDBs() } +// UseRowCache can be more "global" than Tx at the moment, because +// we are sharing the same bool flag in rbf at the moment. If +// this changes then fragment.openStorage() will need a new way +// to determine if it should use the rowCache. Currently it +// doesn't have a tx Tx parameter, so we use the Txf instead. +func (f *TxFactory) UseRowCache() bool { + return rbf.EnableRowCache() +} + // Txo holds the transaction options type Txo struct { Write bool