From 663c725779c65b61215d69a84b0dab8762c78125 Mon Sep 17 00:00:00 2001 From: Matt Jaffee Date: Mon, 11 Mar 2019 13:08:20 -0500 Subject: [PATCH 1/6] import benchmarking tweaks importRoaring large fragment benchmark skip concurrent import benchmarks with testing.short --- fragment_internal_test.go | 41 +++++++++++++++++++++++++++++++++++++++ 1 file changed, 41 insertions(+) diff --git a/fragment_internal_test.go b/fragment_internal_test.go index 3a528c840..f985847af 100644 --- a/fragment_internal_test.go +++ b/fragment_internal_test.go @@ -1970,6 +1970,9 @@ func BenchmarkImportRoaring(b *testing.B) { } func BenchmarkImportRoaringConcurrent(b *testing.B) { + if testing.Short() { + b.SkipNow() + } for _, numRows := range rowCases { data := getZipfRowsSliceRoaring(numRows, 1) b.Logf("%dRows: %.2fMB\n", numRows, float64(len(data))/1024/1024) @@ -2003,6 +2006,9 @@ func BenchmarkImportRoaringConcurrent(b *testing.B) { } } func BenchmarkImportRoaringUpdateConcurrent(b *testing.B) { + if testing.Short() { + b.SkipNow() + } for _, numRows := range rowCases { for _, numCols := range colCases { data := getZipfRowsSliceRoaring(numRows, 1) @@ -2162,6 +2168,41 @@ func BenchmarkImportIntoLargeFragment(b *testing.B) { } } +func BenchmarkImportRoaringIntoLargeFragment(b *testing.B) { + b.StopTimer() + initBigFrag() + updata := getUpdataRoaring(10000000, 11000, 0) + for i := 0; i < b.N; i++ { + origF, err := os.Open(bigFrag) + if err != nil { + b.Fatalf("opening frag file: %v", err) + } + fi, err := ioutil.TempFile(*TempDir, "") + if err != nil { + b.Fatalf("getting temp file: %v", err) + } + _, err = io.Copy(fi, origF) + if err != nil { + b.Fatalf("copying fragment file: %v", err) + } + origF.Close() + fi.Close() + nf := newFragment(fi.Name(), "i", "f", viewStandard, 0) + err = nf.Open() + if err != nil { + b.Fatalf("opening fragment: %v", err) + } + b.StartTimer() + err = nf.importRoaring(updata, false) + b.StopTimer() + if err != nil { + b.Fatalf("bulkImport: %v", err) + } + + nf.Clean(b) + } +} + func TestGetZipfRowsSliceRoaring(t *testing.T) { f := mustOpenFragment("i", "f", viewStandard, 0, DefaultCacheType) data := getZipfRowsSliceRoaring(10, 1) From e33ca2d0ae896e0a9f22073c263666c8ad20b58c Mon Sep 17 00:00:00 2001 From: Matt Jaffee Date: Fri, 8 Mar 2019 14:35:52 -0600 Subject: [PATCH 2/6] use UnionInPlace in import-roaring get the count of the existing fragment and compare it to the incoming bits to decide which should be unioned into the other. This should generally result in far fewer allocations, though there is much work that needs to be done within UnionInPlace to further improve things. unrelatedly, I added a TODO to change the long-query-time option to move it out of cluster. It should probably be happening at the API level so that different handlers can reuse it, but if we're going to do that we'll want to make sure that any potentially time intensive operations are pulled into api from handler (e.g. protobuf decoding) --- cluster.go | 1 + fragment.go | 13 ++++++++++--- server/config.go | 9 +++++---- 3 files changed, 16 insertions(+), 7 deletions(-) diff --git a/cluster.go b/cluster.go index 4565e91f3..5c5c0e7f1 100644 --- a/cluster.go +++ b/cluster.go @@ -184,6 +184,7 @@ type cluster struct { // nolint: maligned ReplicaN int // Threshold for logging long-running queries + // TODO(2.0) move this out of cluster. (why is it here??) longQueryTime time.Duration // Maximum number of Set() or Clear() commands per request. diff --git a/fragment.go b/fragment.go index 067814a0b..c09461470 100644 --- a/fragment.go +++ b/fragment.go @@ -1732,8 +1732,10 @@ func (f *fragment) importRoaring(data []byte, clear bool) error { rowSet := make([]uint64, 0) var lastRow uint64 = math.MaxUint64 + incomingCnt := 0 for iter.Next() { - key, _ := iter.Value() + key, c := iter.Value() + incomingCnt += int(c.N()) // virtual row for the current container vRow := key >> shardVsContainerExponent @@ -1749,8 +1751,13 @@ func (f *fragment) importRoaring(data []byte, clear bool) error { if clear { bm = f.storage.Difference(bm) } else { - if f.storage.Count() > 0 { - bm = f.storage.Union(bm) + if cnt := f.storage.Count(); cnt > 0 { + if incomingCnt > int(cnt) { + bm.UnionInPlace(f.storage) + } else { + f.storage.UnionInPlace(bm) + bm = f.storage + } } } diff --git a/server/config.go b/server/config.go index 6223f07da..9c787f92e 100644 --- a/server/config.go +++ b/server/config.go @@ -75,10 +75,11 @@ type Config struct { Cluster struct { // Disabled controls whether clustering functionality is enabled. - Disabled bool `toml:"disabled"` - Coordinator bool `toml:"coordinator"` - ReplicaN int `toml:"replicas"` - Hosts []string `toml:"hosts"` + Disabled bool `toml:"disabled"` + Coordinator bool `toml:"coordinator"` + ReplicaN int `toml:"replicas"` + Hosts []string `toml:"hosts"` + // TODO(2.0) move this out of cluster. (why is it here??) LongQueryTime toml.Duration `toml:"long-query-time"` } `toml:"cluster"` From 19807ff3a72ed6a2d9d55079af44eae466b16dac Mon Sep 17 00:00:00 2001 From: Matt Jaffee Date: Sat, 9 Mar 2019 21:51:08 -0600 Subject: [PATCH 3/6] use num containers to decide which direction to union avoids doing a potentially expensive f.storage.Count() --- fragment.go | 12 ++++-------- 1 file changed, 4 insertions(+), 8 deletions(-) diff --git a/fragment.go b/fragment.go index c09461470..0c4ff5dd6 100644 --- a/fragment.go +++ b/fragment.go @@ -1750,15 +1750,11 @@ func (f *fragment) importRoaring(data []byte, clear bool) error { if clear { bm = f.storage.Difference(bm) + } else if f.storage.Containers.Size() >= bm.Containers.Size() { + f.storage.UnionInPlace(bm) + bm = f.storage } else { - if cnt := f.storage.Count(); cnt > 0 { - if incomingCnt > int(cnt) { - bm.UnionInPlace(f.storage) - } else { - f.storage.UnionInPlace(bm) - bm = f.storage - } - } + bm.UnionInPlace(f.storage) } for _, rowID := range rowSet { From d0f8304f1c32ade7d3ddf41cdafa7c04ae3eec27 Mon Sep 17 00:00:00 2001 From: Matt Jaffee Date: Mon, 11 Mar 2019 12:03:18 -0500 Subject: [PATCH 4/6] add importRoaring small updates benchmark --- fragment_internal_test.go | 34 ++++++++++++++++++++++++++++++++++ 1 file changed, 34 insertions(+) diff --git a/fragment_internal_test.go b/fragment_internal_test.go index f985847af..dc2b786b4 100644 --- a/fragment_internal_test.go +++ b/fragment_internal_test.go @@ -766,6 +766,40 @@ func BenchmarkFragment_RepeatedSmallImports(b *testing.B) { } } +func BenchmarkFragment_RepeatedSmallImportsRoaring(b *testing.B) { + for _, numUpdates := range []int{100} { + for _, bitsPerUpdate := range []uint64{100, 1000} { + for _, numRows := range []uint64{1000, 100000, 1000000} { + for _, opN := range []int{1, 5000, 50000} { + b.Run(fmt.Sprintf("Rows%dUpdates%dBits%dOpN%d", numRows, numUpdates, bitsPerUpdate, opN), func(b *testing.B) { + for a := 0; a < b.N; a++ { + b.StopTimer() + // build the update data set all at once - this will get applied + // to a fragment in numUpdates batches + f := mustOpenFragment("i", "f", viewStandard, 0, "") + f.MaxOpN = opN + defer f.Clean(b) + err := f.importRoaring(getZipfRowsSliceRoaring(numRows, 1), false) + if err != nil { + b.Fatalf("importing base data for benchmark: %v", err) + } + for i := 0; i < numUpdates; i++ { + data := getUpdataRoaring(numRows, bitsPerUpdate, int64(i)) + b.StartTimer() + err := f.importRoaring(data, false) + b.StopTimer() + if err != nil { + b.Fatalf("doing small roaring import: %v", err) + } + } + } + }) + } + } + } + } +} + func BenchmarkFragment_RepeatedSmallValueImports(b *testing.B) { initialCols := make([]uint64, 0, ShardWidth) initialVals := make([]uint64, 0, ShardWidth) From 52d43fb4e2076e4ccfafa241e6573766a725dec3 Mon Sep 17 00:00:00 2001 From: Matt Jaffee Date: Mon, 11 Mar 2019 12:03:42 -0500 Subject: [PATCH 5/6] add smallPath for importRoaring this converts the rowSet to a map from a slice which might be bad... benchmarks will tell. --- fragment.go | 23 ++++++++++++++++++----- 1 file changed, 18 insertions(+), 5 deletions(-) diff --git a/fragment.go b/fragment.go index 0c4ff5dd6..0801c8001 100644 --- a/fragment.go +++ b/fragment.go @@ -1729,13 +1729,13 @@ func (f *fragment) importRoaring(data []byte, clear bool) error { // get a list of keys in order to update the cache iter, _ := bm.Containers.Iterator(0) - rowSet := make([]uint64, 0) + rowSet := make(map[uint64]struct{}) var lastRow uint64 = math.MaxUint64 - incomingCnt := 0 + incomingCnt := uint64(0) for iter.Next() { key, c := iter.Value() - incomingCnt += int(c.N()) + incomingCnt += uint64(c.N()) // virtual row for the current container vRow := key >> shardVsContainerExponent @@ -1744,10 +1744,23 @@ func (f *fragment) importRoaring(data []byte, clear bool) error { if vRow == lastRow { continue } - rowSet = append(rowSet, vRow) + rowSet[vRow] = struct{}{} lastRow = vRow } + // take smallPath? TODO - ideally instead of checking f.storage.Any(), the + // test here would be if the storage size (in bytes) is significantly + // greater than the size of the incoming bits serialized as append + // operations. Getting the storage size might be a bit expensive though + // especially if the fragment isn't mapped. + if incomingCnt+uint64(f.opN) <= uint64(f.MaxOpN) && f.storage.Any() { + toSet, toClear := bm.Slice(), []uint64{} + if clear { + toSet, toClear = toClear, toSet + } + return f.importPositions(toSet, toClear, rowSet) + } + if clear { bm = f.storage.Difference(bm) } else if f.storage.Containers.Size() >= bm.Containers.Size() { @@ -1757,7 +1770,7 @@ func (f *fragment) importRoaring(data []byte, clear bool) error { bm.UnionInPlace(f.storage) } - for _, rowID := range rowSet { + for rowID := range rowSet { n := bm.CountRange(rowID*ShardWidth, (rowID+1)*ShardWidth) f.cache.BulkAdd(rowID, n) } From a28141c4664dd77eeafc3e395b422e9551596804 Mon Sep 17 00:00:00 2001 From: Matt Jaffee Date: Mon, 11 Mar 2019 16:31:06 -0500 Subject: [PATCH 6/6] revert to Union for importRoaring UnionInPlace is still heavily affected by https://github.com/pilosa/pilosa/issues/1875 where containers that exist in an incoming bitmap can cause massive unnecessary allocations of bitmap containers when an array of short length is all that's needed. --- fragment.go | 7 ++----- 1 file changed, 2 insertions(+), 5 deletions(-) diff --git a/fragment.go b/fragment.go index 0801c8001..32b870fb9 100644 --- a/fragment.go +++ b/fragment.go @@ -1763,11 +1763,8 @@ func (f *fragment) importRoaring(data []byte, clear bool) error { if clear { bm = f.storage.Difference(bm) - } else if f.storage.Containers.Size() >= bm.Containers.Size() { - f.storage.UnionInPlace(bm) - bm = f.storage - } else { - bm.UnionInPlace(f.storage) + } else if f.storage.Any() { + bm = f.storage.Union(bm) } for rowID := range rowSet {