diff --git a/Makefile b/Makefile index e794990ee..e38ed3860 100644 --- a/Makefile +++ b/Makefile @@ -160,12 +160,6 @@ topt-badger: @echo " log.topt.badger green: \c"; cat log.topt.badger | grep PASS |wc -l @echo " log.topt.badger red: \c"; cat log.topt.badger | grep '\-\-\- FAIL' |wc -l -topt-rb: - mv log.topt.roaring_badger log.topt.roaring_badger.prev || true - PILOSA_TXSRC=roaring_badger go test -v -tags='$(BUILD_TAGS)' $(TESTFLAGS) $(NOCHECKPTR) 2>&1 | tee log.topt.badger - @echo " log.topt.roaring_badger green: \c"; cat log.topt.roaring_badger | grep PASS |wc -l - @echo " log.topt.roaring_badger red: \c"; cat log.topt.roaring_badger | grep '\-\-\- FAIL' |wc -l - topt-badger-race: mv log.topt.badger-race log.topt.badger-race.prev || true PILOSA_TXSRC=badger go test -race -v -tags='$(BUILD_TAGS)' $(TESTFLAGS) $(NOCHECKPTR) 2>&1 | tee log.topt.badger-race diff --git a/fragment_internal_test.go b/fragment_internal_test.go index f30e27ef2..a8851a1bb 100644 --- a/fragment_internal_test.go +++ b/fragment_internal_test.go @@ -3902,6 +3902,7 @@ func TestFragment_RoaringImport(t *testing.T) { defer tx.Rollback() for num, input := range test { + vv("num=%v, input='%#v'", num, input) buf := &bytes.Buffer{} bm := roaring.NewBitmap(input...) _, err := bm.WriteTo(buf) @@ -3912,6 +3913,8 @@ func TestFragment_RoaringImport(t *testing.T) { if err != nil { t.Fatalf("importing roaring: %v", err) } + tx.Dump() + exp := calcExpected(test[:num+1]...) for row, expCols := range exp { cols := f.mustRow(tx, uint64(row)).Columns() diff --git a/license.exceptions b/license.exceptions index e703b7c2a..67767e1e2 100644 --- a/license.exceptions +++ b/license.exceptions @@ -10,3 +10,4 @@ ./logger/filewriter.go ./logger/filewriter_test.go ./vprint.go +./rbf/vprint.go diff --git a/rbf/blake3.go b/rbf/blake3.go new file mode 100644 index 000000000..3935d2576 --- /dev/null +++ b/rbf/blake3.go @@ -0,0 +1,97 @@ +// 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. + +package rbf + +import ( + "encoding/binary" + "fmt" + "sync" + + cryptorand "crypto/rand" + "github.com/zeebo/blake3" +) + +// Blake3Hasher is a thread/goroutine safe way to +// obtain a blake3 cryptographic hash of input []byte. +// Reference https://github.com/BLAKE3-team/BLAKE3 +// suggests it is 6x faster than BLAKE2B. +// The Go github.com/zeebo/blake3 version is +// AVX2 and SSE4.1 accelerated. +type Blake3Hasher struct { + hasher *blake3.Hasher + hasherMu sync.Mutex +} + +// NewBlake3Hasher returns a new Blake3Hasher. +func NewBlake3Hasher() *Blake3Hasher { + return &Blake3Hasher{ + hasher: blake3.New(), + } +} + +// CryptoHash writes the blake3 cryptographic hash of +// input into buffer and returns it. +// Like the standard libary's hash.Hash interface's Sum() method, +// the buffer is re-used and overwritten +// to avoid allocation. The caller determines the byte length of +// the outputCryptohash by the size of the supplied buffer +// slice, and this will be exactly equal to the supplies bytes. +// In this way, shorter or longer hashes can be provided as +// needed. +func (w *Blake3Hasher) CryptoHash(input []byte, buffer []byte) (outputCryptohash []byte) { + w.hasherMu.Lock() + w.hasher.Reset() + + // "Write implements part of the hash.Hash interface. It never returns an error." + // -- https://godoc.org/github.com/zeebo/blake3#Hasher.Write + _, _ = w.hasher.Write(input) + + // Digest.Read reads data from the hasher into buffer. + // "It always fills the entire buffer and never errors." + // -- https://godoc.org/github.com/zeebo/blake3#Digest + _, _ = w.hasher.Digest().Read(buffer) + + // no chance of panic, so avoid any defer cost. + w.hasherMu.Unlock() + + return buffer +} + +// blake3sum16 might be slower because we allocate a new hasher every time, but +// it is more conenient for writing debug code. It returns +// a 16 byte hash as a hexidecimal string. +func blake3sum16(input []byte) string { + hasher := blake3.New() + + _, _ = hasher.Write(input) + var buf [16]byte + _, _ = hasher.Digest().Read(buf[0:]) + + return fmt.Sprintf("%x", buf) +} + +// cryptoRandInt64 uses crypto/rand to get an random int64 +func cryptoRandInt64() int64 { + c := 8 + b := make([]byte, c) + _, err := cryptorand.Read(b) + if err != nil { + panic(err) + } + r := int64(binary.LittleEndian.Uint64(b)) + return r +} + +var _ = cryptoRandInt64 // happy linter diff --git a/rbf/tx.go b/rbf/tx.go index 83f1ae244..ca208fd1b 100644 --- a/rbf/tx.go +++ b/rbf/tx.go @@ -14,10 +14,12 @@ package rbf import ( + "bytes" "fmt" "io" "math" "sort" + "strconv" "strings" "sync" @@ -1152,7 +1154,7 @@ func (itr *containerIterator) Close() {} // Next moves the iterator to the next container. func (itr *containerIterator) Next() bool { err := itr.cursor.Next() - return err != nil + return err == nil } // Value returns the current key & container. @@ -1160,3 +1162,163 @@ func (itr *containerIterator) Value() (uint64, *roaring.Container) { cell := itr.cursor.cell() return cell.Key, toContainer(cell, itr.cursor.tx) } + +func (tx *Tx) Dump(index string) { + fmt.Println(tx.DumpString(index)) +} +func (tx *Tx) DumpString(index string) (r string) { + + r = "allkeys:[\n" + + // grab root records, for a list of bitmaps. + records, err := tx.rootRecords() + panicOn(err) + n := 0 + for _, rr := range records { + c, err := tx.cursor(rr.Name) + panicOn(err) + err = c.First() + if err == io.EOF { + r += "" + n++ + continue + } + panicOn(err) + for { + err := c.Next() // hung here? + if err == io.EOF { + break + } + panicOn(err) + cell := c.cell() + ckey := cell.Key + ct := toContainer(cell, tx) + + s := stringOfCkeyCt(ckey, ct, rr.Name, index) + r += s + n++ + } + } + if n == 0 { + return "" + } + // note that we can have a bitmap present, but it can be empty + r += "]\n all-in-blake3:" + blake3sum16([]byte(r)) + "\n" + + return "rbf-" + r +} + +func containerToBytes(ct *roaring.Container) []byte { + ty := roaring.ContainerType(ct) + switch ty { + case containerNil: + panic("nil container") + case containerArray: + return fromArray16(roaring.AsArray(ct)) + case containerBitmap: + return fromArray64(roaring.AsBitmap(ct)) + case containerRun: + return fromInterval16(roaring.AsRuns(ct)) + } + panic(fmt.Sprintf("unknown container type '%v'", int(ty))) +} + +func badgerKey(index, field, view string, shard uint64, roaringContainerKey uint64) []byte { + // The %020d which adds zero padding up to 20 runes is required to + // allow the textual sort to accurately + // reflect a numeric sort order. This is because, as a string, + // math.MaxUint64 is 20 bytes long. + // Example of such a badgerKey with a container-key that is math.MaxUint64: + // ...........................................12345678901234567890 + // idx:'i';fld:'f';vw:'standard';shd:'1';ckey@18446744073709551615 + + prefix := badgerPrefix(index, field, view, shard) + ckey := []byte(fmt.Sprintf("%020d", roaringContainerKey)) + bkey := append(prefix, ckey...) + MustValidateKey(bkey) + return bkey +} + +// badgerPrefix returns everything from badgerKey up to and +// including the '@' fune in a badger key. The prefix excludes the roaring container key itself. +// NB must be kept in sync with badgerKey() and badgerKeyExtractContainerKey(). +func badgerPrefix(index, field, view string, shard uint64) []byte { + return []byte(fmt.Sprintf("idx:'%v';fld:'%v';vw:'%v';shd:'%020v';ckey@", index, field, view, shard)) +} + +// MustValidatekey will panic on a bad badgerKey with an informative message. +func MustValidateKey(bkey []byte) { + n := len(bkey) + if n < 56 { + panic(fmt.Sprintf("bkey too short min size is 56 but we see %v in '%v'", n, string(bkey))) + } + beforeCkey := bkey[n-26 : n-20] + if !bytes.Equal(beforeCkey, ckeyPartExpected) { + panic(fmt.Sprintf(`bkey did not have expected ";ckey@" at 26 bytes from the end of the bkey '%v'; instead had '%v'`, string(bkey), string(beforeCkey))) + } +} + +func bitmapAsString(rbm *roaring.Bitmap) (r string) { + r = "c(" + slc := rbm.Slice() + width := 0 + s := "" + for _, v := range slc { + if width == 0 { + s = fmt.Sprintf("%v", v) + } else { + s = fmt.Sprintf(", %v", v) + } + width += len(s) + r += s + if width > 70 { + r += ",\n" + width = 0 + } + } + if width == 0 && len(r) > 2 { + r = r[:len(r)-2] + } + return r + ")" +} + +// should really be exported from the pilosa/roaring package so we don't get out of sync... +const ( + containerNil byte = iota // no container + containerArray // slice of bit position values + containerBitmap // slice of 1024 uint64s + containerRun // container of run-encoded bits +) + +var ckeyPartExpected = []byte(";ckey@") + +func invName(rbfName string) (field, view string, shard uint64) { + s := strings.Split(rbfName, "\x00") + if len(s) != 3 { + panic("should have 3 parts") + } + field = s[0] + view = s[1] + var err error + shard, err = strconv.ParseUint(s[2], 10, 64) + panicOn(err) + return +} + +func stringOfCkeyCt(ckey uint64, ct *roaring.Container, rrName, index string) (s string) { + + by := containerToBytes(ct) + hash := blake3sum16(by) + + cts := roaring.NewSliceContainers() + cts.Put(ckey, ct) + rbm := &roaring.Bitmap{Containers: cts} + srbm := bitmapAsString(rbm) + + field, view, shard := invName(rrName) + bkey := string(badgerKey(index, field, view, shard, ckey)) + + s = fmt.Sprintf("%v -> %v (%v hot)\n", bkey, hash, ct.N()) + s += " ......." + srbm + "\n" + return +} diff --git a/rbf/tx_test.go b/rbf/tx_test.go index 85195ad1d..e245b73da 100644 --- a/rbf/tx_test.go +++ b/rbf/tx_test.go @@ -464,3 +464,29 @@ func BenchmarkTx_Contains(b *testing.B) { }) } } + +func TestTx_Dump(t *testing.T) { + db := MustOpenDB(t) + defer MustCloseDB(t, db) + tx := MustBegin(t, db, true) + defer tx.Rollback() + + index, field, view, shard := "i", "f", "v", uint64(15) + nm := rbfName(field, view, shard) + + if err := tx.CreateBitmap(nm); err != nil { + t.Fatal(err) + } else if _, err := tx.Add(nm, 0x00000001, 0x00000002, 0x00010003, 0x00030004); err != nil { + t.Fatal(err) + } + + // test that we don't crash, and get *something* back + s := tx.DumpString(index) + if s == "" { + panic("should have had 3 containers!") + } +} + +func rbfName(field, view string, shard uint64) string { + return fmt.Sprintf("%s\x00%s\x00%d", field, view, shard) +} diff --git a/rbf/vprint.go b/rbf/vprint.go new file mode 100644 index 000000000..dfd630c70 --- /dev/null +++ b/rbf/vprint.go @@ -0,0 +1,169 @@ +// home: https://github.com/glyerine/vprint +// Copyright 2019 Jason E. Aten, Ph.D. All rights reserved. +// License: MIT +// +// MIT License +// +// Permission is hereby granted, free of charge, to any person obtaining a copy +// of this software and associated documentation files (the "Software"), to deal +// in the Software without restriction, including without limitation the rights +// to use, copy, modify, merge, publish, distribute, sublicense, and/or sell +// copies of the Software, and to permit persons to whom the Software is +// furnished to do so, subject to the following conditions: +// +// The above copyright notice and this permission notice shall be included in all +// copies or substantial portions of the Software. +// +// THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR +// IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, +// FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE +// AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER +// LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, +// OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE +// SOFTWARE. + +package rbf + +import ( + "fmt" + "io" + "os" + "path" + "runtime" + "runtime/debug" + "sync" + "time" +) + +const RFC3339MsecTz0 = "2006-01-02T15:04:05.000Z07:00" +const RFC3339UsecTz0 = "2006-01-02T15:04:05.000000Z07:00" + +// for tons of debug output +var VerboseVerbose bool = false + +// convience functions for . import +var pp = PP +var vv = VV + +var panicOn = PanicOn + +func init() { + // keeper linter happy + _ = pp + _ = vv +} + +func PanicOn(err error) { + if err != nil { + panic(err) + } +} + +func PP(format string, a ...interface{}) { + if VerboseVerbose { + TSPrintf(format, a...) + } +} + +func VV(format string, a ...interface{}) { + TSPrintf(format, a...) +} + +func AlwaysPrintf(format string, a ...interface{}) { + TSPrintf(format, a...) +} + +var tsPrintfMut sync.Mutex + +// time-stamped printf +func TSPrintf(format string, a ...interface{}) { + tsPrintfMut.Lock() + Printf("\n%s %s ", FileLine(3), ts()) + Printf(format+"\n", a...) + tsPrintfMut.Unlock() +} + +// get timestamp for logging purposes +func ts() string { + return time.Now().Format(RFC3339UsecTz0) +} + +// so we can multi write easily, use our own printf +var OurStdout io.Writer = os.Stdout + +// Printf formats according to a format specifier and writes to standard output. +// It returns the number of bytes written and any write error encountered. +func Printf(format string, a ...interface{}) (n int, err error) { + return fmt.Fprintf(OurStdout, format, a...) +} + +func FileLine(depth int) string { + _, fileName, fileLine, ok := runtime.Caller(depth) + var s string + if ok { + s = fmt.Sprintf("%s:%d", path.Base(fileName), fileLine) + } else { + s = "" + } + return s +} + +func stack() string { + return string(debug.Stack()) +} + +func FileExists(name string) bool { + fi, err := os.Stat(name) + if err != nil { + return false + } + if fi.IsDir() { + return false + } + return true +} + +func DirExists(name string) bool { + fi, err := os.Stat(name) + if err != nil { + return false + } + if fi.IsDir() { + return true + } + return false +} + +func FileSize(name string) (int64, error) { + fi, err := os.Stat(name) + if err != nil { + return -1, err + } + return fi.Size(), nil +} + +// Caller returns the name of the calling function. +func Caller(upStack int) string { + // elide ourself and runtime.Callers + target := upStack + 2 + + pc := make([]uintptr, target+2) + n := runtime.Callers(0, pc) + + f := runtime.Frame{Function: "unknown"} + if n > 0 { + frames := runtime.CallersFrames(pc[:n]) + for i := 0; i <= target; i++ { + contender, more := frames.Next() + if i == target { + f = contender + } + if !more { + break + } + } + } + return f.Function +} + +var _ = stack // happy linter diff --git a/tx.go b/tx.go index 051397245..ff544aca7 100644 --- a/tx.go +++ b/tx.go @@ -910,7 +910,8 @@ func (tx *RoaringTx) RoaringBitmapReader(index, field, view string, shard uint64 } type RBFTx struct { - tx *rbf.Tx + index string + tx *rbf.Tx } func (tx *RBFTx) Type() string { @@ -1035,7 +1036,7 @@ func (tx *RBFTx) Pointer() string { } func (tx *RBFTx) Dump() { - // todo + tx.tx.Dump(tx.index) } // Readonly is true if the transaction is not read-and-write, but only doing reads.