diff --git a/field.go b/field.go index 899e6f399..247c8a677 100644 --- a/field.go +++ b/field.go @@ -408,7 +408,7 @@ func (f *Field) AvailableShards() *roaring.Bitmap { b := f.remoteAvailableShards.Clone() for _, view := range f.viewMap { - b = b.Union(view.availableShards()) + b.UnionInPlace(view.availableShards()) } return b } diff --git a/index.go b/index.go index 9c53083fd..a5b6245ba 100644 --- a/index.go +++ b/index.go @@ -347,7 +347,7 @@ func (i *Index) AvailableShards() *roaring.Bitmap { b := roaring.NewBitmap() for _, f := range i.fields { - b = b.Union(f.AvailableShards()) + b.UnionInPlace(f.AvailableShards()) } i.Stats.Gauge("maxShard", float64(b.Max()), 1.0) diff --git a/roaring/roaring.go b/roaring/roaring.go index c399b8df6..369e8c2a7 100644 --- a/roaring/roaring.go +++ b/roaring/roaring.go @@ -2916,11 +2916,34 @@ func (c *Container) runToBitmap() *Container { return c } bitmap := make([]uint64, bitmapN) - for _, r := range c.runs() { - // TODO this can be ~64x faster for long runs by setting maxBitmap instead of single bits - //note v must be int or will overflow - for v := int(r.start); v <= int(r.last); v++ { - bitmap[v/64] |= (uint64(1) << uint(v%64)) + for _, iv := range c.runs() { + w1, w2 := iv.start/64, iv.last/64 + b1, b2 := iv.start&63, iv.last&63 + // a mask for everything under bit X looks like + // (1 << x) - 1. Say b1 is 4; our mask will want + // to have the bottom 4 bits be zero, so we shift + // left 4, getting 10000, then subtract 1, and + // get 01111, which is the mask to *remove*. + m1 := (uint64(1) << b1) - 1 + // inclusive mask: same thing, then shift left 1 and + // or in 1. So for 4, we'd get 011111, which is the + // mask to *keep*. + m2 := (((uint64(1) << b2) - 1) << 1) | 1 + if w1 == w2 { + // If we only had bit 4 in the range, this would + // end up being 011111 &^ 01111, or 010000. + bitmap[w1] |= (m2 &^ m1) + continue + } + // for w2, the "To" field, we want to set the bottom N + // bits. For w1, the "From" word, we want to set all *but* + // the bottom N bits. + bitmap[w2] |= m2 + bitmap[w1] |= ^m1 + words := bitmap[w1+1 : w2] + // set every bit between them + for i := range words { + words[i] = ^uint64(0) } } if c.frozen() { @@ -4335,12 +4358,14 @@ func differenceRunBitmap(a, b *Container) *Container { if len(ra) > 0 && ra[0].start == 0 && ra[0].last == 65535 { return flipBitmap(b) } + bb := b.bitmap()[:1024] runs := make([]interval16, 0, len(ra)) for _, inputRun := range ra { run := inputRun add := true for bit := inputRun.start; bit <= inputRun.last; bit++ { - if b.bitmapContains(bit) { + idx, exp := int(bit>>6), bit&63 + if (bb[idx]>>exp)&1 != 0 { if run.start == bit { if bit == 65535 { //overflow add = false @@ -4352,6 +4377,10 @@ func differenceRunBitmap(a, b *Container) *Container { } else { run.last = bit - 1 if run.last >= run.start { + if len(runs) >= runMaxSize { + asBitmap := a.runToBitmap() + return differenceBitmapBitmap(asBitmap, b) + } runs = append(runs, run) } run.start = bit + 1 @@ -4368,6 +4397,10 @@ func differenceRunBitmap(a, b *Container) *Container { } if run.start <= run.last { if add { + if len(runs) >= runMaxSize { + asBitmap := a.runToBitmap() + return differenceBitmapBitmap(asBitmap, b) + } runs = append(runs, run) } } diff --git a/view.go b/view.go index 06d38dce8..2377781ec 100644 --- a/view.go +++ b/view.go @@ -23,6 +23,7 @@ import ( "strconv" "strings" "sync" + "sync/atomic" "time" "github.com/pilosa/pilosa/v2/logger" @@ -61,6 +62,9 @@ type view struct { logger logger.Logger snapshotQueue snapshotQueue remoteShardPresent func(uint64) bool + + knownShards *roaring.Bitmap + knownShardsCopied uint32 } // newView returns a new instance of View. @@ -81,11 +85,48 @@ func newView(path, index, field, name string, fieldOptions FieldOptions) *view { stats: stats.NopStatsClient, logger: logger.NopLogger, remoteShardPresent: func(uint64) bool { return false }, + knownShards: roaring.NewSliceBitmap(), } } +// addKnownShard adds a known shard to v, which you should only do when +// holding the lock -- but that's probably a given, since you're presumably +// calling it because you were potentially altering the shard list. Since +// you have the write lock, availableShards() can't be happening right now. +// Either it'll get the previous value or the next value of knownShards, +// and either is probably fine. +// +// This means that we only copy the (probably tiny) bitmap if we're +// modifying it after it's been read. If it never gets read, knownShardsCopied +// never changes. If it gets read, then we treat that one as immutable -- +// we never modify it again, because the field code might be reading it, so +// we make a fresh copy. Since shards almost never change, the expected +// behavior is that we call addKnownShard a lot during initial startup, +// when knownShardsCopied is 0, and then after that calls to availableShards +// return that bitmap, and set knownShardsCopied to 1, but we rarely modify +// the list. +func (v *view) addKnownShard(shard uint64) { + if atomic.LoadUint32(&v.knownShardsCopied) == 1 { + v.knownShards = v.knownShards.Clone() + atomic.StoreUint32(&v.knownShardsCopied, 0) + } + _, _ = v.knownShards.Add(shard) +} + +// removeKnownShard removes a known shard from v. See the notes on addKnownShard. +func (v *view) removeKnownShard(shard uint64) { + if atomic.LoadUint32(&v.knownShardsCopied) == 1 { + v.knownShards = v.knownShards.Clone() + atomic.StoreUint32(&v.knownShardsCopied, 0) + } + _, _ = v.knownShards.Remove(shard) +} + // open opens and initializes the view. func (v *view) open() error { + if v.knownShards == nil { + v.knownShards = roaring.NewSliceBitmap() + } // Never keep a cache for field views. if strings.HasPrefix(v.name, viewBSIGroupPrefix) { @@ -169,6 +210,7 @@ fileLoop: v.logger.Debugf("add index/field/view/fragment to view.fragments: %s/%s/%s/%d", v.index, v.field, v.name, shard) mu.Lock() v.fragments[frag.shard] = frag + v.addKnownShard(frag.shard) mu.Unlock() return nil }) @@ -206,6 +248,7 @@ fragLoop: } err := eg.Wait() v.fragments = make(map[uint64]*fragment) + v.knownShards = nil return err } @@ -220,14 +263,15 @@ func (v *view) flags() byte { // availableShards returns a bitmap of shards which contain data. func (v *view) availableShards() *roaring.Bitmap { + // A read lock prevents anything with the write lock from being + // active, so anything that's calling add/removeKnownShard won't + // be doing it here. But we do need to indicate that we came + // through, but we don't want to block on a write lock. So we + // use an atomic for that. v.mu.RLock() defer v.mu.RUnlock() - - b := roaring.NewBitmap() - for shard := range v.fragments { - _, _ = b.Add(shard) // ignore error, no writer attached - } - return b + atomic.StoreUint32(&v.knownShardsCopied, 1) + return v.knownShards } // fragmentPath returns the path to a fragment in the view. @@ -278,6 +322,7 @@ func (v *view) CreateFragmentIfNotExists(shard uint64) (*fragment, error) { frag.RowAttrStore = v.rowAttrStore v.fragments[shard] = frag + v.addKnownShard(shard) v.notifyIfNewShard(shard) return frag, nil } @@ -355,6 +400,7 @@ func (v *view) deleteFragment(shard uint64) error { } delete(v.fragments, shard) + v.removeKnownShard(shard) return nil }