// Copyright 2020 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. // +build amd64 package pilosa import ( "bytes" "fmt" "io" "io/ioutil" "log" "math" "os" "path/filepath" "runtime" "sort" "strings" "sync" "sync/atomic" "github.com/glycerine/lmdb-go/lmdb" "github.com/pilosa/pilosa/v2/roaring" "github.com/pilosa/pilosa/v2/txkey" "github.com/pkg/errors" ) // lmdbRegistrar facilitates shutdown // of all the lmdb databases started under // tests. Its needed because most tests don't cleanup // the *Index(es) they create. But we still // want to shutdown lmdbDB goroutines // after tests run. // // It also allows opening the same path twice to // result in sharing the same open database handle, and // thus the same transactional guarantees. // type lmdbRegistrar struct { mu sync.Mutex mp map[*LMDBWrapper]bool path2db map[string]*LMDBWrapper } var globalLMDBReg *lmdbRegistrar = newLMDBTestRegistrar() func newLMDBTestRegistrar() *lmdbRegistrar { return &lmdbRegistrar{ mp: make(map[*LMDBWrapper]bool), path2db: make(map[string]*LMDBWrapper), } } // register each lmdb created under tests, so we // can clean them up. This is called by openLMDBWrapper() while // holding the r.mu.Lock, since it needs to atomically // check the registry and make a new instance only // if one does not exist for its path, and otherwise // return the existing instance. func (r *lmdbRegistrar) unprotectedRegister(w *LMDBWrapper) { r.mp[w] = true r.path2db[w.path] = w } // unregister removes w from r func (r *lmdbRegistrar) unregister(w *LMDBWrapper) { r.mu.Lock() delete(r.mp, w) delete(r.path2db, w.path) r.mu.Unlock() } func DumpAllLMDB() { globalLMDBReg.mu.Lock() defer globalLMDBReg.mu.Unlock() for w := range globalLMDBReg.mp { _ = w AlwaysPrintf("this lmdb path='%v' has: \n%v\n", w.path, w.StringifiedLMDBKeys(nil)) } } // lmdbPath is a helper for determining the full directory // in which the lmdb database will be stored. func lmdbPath(path string) string { if !strings.HasSuffix(path, "-lmdb") { return path + "-lmdb" } return path } // openLMDBDB opens the database in the bpath directoy // without deleting any prior content. Any LMDBDB // database directory will have the "-lmdb" suffix. // // openLMDBDB will check the registry and make a new instance only // if one does not exist for its bpath. Otherwise it returns // the existing instance. This insures only one lmdbDB // per bpath in this pilosa node. func (r *lmdbRegistrar) openLMDBWrapper(path0 string) (*LMDBWrapper, error) { path := lmdbPath(path0) r.mu.Lock() defer r.mu.Unlock() w, ok := r.path2db[path] if ok { // creates the effect of having only one lmdb open per pilosa node. return w, nil } // otherwise, make a new lmdb and store it in globalLMDBReg runtime.LockOSThread() const MaxReaders = 256 // default is 126 env, err := lmdb.NewEnvMaxReaders(MaxReaders) panicOn(err) err = env.SetMaxDBs(1) panicOn(err) //err = env.SetMapSize(256 << 30) // 256GB err = env.SetMapSize(16 << 30) // 16GB panicOn(err) panicOn(os.MkdirAll(filepath.Dir(path), 0755)) flags := uint(lmdb.NoReadahead | lmdb.NoSubdir) // unsafe, but get upper bound on performance. // WriteMap = C.MDB_WRITEMAP // Use a writable memory map. // NoMetaSync = C.MDB_NOMETASYNC // Don't fsync metapage after commit. // NoSync = C.MDB_NOSYNC // Don't fsync after commit. // flags = flags | lmdb.WriteMap | lmdb.NoMetaSync | lmdb.NoSync // about the same speed // flags = flags | lmdb.NoMetaSync | lmdb.NoSync // slows things down //flags = flags | lmdb.WriteMap // seems faster than without: // kRemove N= 710401 avg/op: 7.714µs sd: 27.83µs total: 5.480656859s // kAdd N= 722835 avg/op: 9.096µs sd: 105.787µs total: 6.575497725s // ACI not ACID at the moment; no durability flags = flags | lmdb.NoMemInit | // Disable LMDB memory initialization // Note that lmdb.WriteMap requests a big, writable, memory map. // On my darwin/OSX laptop with 16GB ram, for instance, we // can have difficulty obtaining this, resulting in // panic: mdb_env_open: no space left on device lmdb.WriteMap | // Use a writable memory map. // default ACI (not Durable) transactions; 300% faster write speed results. lmdb.NoMetaSync | // Don't fsync metapage after commit. lmdb.NoSync | // Don't fsync after commit. lmdb.MapAsync // Flush asynchronously when using the WriteMap flag. err = env.Open(path, flags, 0644) if err != nil { AlwaysPrintf("error env.Open(path='%v'): '%v'; on gid = '%v'", path, err, curGID()) } panicOn(err) // In any real application it is important to check for readers that were // never closed by their owning process, and for which the owning process // has exited. See the documentation on transactions for more information. staleReaders, err := env.ReaderCheck() panicOn(err) if staleReaders > 0 { log.Printf("cleared %d reader slots from dead processes", staleReaders) } // Open a database handle that will be used for the entire lifetime of this // application. Because the database may not have existed before, and the // database may need to be created, we need to get the database handle in // an update transacation. var dbi lmdb.DBI name := filepath.Base(path) err = env.Update(func(txn *lmdb.Txn) (err error) { dbi, err = txn.CreateDBI(name) return err }) panicOn(err) w = &LMDBWrapper{ name: name, env: env, reg: r, path: path, dbi: dbi, } r.unprotectedRegister(w) w.startStack = stack() return w, nil } var ErrShutdown = fmt.Errorf("shutting down") // DeleteIndex deletes all the containers associated with // the named index from the lmdb database. func (w *LMDBWrapper) DeleteIndex(indexName string) error { // We use the apostrophie rune `'` to locate the end of the // index name in the key prefix, so we cannot allow indexNames // themselves to contain apostrophies. if strings.Contains(indexName, "'") { return fmt.Errorf("error: bad indexName `%v` in LMDBWrapper.DeleteIndex() call: indexName cannot contain apostrophes/single quotes.", indexName) } prefix := txkey.IndexOnlyPrefix(indexName) return w.DeletePrefix(prefix) } // statically confirm that LMDBTx satisfies the Tx interface. var _ Tx = (*LMDBTx)(nil) // LMDBWrapper provides the NewLMDBTx() method. // Execute lmdbJob's via LMDBWrapper.submit(); these must // be done by the lmdb goroutine worker pool. type LMDBWrapper struct { env *lmdb.Env muDb sync.Mutex path string name string dbi lmdb.DBI // track our registrar for Close / goro leak reporting purposes. reg *lmdbRegistrar // make LMDBWrapper.Close() idempotent, avoiding panic on double Close() closed bool // doAllocZero sets the corresponding flag on all new LMDBTx. // When doAllocZero is true, we zero out any data from lmdb // after transcation commit and rollback. This simulates // what would happen if we were to use the mmap-ed data // from lmdb directly. Currently we copy by default for // safety because otherwise TestAPI_ImportColumnAttrs sees // corrupted data. doAllocZero bool // stack() from our creation point, to track tests // that haven't closed us. startStack string DeleteEmptyContainer bool nextTxSn int64 } func (w *LMDBWrapper) IsClosed() bool { w.muDb.Lock() defer w.muDb.Unlock() return w.closed } // NewLMDBTx produces LMDB based ACID transactions. If // the transaction will modify data, then the write flag must be true. // Read-only queries should set write to false, to allow more concurrency. // Methods on a LMDBTx are thread-safe, and can be called from // different goroutines. // // initialIndexName is optional. It is set by the TxFactory from the Txo // options provided at the Tx creation point. It allows us to recognize // and isolate cross-index queries more quickly. It can always be empty "" // but when set is highly useful for debugging. It has no impact // on transaction behavior. // func (w *LMDBWrapper) NewLMDBTx(write bool, initialIndexName string, frag *fragment) (tx *LMDBTx) { //w.muDb.Lock() //defer w.muDb.Unlock() rwflag := uint(0) // writable txn denotated by lack of the lmdb.Readonly flag. if !write { rwflag = lmdb.Readonly } runtime.LockOSThread() sn := atomic.AddInt64(&w.nextTxSn, 1) lmdbTxn, err := w.env.BeginTxn(nil, rwflag) panicOn(err) lmdbTxn.RawRead = true tx = &LMDBTx{ sn: sn, write: write, tx: lmdbTxn, dbi: w.dbi, Db: w, frag: frag, //initloc: stack(), doAllocZero: w.doAllocZero, initialIndexName: initialIndexName, DeleteEmptyContainer: w.DeleteEmptyContainer, } return } // Close shuts down the LMDB database. func (w *LMDBWrapper) Close() (err error) { w.muDb.Lock() defer w.muDb.Unlock() if !w.closed { w.reg.unregister(w) w.closed = true w.env.CloseDBI(w.dbi) w.env.Close() w.env = nil } return nil } // LMDBTx wraps a lmdb.Txn and provides the Tx interface // method implementations. // The methods on LMDBTx are thread-safe, and can be called // from different goroutines. type LMDBTx struct { // mu serializes lmdb operations on this single txn instance. mu sync.Mutex sn int64 // serial number write bool dbi lmdb.DBI Db *LMDBWrapper tx *lmdb.Txn frag *fragment opcount int //initloc string // stack trace of where we were initially created. doAllocZero bool initialIndexName string DeleteEmptyContainer bool unlocked bool // runtime.UnlockOSThread has been done. } func (tx *LMDBTx) Type() string { return LmdbTxn } func (tx *LMDBTx) UseRowCache() bool { return true } // Pointer gives us a memory address for the underlying transaction for debugging. // It is public because we use it in roaring to report invalid container memory access // outside of a transaction. func (tx *LMDBTx) Pointer() string { return fmt.Sprintf("%p", tx) } // Rollback rolls back the transaction. func (tx *LMDBTx) Rollback() { tx.mu.Lock() defer tx.mu.Unlock() //pp("LMDBTx.Rollback p=%p, its: '%v' initloc: '%v',\n rollbackloc:'%v'", tx, tx.Db.UnprotectedListOpenItAsString(), tx.initloc, stack()) tx.tx.Abort() // must hold tx.mu mutex lock if !tx.unlocked { runtime.UnlockOSThread() tx.unlocked = true } } // Commit commits the transaction to permanent storage. // Commits can handle up to 100k updates to fragments // at once, but not more. This is a LMDBDB imposed limit. func (tx *LMDBTx) Commit() error { tx.mu.Lock() defer tx.mu.Unlock() //pp("LMDBTx.Commit (write:%v) p=%p, stackID=%x openit: '%v' initloc: '%v', commitloc:\n%v", tx.write, tx, stackID, tx.Db.UnprotectedListOpenItAsString(), tx.initloc, stack()) err := tx.tx.Commit() // must hold tx.mu mutex lock panicOn(err) if !tx.unlocked { runtime.UnlockOSThread() tx.unlocked = true } //pp("done committing LMDBTx sn=%v", tx.sn) return err } // Readonly returns true iff the LMDBTx is read-only. func (tx *LMDBTx) Readonly() bool { return !tx.write } // RoaringBitmap returns the roaring.Bitmap for all bits in the fragment. func (tx *LMDBTx) RoaringBitmap(index, field, view string, shard uint64) (*roaring.Bitmap, error) { return tx.OffsetRange(index, field, view, shard, 0, 0, LeftShifted16MaxContainerKey) } // Container returns the requested roaring.Container, selected by fragment and ckey func (tx *LMDBTx) Container(index, field, view string, shard uint64, ckey uint64) (c *roaring.Container, err error) { // values returned from Get() are only valid while the transaction // is open. If you need to use a value outside of the transaction then // you must use copy() to copy it to another byte slice. // BUT here we are already inside the Txn. bkey := txkey.Key(index, field, view, shard, ckey) tx.mu.Lock() v, err := tx.tx.Get(tx.dbi, bkey) tx.mu.Unlock() if lmdb.IsNotFound(err) { // Seems crazy, but we, for now at least, // match what RoaringTx does by returning nil, nil. return nil, nil } else { if err != nil { panicOn(err) return nil, nil } } n := len(v) if n > 0 { c = tx.toContainer(v[n-1], v[0:(n-1)]) } return } // PutContainer stores rc under the specified fragment and container ckey. func (tx *LMDBTx) PutContainer(index, field, view string, shard uint64, ckey uint64, rc *roaring.Container) error { bkey := txkey.Key(index, field, view, shard, ckey) var by []byte ct := roaring.ContainerType(rc) switch ct { case roaring.ContainerArray: by = fromArray16(roaring.AsArray(rc)) case roaring.ContainerBitmap: by = fromArray64(roaring.AsBitmap(rc)) case roaring.ContainerRun: by = fromInterval16(roaring.AsRuns(rc)) case roaring.ContainerNil: panic("wat? nil container is unexpected, no?!?") default: panic(fmt.Sprintf("unknown container type: %v", ct)) } tx.mu.Lock() err := tx.tx.Put(tx.dbi, bkey, append(by, ct), 0) // TODO: this might make a copy; can meta byte be stored elsewhere? tx.mu.Unlock() //panicOn(err) // mdb_put: invalid argument // TODO(jea): need to handle? // lmdb.TxnFull // lmdb.CursorFull // lmdb.PageFull return err } // RemoveContainer deletes the container specified by the shard and container key ckey func (tx *LMDBTx) RemoveContainer(index, field, view string, shard uint64, ckey uint64) error { bkey := txkey.Key(index, field, view, shard, ckey) tx.mu.Lock() err := tx.tx.Del(tx.dbi, bkey, nil) tx.mu.Unlock() if lmdb.IsNotFound(err) { return nil } return err } // Add sets all the a bits hot in the specified fragment. func (tx *LMDBTx) Add(index, field, view string, shard uint64, batched bool, a ...uint64) (changeCount int, err error) { return tx.addOrRemove(index, field, view, shard, batched, false, a...) } // Remove clears all the specified a bits in the chosen fragment. func (tx *LMDBTx) Remove(index, field, view string, shard uint64, a ...uint64) (changeCount int, err error) { const batched = false const remove = true return tx.addOrRemove(index, field, view, shard, batched, remove, a...) } func (tx *LMDBTx) addOrRemove(index, field, view string, shard uint64, batched, remove bool, a ...uint64) (changeCount int, err error) { // pure hack to match RoaringTx defer func() { if !remove && !batched { if changeCount > 0 { changeCount = 1 } } }() if len(a) == 0 { return 0, nil } // have to sort, b/c input is not always sorted. sort.Slice(a, func(i, j int) bool { return a[i] < a[j] }) var lastHi uint64 = math.MaxUint64 // highbits is always less than this starter. var rc *roaring.Container var hi uint64 var lo uint16 for i, v := range a { hi, lo = highbits(v), lowbits(v) if hi != lastHi { // either first time through, or changed to a different container. // do we need put the last updated container now? if i > 0 { // not first time through, write what we got. if remove && (rc == nil || rc.N() == 0) { err = tx.RemoveContainer(index, field, view, shard, lastHi) panicOn(err) } else { err = tx.PutContainer(index, field, view, shard, lastHi, rc) panicOn(err) } } // get the next container rc, err = tx.Container(index, field, view, shard, hi) panicOn(err) } // else same container, keep adding bits to rct. chng := false // rc can be nil before, and nil after, in both Remove/Add below. // The roaring container add() and remove() methods handle this. if remove { rc, chng = rc.Remove(lo) } else { rc, chng = rc.Add(lo) } if chng { changeCount++ } lastHi = hi } // write the last updates. if remove { if rc == nil || rc.N() == 0 { err = tx.RemoveContainer(index, field, view, shard, hi) panicOn(err) } else { err = tx.PutContainer(index, field, view, shard, hi, rc) panicOn(err) } } else { if rc == nil || rc.N() == 0 { panic("there should be no way to have an empty bitmap AFTER an Add() operation") } err = tx.PutContainer(index, field, view, shard, hi, rc) panicOn(err) } return } // Contains returns exists true iff the bit chosen by key is // hot (set to 1) in specified fragment. func (tx *LMDBTx) Contains(index, field, view string, shard uint64, key uint64) (exists bool, err error) { lo, hi := lowbits(key), highbits(key) bkey := txkey.Key(index, field, view, shard, hi) tx.mu.Lock() var v []byte v, err = tx.tx.Get(tx.dbi, bkey) tx.mu.Unlock() if lmdb.IsNotFound(err) { return false, nil } if err != nil { return false, err } n := len(v) if n > 0 { c := tx.toContainer(v[n-1], v[0:(n-1)]) exists = c.Contains(lo) } return exists, err } func (tx *LMDBTx) SliceOfShards(index, field, view, optionalViewPath string) (sliceOfShards []uint64, err error) { prefix := txkey.AllShardPrefix(index, field, view) bi := NewLMDBIterator(tx, prefix) defer bi.Close() lastShard := uint64(0) firstDone := false for bi.Next() { shard := txkey.ShardFromKey(bi.lastKey) if firstDone { if shard != lastShard { sliceOfShards = append(sliceOfShards, shard) } lastShard = shard } else { // first time lastShard = shard firstDone = true sliceOfShards = append(sliceOfShards, shard) } } return } // key is the container key for the first roaring Container // roaring docs: Iterator returns a ContainterIterator which *after* a call to Next(), a call to Value() will // return the first container at or after key. found will be true if a // container is found at key. // // LMDBTx notes: We auto-stop at the end of this shard, not going beyond. func (tx *LMDBTx) ContainerIterator(index, field, view string, shard uint64, firstRoaringContainerKey uint64) (citer roaring.ContainerIterator, found bool, err error) { // needle example: "idx:'i';fld:'f';vw:'v';shd:'00000000000000000000';key@00000000000000000000" needle := txkey.Key(index, field, view, shard, firstRoaringContainerKey) // prefix example: "idx:'i';fld:'f';vw:'v';shard:'00000000000000000000';key@" prefix := txkey.Prefix(index, field, view, shard) bi := NewLMDBIterator(tx, prefix) ok := bi.Seek(needle) if !ok { return bi, false, nil } // have to compare b/c lmdb might give us valid iterator // that is past our needle if needle isn't present. return bi, bytes.Equal(bi.lastKey, needle), nil } // LMDBIterator is the iterator returned from a LMDBTx.ContainerIterator() call. // It implements the roaring.ContainerIterator interface. type LMDBIterator struct { tx *LMDBTx cur *lmdb.Cursor dbi lmdb.DBI prefix []byte seekto []byte // seen counts how many Next() calls we have seen. // It is used to match roaring.ContainerIterator semantics. // Also useful for testing. seen int lastKey []byte lastVal []byte // *roaring.Container lastOK bool lastConsumed bool } // NewLMDBIterator creates an iterator on tx that will // only return badgerKeys that start with prefix. func NewLMDBIterator(tx *LMDBTx, prefix []byte) (bi *LMDBIterator) { cur, err := tx.tx.OpenCursor(tx.dbi) panicOn(err) bi = &LMDBIterator{ dbi: tx.dbi, tx: tx, cur: cur, prefix: prefix, } return } // Close tells the database and transaction that the user is done // with the iterator. func (bi *LMDBIterator) Close() { bi.cur.Close() } // Valid returns false if there are no more values in the iterator's range. func (bi *LMDBIterator) Valid() bool { return bi.lastOK } // Seek allows the iterator to start at needle instead of the global begining. func (bi *LMDBIterator) Seek(needle []byte) (ok bool) { bi.seen++ // if ommited, red TestLMDB_ContainerIterator_empty_iteration_loop() in lmdb_test.go. var k, v []byte var err error getflag := uint(lmdb.SetRange) if len(needle) == 0 { k, v, err = bi.cur.Get(oneByteSliceOfZero, nil, getflag) } else { k, v, err = bi.cur.Get(needle, nil, getflag) } if lmdb.IsNotFound(err) { bi.lastKey = nil bi.lastVal = nil bi.lastOK = false bi.lastConsumed = false return false } if len(bi.prefix) > 0 { ok = bytes.HasPrefix(k, bi.prefix) if !ok { bi.lastKey = nil bi.lastVal = nil bi.lastOK = false bi.lastConsumed = false return false } } if len(k) == 0 { bi.lastKey = nil bi.lastVal = nil bi.lastOK = false bi.lastConsumed = false return false } bi.lastKey = k bi.lastVal = v bi.lastOK = true bi.lastConsumed = false return true } func (bi *LMDBIterator) ValidForPrefix(prefix []byte) bool { if !bi.lastOK { return false } if len(bi.prefix) == 0 { return true } return bytes.HasPrefix(bi.lastKey, bi.prefix) } func (bi *LMDBIterator) String() (r string) { return fmt.Sprintf("LMDBIterator{prefix: '%v', seekto: '%v', seen:%v, lastKey:'%v', lastOK:%v, lastConsumed:%v}", string(bi.prefix), string(bi.seekto), bi.seen, string(bi.lastKey), bi.lastOK, bi.lastConsumed) } var oneByteSliceOfZero = []byte{0} // Next advances the iterator. func (bi *LMDBIterator) Next() (ok bool) { if bi.lastOK && !bi.lastConsumed { bi.seen++ bi.lastConsumed = true if len(bi.lastVal) == 0 { panic("bi.lastVal should not have len 0 if lastOK true") } return true } getflag := uint(lmdb.Next) prefix := bi.prefix if bi.seen == 0 { if len(bi.prefix) > 0 { getflag = lmdb.SetRange } } else { prefix = nil } bi.seen++ skipEmpty: var k, v []byte var err error if getflag == lmdb.SetRange && len(prefix) == 0 { // don't do nil as key on setrange, will panic // b/c keys in LMDB must be at least one byte long. // http://www.lmdb.tech/doc/group__mdb.html#structMDB__val // "Key sizes must be between 1 and mdb_env_get_maxkeysize() inclusive." // But if getflag == lmdb.Next, key can be nil. k, v, err = bi.cur.Get(oneByteSliceOfZero, nil, getflag) } else { k, v, err = bi.cur.Get(prefix, nil, getflag) } if lmdb.IsNotFound(err) { bi.lastKey = nil bi.lastVal = nil bi.lastOK = false bi.lastConsumed = false return false } if len(bi.prefix) > 0 { ok = bytes.HasPrefix(k, bi.prefix) if !ok { bi.lastKey = nil bi.lastVal = nil bi.lastOK = false bi.lastConsumed = false return false } } bi.lastKey = k bi.lastVal = v if len(v) == 0 { // actually under !tx.DeleteEmptyContainer, we can have empty containers! goto skipEmpty } bi.lastOK = true bi.lastConsumed = true return true } // Value retrieves what is pointed at currently by the iterator. func (bi *LMDBIterator) Value() (containerKey uint64, c *roaring.Container) { if !bi.lastOK { panic("bi.cur not valid") } containerKey = txkey.KeyExtractContainerKey(bi.lastKey) v := bi.lastVal n := len(v) if n > 0 { c = bi.tx.toContainer(v[n-1], v[0:(n-1)]) } else { panic("v should not be empty!") } return } // lmdbFinder implements roaring.IteratorFinder. // It is used by LMDBTx.ForEach() type lmdbFinder struct { tx *LMDBTx index string field string view string shard uint64 needClose []Closer } // FindIterator lets lmdbFinder implement the roaring.FindIterator interface. func (bf *lmdbFinder) FindIterator(seek uint64) (roaring.ContainerIterator, bool) { a, found, err := bf.tx.ContainerIterator(bf.index, bf.field, bf.view, bf.shard, seek) panicOn(err) bf.needClose = append(bf.needClose, a) return a, found } // Close closes all bf.needClose listed Closers. func (bf *lmdbFinder) Close() { for _, i := range bf.needClose { i.Close() } } // NewTxIterator returns a *roaring.Iterator that MUST have Close() called on it BEFORE // the transaction Commits or Rollsback. func (tx *LMDBTx) NewTxIterator(index, field, view string, shard uint64) *roaring.Iterator { bf := &lmdbFinder{tx: tx, index: index, field: field, view: view, shard: shard, needClose: make([]Closer, 0)} itr := roaring.NewIterator(bf) return itr } // ForEach applies fn to each bitmap in the fragment. func (tx *LMDBTx) ForEach(index, field, view string, shard uint64, fn func(i uint64) error) error { itr := tx.NewTxIterator(index, field, view, shard) defer itr.Close() // Seek can create many container iterators, thus bf.Close() needClose list. itr.Seek(0) // v is the bit we are operating on. for v, eof := itr.Next(); !eof; v, eof = itr.Next() { if err := fn(v); err != nil { return err } } return nil } // ForEachRange applies fn on the selected range of bits on the chosen fragment. func (tx *LMDBTx) ForEachRange(index, field, view string, shard uint64, start, end uint64, fn func(uint64) error) error { itr := tx.NewTxIterator(index, field, view, shard) defer itr.Close() itr.Seek(start) // v is the bit we are operating on. for v, eof := itr.Next(); !eof && v < end; v, eof = itr.Next() { if err := fn(v); err != nil { return err } } return nil } // Count operates on the full bitmap level, so it sums over all the containers // in the bitmap. func (tx *LMDBTx) Count(index, field, view string, shard uint64) (uint64, error) { a, found, err := tx.ContainerIterator(index, field, view, shard, 0) panicOn(err) defer a.Close() if !found { return 0, nil } result := int32(0) for a.Next() { ckey, cont := a.Value() _ = ckey result += cont.N() } return uint64(result), nil } // Max is the maximum bit-value in your bitmap. // Returns zero if the bitmap is empty. Odd, but this is what roaring.Max does. func (tx *LMDBTx) Max(index, field, view string, shard uint64) (uint64, error) { prefix := txkey.Prefix(index, field, view, shard) seekto := txkey.Prefix(index, field, view, shard+1) cur, err := tx.tx.OpenCursor(tx.dbi) panicOn(err) defer cur.Close() var k, v []byte if len(seekto) == 0 { _, _, err = cur.Get(oneByteSliceOfZero, nil, lmdb.SetRange) } else { _, _, err = cur.Get(seekto, nil, lmdb.SetRange) } if lmdb.IsNotFound(err) { // we have nothing >= seekto, but we might have stuff before it, and we'll wrap backwards. k, v, err = cur.Get(nil, nil, lmdb.Prev) if lmdb.IsNotFound(err) { // empty database return 0, nil } } else { // we found something >= seekto, so backup by 1. k, v, err = cur.Get(nil, nil, lmdb.Prev) if lmdb.IsNotFound(err) { // nothing before seekto return 0, nil } } // have something, are we in [prefix, seekto) ? cmp := bytes.Compare(k, prefix) if cmp >= 0 { // good, got max in k, v } else { return 0, nil // nothing in [prefix, seekto). } hb := txkey.KeyExtractContainerKey(k) n := len(v) if n == 0 { return 0, nil } rc := tx.toContainer(v[n-1], v[0:(n-1)]) lb := rc.Max() return hb<<16 | uint64(lb), nil } // Min returns the smallest bit set in the fragment. If no bit is hot, // the second return argument is false. func (tx *LMDBTx) Min(index, field, view string, shard uint64) (uint64, bool, error) { // Seek can create many container iterators, thus the bf.Close() needClose list. bf := &lmdbFinder{tx: tx, index: index, field: field, view: view, shard: shard, needClose: make([]Closer, 0)} defer bf.Close() itr := roaring.NewIterator(bf) itr.Seek(0) // v is the bit we are operating on. v, eof := itr.Next() if eof { return 0, false, nil } return v, true, nil } // UnionInPlace unions all the others Bitmaps into a new Bitmap, and then writes it to the // specified fragment. func (tx *LMDBTx) UnionInPlace(index, field, view string, shard uint64, others ...*roaring.Bitmap) error { rbm, err := tx.RoaringBitmap(index, field, view, shard) panicOn(err) rbm.UnionInPlace(others...) // iterate over the containers that changed within rbm, and write them back to disk. it, found := rbm.Containers.Iterator(0) _ = found // don't care about the value of found, because first containerKey might be > 0 for it.Next() { containerKey, rc := it.Value() // TODO: only write the changed ones back, as optimization? // Compare to ImportRoaringBits. err := tx.PutContainer(index, field, view, shard, containerKey, rc) panicOn(err) } return nil } // CountRange returns the count of hot bits in the start, end range on the fragment. // roaring.countRange counts the number of bits set between [start, end). func (tx *LMDBTx) CountRange(index, field, view string, shard uint64, start, end uint64) (n uint64, err error) { if start >= end { return 0, nil } skey := highbits(start) ekey := highbits(end) citer, found, err := tx.ContainerIterator(index, field, view, shard, skey) _ = found panicOn(err) defer citer.Close() // If range is entirely in one container then just count that range. if skey == ekey { citer.Next() _, c := citer.Value() return uint64(c.CountRange(int32(lowbits(start)), int32(lowbits(end)))), nil } for citer.Next() { k, c := citer.Value() if k < skey { citer.Close() panic(fmt.Sprintf("should be impossible for k(%v) to be less than skey(%v). tx p=%p", k, skey, tx)) } // k > ekey handles the case when start > end and where start and end // are in different containers. Same container case is already handled above. if k > ekey { break } if k == skey { n += uint64(c.CountRange(int32(lowbits(start)), roaring.MaxContainerVal+1)) continue } if k < ekey { n += uint64(c.N()) continue } if k == ekey { n += uint64(c.CountRange(0, int32(lowbits(end)))) break } } return n, nil } // OffsetRange creates a new roaring.Bitmap to return in other. For all the // hot bits in [start, endx) of the chosen fragment, it stores // them into other but with offset added to their bit position. // The primary client is doing this, using ShardWidth, already; see // fragment.rowFromStorage() in fragment.go. For example: // // data, err := tx.OffsetRange(f.index, f.field, f.view, f.shard, // f.shard*ShardWidth, rowID*ShardWidth, (rowID+1)*ShardWidth) // ^ offset ^ start ^ endx // // The start and endx arguments are container keys that have been shifted left by 16 bits; // their highbits() will be taken to determine the actual container keys. This // is done to conform to the roaring.OffsetRange() argument convention. // func (tx *LMDBTx) OffsetRange(index, field, view string, shard, offset, start, endx uint64) (other *roaring.Bitmap, err error) { // roaring does these three checks in its OffsetRange if lowbits(offset) != 0 { panic("offset must not contain low bits") } if lowbits(start) != 0 { panic("range start must not contain low bits") } if lowbits(endx) != 0 { panic("range end must not contain low bits") } other = roaring.NewSliceBitmap() off := highbits(offset) hi0, hi1 := highbits(start), highbits(endx) needle := txkey.Key(index, field, view, shard, hi0) prefix := txkey.Prefix(index, field, view, shard) n2, pre2 := txkey.KeyAndPrefix(index, field, view, shard, hi0) if string(n2) != string(needle) { panic(fmt.Sprintf("problem! n2(%v) != needle(%v), txkey.KeyAndPrefix not consitent with txkey.Key()", string(n2), string(needle))) } if string(pre2) != string(prefix) { panic(fmt.Sprintf("problem! pre2(%v) != prefix(%v), txkey.KeyAndPrefix not consitent with txkey.Key()", string(pre2), string(prefix))) } it := NewLMDBIterator(tx, prefix) defer it.Close() it.Seek(needle) for ; it.ValidForPrefix(prefix); it.Next() { bkey := it.lastKey k := txkey.KeyExtractContainerKey(bkey) // >= hi1 is correct b/c endx cannot have any lowbits set. if uint64(k) >= hi1 { break } destCkey := off + (k - hi0) v := it.lastVal n := len(v) if n == 0 { continue } c := tx.toContainer(v[n-1], v[0:(n-1)]) other.Containers.Put(destCkey, c.Freeze()) } return other, nil } // IncrementOpN increments the tx opcount by changedN func (tx *LMDBTx) IncrementOpN(index, field, view string, shard uint64, changedN int) { tx.opcount += changedN } // ImportRoaringBits handles deletes by setting clear=true. // rowSet[rowID] returns the number of bit changed on that rowID. func (tx *LMDBTx) ImportRoaringBits(index, field, view string, shard uint64, itr roaring.RoaringIterator, clear bool, log bool, rowSize uint64, data []byte) (changed int, rowSet map[uint64]int, err error) { n := itr.Len() if n == 0 { return } rowSet = make(map[uint64]int) var currRow uint64 var oldC *roaring.Container for itrKey, synthC := itr.NextContainer(); synthC != nil; itrKey, synthC = itr.NextContainer() { if rowSize != 0 { currRow = itrKey / rowSize } nsynth := int(synthC.N()) if nsynth == 0 { continue } // INVAR: nsynth > 0 oldC, err = tx.Container(index, field, view, shard, itrKey) panicOn(err) if err != nil { return } if oldC == nil || oldC.N() == 0 { // no container at the itrKey in lmdb (or all zero container). if clear { // changed of 0 and empty rowSet is perfect, no need to change the defaults. continue } else { changed += nsynth rowSet[currRow] += nsynth err = tx.PutContainer(index, field, view, shard, itrKey, synthC) if err != nil { return } continue } } if clear { existN := oldC.N() // number of bits set in the old container newC := oldC.Difference(synthC) // update rowSet and changes if newC.N() == existN { // INVAR: do changed need adjusting? nope. same bit count, // so no change could have happened. continue } else { changes := int(existN - newC.N()) changed += changes rowSet[currRow] -= changes if tx.DeleteEmptyContainer && newC.N() == 0 { err = tx.RemoveContainer(index, field, view, shard, itrKey) if err != nil { return } continue } err = tx.PutContainer(index, field, view, shard, itrKey, newC) if err != nil { return } continue } } else { // setting bits existN := oldC.N() if existN == roaring.MaxContainerVal+1 { // completely full container already, set will do nothing. so changed of 0 default is perfect. continue } if existN == 0 { // can nsynth be zero? No, because of the continue/invariant above where nsynth > 0 changed += nsynth rowSet[currRow] += nsynth err = tx.PutContainer(index, field, view, shard, itrKey, synthC) if err != nil { return } continue } newC := roaring.Union(oldC, synthC) // UnionInPlace was giving us crashes on overly large containers. if roaring.ContainerType(newC) == roaring.ContainerBitmap { newC.Repair() // update the bit-count so .n is valid. b/c UnionInPlace doesn't update it. } if newC.N() != existN { changes := int(newC.N() - existN) changed += changes rowSet[currRow] += changes err = tx.PutContainer(index, field, view, shard, itrKey, newC) if err != nil { panicOn(err) return } continue } } } return } func (tx *LMDBTx) toContainer(typ byte, v []byte) (r *roaring.Container) { if len(v) == 0 { return nil } var w []byte useRowCache := tx.UseRowCache() if tx.doAllocZero || useRowCache { // Do electric fence-inspired bad-memory read detection. // // The v []byte lives in LMDBDB's memory-mapped vlog-file, // and LMDB will recycle it after tx ends with rollback or commit. // // Problem is, at least some operations were not respecting transaction boundaries. // This technique helped us find them. The rowCache was an example. // // See the global const DetectMemAccessPastTx // at the top of txfactory.go to activate/deactivate this. // // Seebs suggested this nice variation: we could use individual mmaps for these // copies, which would be unusable in production, but workable for testing, and then unmap them, // which would get us probable segfaults on future accesses to them. // // The go runtime also has an -efence flag which may be similarly useful if really pressed. // w = make([]byte, len(v)) copy(w, v) } else { w = v } switch typ { case roaring.ContainerArray: c := roaring.NewContainerArray(toArray16(w)) return c case roaring.ContainerBitmap: c := roaring.NewContainerBitmap(-1, toArray64(w)) return c case roaring.ContainerRun: c := roaring.NewContainerRun(toInterval16(w)) return c default: panic(fmt.Sprintf("unknown container: %v", typ)) } } // StringifiedLMDBKeys returns a string with all the container // keys available in lmdb. func (w *LMDBWrapper) StringifiedLMDBKeys(optionalUseThisTx Tx) (r string) { if optionalUseThisTx == nil { tx := w.NewLMDBTx(!writable, "", nil) defer tx.Rollback() r = stringifiedLMDBKeysTx(tx) return } btx, ok := optionalUseThisTx.(*LMDBTx) if !ok { return fmt.Sprintf("", optionalUseThisTx) } r = stringifiedLMDBKeysTx(btx) return } // countBitsSet returns the number of bits set (or "hot") in // the roaring container value found by the txkey.Key() // formatted bkey. func (tx *LMDBTx) countBitsSet(bkey []byte) (n int) { v, err := tx.tx.Get(tx.dbi, bkey) if lmdb.IsNotFound(err) { // some queries bkey may not be present! don't panic. return 0 } panicOn(err) n = len(v) if n > 0 { rc := tx.toContainer(v[n-1], v[0:(n-1)]) n = int(rc.N()) } return } func (tx *LMDBTx) Dump() { fmt.Printf("%v\n", stringifiedLMDBKeysTx(tx)) } // stringifiedLMDBKeysTx reports all the lmdb keys and a // corresponding blake3 hash viewable by txn within the entire // lmdb database. // It also reports how many bits are hot in the roaring container // (how many bits are set, or 1 rather than 0). // // By convention, we must return the empty string if there // are no keys present. The tests use this to confirm // an empty database. func stringifiedLMDBKeysTx(tx *LMDBTx) (r string) { r = "allkeys:[\n" it := NewLMDBIterator(tx, nil) defer it.Close() any := false for it.Next() { any = true bkey := it.lastKey key := string(bkey) ckey := txkey.KeyExtractContainerKey(bkey) hash := "" srbm := "" v := it.lastVal n := len(v) if n == 0 { panic("should not have empty v here") } hash = blake3sum16(v[0:(n - 1)]) ct := tx.toContainer(v[n-1], v[0:(n-1)]) cts := roaring.NewSliceContainers() cts.Put(ckey, ct) rbm := &roaring.Bitmap{Containers: cts} srbm = bitmapAsString(rbm) r += fmt.Sprintf("%v -> %v (%v hot)\n", key, hash, tx.countBitsSet(bkey)) r += " ......." + srbm + "\n" } r += "]\n all-in-blake3:" + blake3sum16([]byte(r)) if !any { return "" } return "lmdb-" + r } func (w *LMDBWrapper) DeleteDBPath(path string) (err error) { err = os.RemoveAll(path) if err != nil { return errors.Wrap(err, "DeleteDBPath") } lockfile := path + "-lock" if FileExists(lockfile) { err = os.RemoveAll(lockfile) } return } func (w *LMDBWrapper) DeleteField(index, field, fieldPath string) error { // under blue-green roaring_lmdb, the directory will not be found, b/c roaring will have // already done the os.RemoveAll(). BUT, RemoveAll returns nil error in this case. Docs: // "If the path does not exist, RemoveAll returns nil (no error)" err := w.DeleteDBPath(fieldPath) if err != nil { return errors.Wrap(err, "removing directory") } prefix := txkey.FieldPrefix(index, field) return w.DeletePrefix(prefix) } func (w *LMDBWrapper) DeleteFragment(index, field, view string, shard uint64, frag interface{}) error { prefix := txkey.Prefix(index, field, view, shard) return w.DeletePrefix(prefix) } func (w *LMDBWrapper) DeletePrefix(prefix []byte) error { tx := w.NewLMDBTx(writable, w.name, nil) // NewLMDBTx will grab these, so don't lock until after it. w.muDb.Lock() defer w.muDb.Unlock() bi := NewLMDBIterator(tx, prefix) for bi.Next() { err := bi.cur.Del(0) panicOn(err) } bi.Close() err := tx.Commit() panicOn(err) return nil } func (tx *LMDBTx) RoaringBitmapReader(index, field, view string, shard uint64, fragmentPathForRoaring string) (r io.ReadCloser, sz int64, err error) { rbm, err := tx.RoaringBitmap(index, field, view, shard) if err != nil { return nil, -1, errors.Wrap(err, "RoaringBitmapReader RoaringBitmap") } var buf bytes.Buffer sz, err = rbm.WriteTo(&buf) if err != nil { return nil, -1, errors.Wrap(err, "RoaringBitmapReader rbm.WriteTo(buf)") } return ioutil.NopCloser(&buf), sz, err }