From 394b8522d185e1082b1cfc30e6d095794421af4d Mon Sep 17 00:00:00 2001 From: Jason Aten Date: Fri, 31 Jul 2020 11:47:02 -0400 Subject: [PATCH] Follow roaring.Union() with optimize() to avoid overly large containers. The cmd/loader is a preliminary sketch of the load testing tool. --- badger.go | 26 ++++++- cmd/loader/loader.go | 150 ++++++++++++++++++++++++++++++++++++ cmd/loader/vprint.go | 177 +++++++++++++++++++++++++++++++++++++++++++ license.exceptions | 1 + roaring/roaring.go | 9 ++- 5 files changed, 360 insertions(+), 3 deletions(-) create mode 100644 cmd/loader/loader.go create mode 100644 cmd/loader/vprint.go diff --git a/badger.go b/badger.go index ce28de54e..bcd95f8da 100644 --- a/badger.go +++ b/badger.go @@ -296,6 +296,7 @@ func (r *badgerRegistrar) openBadgerDBWrapper(bpath string) (*BadgerDBWrapper, e opt.Compression = badgeroptions.None // turn off compression. opt.ZSTDCompressionLevel = 0 // really, just in case. + opt.SyncWrites = true // default is true, safe. // MaxCacheSize docs: // @@ -1535,7 +1536,7 @@ func (tx *BadgerTx) ImportRoaringBits(index, field, view string, shard uint64, i continue } - newC := oldC.UnionInPlace(synthC) + newC := roaring.Union(oldC, synthC) // UnionInPlace was giving us crashes on overly large containers. if roaring.ContainerType(newC) == containerBitmap { newC.Repair() // update the bit-count so .n is valid. b/c UnionInPlace doesn't update it. @@ -1634,6 +1635,9 @@ func fromArray16(a []uint16) []byte { if len(a) == 0 { return []byte{} } + if len(a) > 4096 { + panic(fmt.Sprintf("cannot put more than 4096 integers into an array container: %v too big", len(a))) + } return (*[8192]byte)(unsafe.Pointer(&a[0]))[: len(a)*2 : len(a)*2] } @@ -1650,6 +1654,9 @@ func fromInterval16(a []roaring.Interval16) []byte { if len(a) == 0 { return []byte{} } + if len(a) > 2048 { + panic(fmt.Sprintf("cannot put more than 2048 roaring.Interval16 into a container: %v too big", len(a))) + } return (*[8192]byte)(unsafe.Pointer(&a[0]))[: len(a)*4 : len(a)*4] } @@ -1769,6 +1776,23 @@ func asInts(a []uint64) (r []int) { return } +var _ = zeroKeyContainerAsString // happy linter + +// for debugging +func zeroKeyContainerAsString(ct *roaring.Container) (r string) { + cts := roaring.NewSliceContainers() + cts.Put(0, ct) + rbm := &roaring.Bitmap{Containers: cts} + r = fmt.Sprintf("[%v]:", containerTypeNames[roaring.ContainerType(ct)]) + bitmapAsString(rbm) + return +} + +var containerTypeNames = map[byte]string{ + containerArray: "array", + containerBitmap: "bitmap", + containerRun: "run", +} + func bitmapAsString(rbm *roaring.Bitmap) (r string) { r = "c(" slc := rbm.Slice() diff --git a/cmd/loader/loader.go b/cmd/loader/loader.go new file mode 100644 index 000000000..2a6a00ff2 --- /dev/null +++ b/cmd/loader/loader.go @@ -0,0 +1,150 @@ +// 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 main + +import ( + "archive/tar" + "compress/gzip" + "context" + "time" + //"fmt" + "fmt" + "github.com/pilosa/pilosa/v2" + "github.com/pilosa/pilosa/v2/http" + "io" + "io/ioutil" + gohttp "net/http" + //"log" + "os" + //"path/filepath" + //"sort" + "strconv" + "strings" +) + +func UploadTar(srcFile string, client *http.InternalClient) error { + t0 := time.Now() + + f, err := os.Open(srcFile) + if err != nil { + return (err) + } + defer f.Close() + var tarReader *tar.Reader + if strings.HasSuffix(srcFile, "gz") { + gzf, err := gzip.NewReader(f) + if err != nil { + return err + } + tarReader = tar.NewReader(gzf) + } else { + tarReader = tar.NewReader(f) + } + viewData := make(map[string][]byte) + //given ordered by index/field/view + //trait_store/product_count__commercial_cd_or_share_certificate/views/bsig_product_count__commercial_cd_or_share_certificate/fragments/255 + lastIndex := "" + lastField := "" + lastShard := uint64(0) + //vv("top of tar loop") + n := 0 + for { + header, err := tarReader.Next() + if err == io.EOF { + if header != nil { + panic("header should not be nil on err io.EOF") + } + //submit any stuff we have left + if len(viewData) > 0 { + request := &pilosa.ImportRoaringRequest{ + Views: viewData, + } + // Submit(lastIndex, lastField, lastShard, request) + //vv("about to submit lastIndex='%v' lastShard='%v'", lastIndex, lastShard) + uri := GetImportRoaringURI(lastIndex, lastShard) + err := client.ImportRoaring(context.Background(), uri, lastIndex, lastField, lastShard, false, request) + panicOn(err) + //vv("done with submit lastIndex='%v' lastShard='%v'", lastIndex, lastShard) + } + return nil + } + //vv("got header '%v'", header.Name) + n++ + if n%500 == 0 { + vv("n = %v, progress, elapsed '%v'", n, time.Since(t0)) + } + parts := strings.Split(header.Name, "/") + index := parts[0] + field := parts[1] + view := parts[3] + shard, err := strconv.ParseUint(parts[5], 10, 64) + if err != nil { + return err + } + // TODO: shards can be loaded in parallel, so maybe farm out to a worker set of goro. + if index != lastIndex || field != lastField || shard != lastShard { + if len(viewData) > 0 { + request := &pilosa.ImportRoaringRequest{ + Views: viewData, + } + //vv("about to submit lastIndex='%v' lastShard='%v'", lastIndex, lastShard) + uri := GetImportRoaringURI(lastIndex, lastShard) + panicOn(client.ImportRoaring(context.Background(), uri, lastIndex, lastField, lastShard, false, request)) + viewData = make(map[string][]byte) + //vv("done with submit lastIndex='%v' lastShard='%v'; took='%v'", lastIndex, lastShard, time.Since(t0)) + + } + } + roaringData, err := ioutil.ReadAll(tarReader) + if err != nil { + return err + } + if _, already := viewData[view]; already { + panic(fmt.Sprintf("view '%v' already present!", view)) + } + viewData[view] = roaringData + lastIndex = index + lastField = field + + //lastShard = shard + //vv("bottom of loop") + } +} + +func main() { + + host := "127.0.0.1:10101" + h := &gohttp.Client{} + c, err := http.NewInternalClient(host, h) + panicOn(err) + + tarSrcPath := "q2.tar.gz" + t0 := time.Now() + panicOn(UploadTar(tarSrcPath, c)) + vv("total elapsed '%v'", time.Since(t0)) +} + +var globURI *pilosa.URI + +func init() { + var err error + globURI, err = pilosa.NewURIFromHostPort("127.0.0.1", 10101) + panicOn(err) +} + +// get correct node to go to. +func GetImportRoaringURI(index string, shard uint64) *pilosa.URI { + return globURI +} diff --git a/cmd/loader/vprint.go b/cmd/loader/vprint.go new file mode 100644 index 000000000..e6bea8d55 --- /dev/null +++ b/cmd/loader/vprint.go @@ -0,0 +1,177 @@ +// 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 main + +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 +} + +// happy linter: +var _ = DirExists +var _ = FileExists +var _ = Caller +var _ = stack +var _ = RFC3339MsecTz0 +var _ = RFC3339UsecTz0 +var _ = AlwaysPrintf +var _ = FileSize diff --git a/license.exceptions b/license.exceptions index 67767e1e2..ace441ae6 100644 --- a/license.exceptions +++ b/license.exceptions @@ -11,3 +11,4 @@ ./logger/filewriter_test.go ./vprint.go ./rbf/vprint.go +./cmd/loader/vprint.go diff --git a/roaring/roaring.go b/roaring/roaring.go index 41ff21a17..9219b01ef 100644 --- a/roaring/roaring.go +++ b/roaring/roaring.go @@ -4368,6 +4368,7 @@ func unionArrayArray(a, b *Container) *Container { break } } + // note: len(output) CAN be > 4096 return NewContainerArray(output) } @@ -6865,8 +6866,12 @@ func ConvertRunToBitmap(c *Container) { func Optimize(c *Container) { c.optimize() } -func Union(a, b *Container) *Container { - return union(a, b) +func Union(a, b *Container) (c *Container) { + c = union(a, b) + // c can be have arrays that are too big, and need + // to be optimized into raw bitmaps. + c.optimize() + return c } func Difference(a, b *Container) *Container {