Merge pull request #1853 from molecula/fb-1115-rip-rowcache

FB-1115 rip out rowcache
This commit is contained in:
Matthew Jaffee 2022-01-11 14:10:05 -06:00 committed by GitHub
commit afc1cd62b7
No known key found for this signature in database
GPG key ID: 4AEE18F83AFDEB23
10 changed files with 8 additions and 203 deletions

View file

@ -567,37 +567,6 @@ func (p uint64Slice) Swap(i, j int) { p[i], p[j] = p[j], p[i] }
func (p uint64Slice) Len() int { return len(p) }
func (p uint64Slice) Less(i, j int) bool { return p[i] < p[j] }
// 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
// node to run out of memory.
type simpleCache struct {
cache map[uint64]*Row
}
// Fetch retrieves the bitmap at the id in the cache.
func (s *simpleCache) Fetch(id uint64) (*Row, bool) {
m, ok := s.cache[id]
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) {
if b != nil {
s.cache[id] = b
} else {
delete(s.cache, id)
}
}
// nopCache represents a no-op Cache implementation.
type nopCache struct {
stats stats.StatsClient

View file

@ -84,9 +84,6 @@ func BuildServerFlags(cmd *cobra.Command, srv *server.Command) {
flags.StringVar(&srv.Config.Storage.Backend, "storage.backend", storage.DefaultBackend, fmt.Sprintf("transaction/storage to use: one of roaring or rbf. The default is: %v. The env var PILOSA_STORAGE_BACKEND is over-ridden by --storage.backend option on the command line.", storage.DefaultBackend))
flags.BoolVar(&srv.Config.Storage.FsyncEnabled, "storage.fsync", true, "enable fsync fully safe flush-to-disk")
// RowcacheOn
flags.BoolVar((&srv.Config.RowcacheOn), "rowcache-on", srv.Config.RowcacheOn, "Do not use, permanently disabled. Flag exists for backwards compatibility and will be removed.")
// RBF specific flags. See pilosa/rbf/cfg/cfg.go for definitions.
srv.Config.RBFConfig.DefineFlags(flags)

View file

@ -34,7 +34,6 @@ import (
"github.com/molecula/featurebase/v2/roaring"
"github.com/molecula/featurebase/v2/shardwidth"
"github.com/molecula/featurebase/v2/stats"
"github.com/molecula/featurebase/v2/storage"
"github.com/molecula/featurebase/v2/testhook"
"github.com/molecula/featurebase/v2/topology"
"github.com/molecula/featurebase/v2/tracing"
@ -163,9 +162,6 @@ type fragment struct {
CacheSize uint32
// Cache containing full rows (not just counts).
rowCache *simpleCache
// Cached checksums for each block.
checksums map[int][]byte
@ -427,13 +423,8 @@ func (f *fragment) inspectStorage(data []byte, file *os.File, newGen generation,
// logic is now mostly in importStorage (reading in a bitmap) and applyStorage
// (remapping an existing bitmap to match a new backing store).
func (f *fragment) openStorage(unmarshalData bool) error {
useRowCache := storage.RowCacheEnabled()
if !f.idx.NeedsSnapshot() {
f.gen = &NopGeneration{}
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.
@ -447,9 +438,7 @@ func (f *fragment) openStorage(unmarshalData bool) error {
// unmarshal this data in order to have any.
unmarshalData = true
}
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
@ -612,24 +601,10 @@ func (f *fragment) mustRow(tx Tx, rowID uint64) *Row {
// unprotectedRow returns a row from the row cache if available or from storage
// (updating the cache).
func (f *fragment) unprotectedRow(tx Tx, rowID uint64) (*Row, error) {
useRowCache := storage.RowCacheEnabled()
if useRowCache {
if f.rowCache == nil {
f.rowCache = newSimpleCache()
}
r, ok := f.rowCache.Fetch(rowID)
if ok && r != nil {
return r, nil
}
}
row, err := f.rowFromStorage(tx, rowID)
if err != nil {
return nil, err
}
if useRowCache {
f.rowCache.Add(rowID, row)
}
return row, nil
}
@ -742,11 +717,6 @@ func (f *fragment) unprotectedSetBit(tx Tx, rowID, columnID uint64) (changed boo
}
f.cache.Add(rowID, n)
}
// Drop the rowCache entry; it's wrong, and we don't want to force
// a new copy if no one's reading it.
if storage.RowCacheEnabled() && f.rowCache != nil {
f.rowCache.Add(rowID, nil)
}
f.stats.Count(MetricSetBit, 1, 1.0)
@ -807,11 +777,6 @@ func (f *fragment) unprotectedClearBit(tx Tx, rowID, columnID uint64) (changed b
}
f.cache.Add(rowID, n)
}
// Drop the rowCache entry; it's wrong, and we don't want to force
// a new copy if no one's reading it.
if storage.RowCacheEnabled() && f.rowCache != nil {
f.rowCache.Add(rowID, nil)
}
f.stats.Count(MetricClearBit, 1, 1.0)
@ -877,11 +842,6 @@ func (f *fragment) unprotectedSetRow(tx Tx, row *Row, rowID uint64) (changed boo
}
}
// invalidate rowCache for this row.
if storage.RowCacheEnabled() && f.rowCache != nil {
f.rowCache.Add(rowID, nil)
}
// Snapshot storage.
f.holder.SnapshotQueue.Enqueue(f)
f.stats.Count("setRow", 1, 1.0)
@ -929,9 +889,6 @@ func (f *fragment) unprotectedClearRow(tx Tx, rowID uint64) (changed bool, err e
// Clear the row in cache.
f.cache.Add(rowID, 0)
if storage.RowCacheEnabled() && f.rowCache != nil {
f.rowCache.Add(rowID, nil)
}
// Snapshot storage.
f.holder.SnapshotQueue.Enqueue(f)
@ -2416,7 +2373,6 @@ func (f *fragment) importPositions(tx Tx, set, clear []uint64, rowSet map[uint64
if f.storage != nil {
wp = &f.storage.OpWriter
}
useRowCache := storage.RowCacheEnabled()
doFunc := func() error {
if len(set) > 0 {
@ -2457,9 +2413,6 @@ func (f *fragment) importPositions(tx Tx, set, clear []uint64, rowSet map[uint64
f.cache.BulkAdd(rowID, n)
}
if useRowCache && f.rowCache != nil {
f.rowCache.Add(rowID, nil)
}
}
if f.CacheType != CacheTypeNone {
@ -3328,14 +3281,8 @@ func (f *fragment) intRowIterator(tx Tx, wrap bool, filters ...roaring.BitmapFil
// accumulator [column ID] -> [int value]
acc := make(map[uint64]int64)
if storage.RowCacheEnabled() {
// 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()
}
f.mu.RLock()
defer f.mu.RUnlock()
callback := func(rid uint64) error {
// skip exist(0) and sign(1) rows
if rid == bsiExistsBit || rid == bsiSignBit {

View file

@ -115,59 +115,6 @@ func TestFragment_ClearBit(t *testing.T) {
}
}
/* We suspect this test is no longer valid under the new Tx
framework in which we always copy mmap-ed rows before
returning them. So we will comment it out for now.
If someone knows any reason for this to stick around,
let us know; we couldn't figure out how to adapt
to do a meaningful test under Tx. - jaten / tgruben
// What about rowcache timing.
func TestFragment_RowcacheMap(t *testing.T) {
var done int64
f, _, tx := mustOpenFragment(t, "i", "f", viewStandard, 0, "")
// Under -race, this test turns out to take a fairly long time
// to run with larger OpN, because we write 50,000 bits to
// the bitmap, and everything is being race-detected, and we don't
// actually need that many to get the result we care about.
f.MaxOpN = 2000
defer f.Clean(t) // failing here with TestFragment_RowcacheMap: fragment_internal_test.go:2859: fragment /var/folders/2x/hm9gp5ys3k9gmm5f_vzm_6wc0000gn/T/pilosa-fragment-001943331: unmarshalled bitmap different: differing containers for key 0: <run container, N=10131, len 2000x interval> vs <run container, N=12000, len 2000x interval>
ch := make(chan struct{})
for i := 0; i < f.MaxOpN; i++ {
_, _ = f.setBit(tx, 0, uint64(i*32))
}
// force snapshot so we get a mmapped row...
_ = f.Snapshot()
row := f.mustRow(tx, 0)
tx.Commit(0)
segment := row.Segments()[0]
bitmap := segment.data
// request information from the frozen bitmap we got back
go func() {
for atomic.LoadInt64(&done) == 0 {
for i := 0; i < f.MaxOpN; i++ {
_ = bitmap.Contains(uint64(i * 32))
}
}
close(ch)
}()
// modify the original bitmap, until it causes a snapshot, which
// then invalidates the other map...
for j := 0; j < 5; j++ {
for i := 0; i < f.MaxOpN; i++ {
_, _ = f.setBit(tx, 0, uint64(i*32+j+1))
}
}
atomic.StoreInt64(&done, 1)
<-ch
}
*/
// Ensure a fragment can clear a row.
func TestFragment_ClearRow(t *testing.T) {
f, idx, tx := mustOpenFragment(t, "i", "f", viewStandard, 0, "")

View file

@ -149,9 +149,6 @@ type HolderOpts struct {
// StorageBackend controls the tx/storage engine we instatiate. Set by
// server.go OptServerStorageConfig
StorageBackend string
// RowcacheOn, if true, turns on the row cache for all storage backends.
RowcacheOn bool
}
func (h *Holder) StartTransaction(ctx context.Context, id string, timeout time.Duration, exclusive bool) (*Transaction, error) {
@ -213,7 +210,6 @@ type HolderConfig struct {
CacheFlushInterval time.Duration
StatsClient stats.StatsClient
Logger logger.Logger
RowcacheOn bool
StorageConfig *storage.Config
RBFConfig *rbfcfg.Config
@ -273,7 +269,7 @@ func NewHolder(path string, cfg *HolderConfig) *Holder {
sharder: cfg.Sharder,
schemator: cfg.Schemator,
Logger: cfg.Logger,
Opts: HolderOpts{StorageBackend: cfg.StorageConfig.Backend, RowcacheOn: cfg.RowcacheOn},
Opts: HolderOpts{StorageBackend: cfg.StorageConfig.Backend},
SnapshotQueue: defaultSnapshotQueue,
@ -284,8 +280,6 @@ func NewHolder(path string, cfg *HolderConfig) *Holder {
indexes: make(map[string]*Index),
}
storage.SetRowCacheOn(cfg.RowcacheOn)
txf, err := NewTxFactory(cfg.StorageConfig.Backend, h.IndexesPath(), h)
vprint.PanicOn(err)
h.txf = txf

View file

@ -7,10 +7,8 @@ import (
"io"
"math"
"os"
"unsafe"
"github.com/molecula/featurebase/v2/roaring"
"github.com/molecula/featurebase/v2/storage"
"github.com/pkg/errors"
)
@ -162,8 +160,8 @@ func intoContainer(l leafCell, tx *Tx, replacing *roaring.Container, target []by
orig := l.Data
var cpMaybe []byte
var mapped bool
if storage.RowCacheEnabled() || tx.db.cfg.DoAllocZero {
// make a copy, otherwise the rowCache will see corrupted data
if tx.db.cfg.DoAllocZero {
// make a copy so no one will see corrupted data
// or mmapped data that may disappear.
cpMaybe = target[:len(orig)]
copy(cpMaybe, orig)
@ -179,10 +177,6 @@ func intoContainer(l leafCell, tx *Tx, replacing *roaring.Container, target []by
case ContainerTypeBitmapPtr:
_, bm, _ := tx.leafCellBitmap(toPgno(cpMaybe))
cloneMaybe := bm
if storage.RowCacheEnabled() {
cloneMaybe = (*[1024]uint64)(unsafe.Pointer(&target[0]))[:1024]
copy(cloneMaybe, bm)
}
c = roaring.RemakeContainerBitmap(replacing, cloneMaybe)
case ContainerTypeBitmap:
c = roaring.RemakeContainerBitmapN(replacing, toArray64(cpMaybe), int32(l.BitN))
@ -205,8 +199,8 @@ func toContainer(l leafCell, tx *Tx) (c *roaring.Container) {
orig := l.Data
var cpMaybe []byte
var mapped bool
if storage.RowCacheEnabled() || tx.db.cfg.DoAllocZero {
// make a copy, otherwise the rowCache will see corrupted data
if tx.db.cfg.DoAllocZero {
// make a copy, otherwise someone could see corrupted data
// or mmapped data that may disappear.
cpMaybe = make([]byte, len(orig))
copy(cpMaybe, orig)
@ -222,10 +216,6 @@ func toContainer(l leafCell, tx *Tx) (c *roaring.Container) {
case ContainerTypeBitmapPtr:
_, bm, _ := tx.leafCellBitmap(toPgno(cpMaybe))
cloneMaybe := bm
if storage.RowCacheEnabled() {
cloneMaybe = make([]uint64, len(bm))
copy(cloneMaybe, bm)
}
c = roaring.NewContainerBitmap(-1, cloneMaybe)
case ContainerTypeBitmap:
c = roaring.NewContainerBitmap(-1, toArray64(cpMaybe))

View file

@ -341,15 +341,6 @@ func OptServerStorageConfig(cfg *storage.Config) ServerOption {
}
}
// OptServerRowcacheOn is a functional option on Server
// used to turn on the row cache.
func OptServerRowcacheOn(rowcacheOn bool) ServerOption {
return func(s *Server) error {
s.holderConfig.RowcacheOn = rowcacheOn
return nil
}
}
// OptServerRBFConfig conveys the RBF flags to the Holder.
func OptServerRBFConfig(cfg *rbfcfg.Config) ServerOption {
return func(s *Server) error {

View file

@ -203,11 +203,6 @@ type Config struct {
// "rbf".
Storage *storage.Config `toml:"storage"`
// RowcacheOn permanently disabled. No longer useful w/ RBF. Left
// for backward compatibility but will be removed in a future
// version.
RowcacheOn bool `toml:"rowcache-on"`
// RBFConfig defines all externally configurable RBF flags.
RBFConfig *rbfcfg.Config `toml:"rbf"`

View file

@ -485,7 +485,6 @@ func (m *Command) SetupServer() error {
pilosa.OptServerClusterName(m.Config.Cluster.Name),
pilosa.OptServerSerializer(proto.Serializer{}),
pilosa.OptServerStorageConfig(m.Config.Storage),
pilosa.OptServerRowcacheOn(false),
pilosa.OptServerRBFConfig(m.Config.RBFConfig),
pilosa.OptServerMaxQueryMemory(m.Config.MaxQueryMemory),
pilosa.OptServerQueryHistoryLength(m.Config.QueryHistoryLength),

View file

@ -1,24 +0,0 @@
// Copyright 2021 Molecula Corp. All rights reserved.
package storage
import (
"sync/atomic"
)
// if enableRowCache, then we must not return mmap-ed memory
// directly, but only a copy.
var enableRowcache int64 = 1
// SetRowCacheOn should only be called in NewHolder before
// all other reads.
func SetRowCacheOn(on bool) {
if on {
atomic.StoreInt64(&enableRowcache, 1)
} else {
atomic.StoreInt64(&enableRowcache, 0)
}
}
func RowCacheEnabled() bool {
return atomic.LoadInt64(&enableRowcache) == 1
}