From e4e94a5668f14f1bf80b013eaf5e2a2760c99fc2 Mon Sep 17 00:00:00 2001 From: Seebs Date: Thu, 10 Dec 2020 14:42:47 -0600 Subject: [PATCH] prevent weird rare failures in mutex imports Added additional mutex sample data and batches of it so we can confirm that overwrite works. It didn't work, so that needed to be fixed. Couple of things: (1) Wasn't updating "last value seen" so the check for an unsorted list didn't work. (2) Also didn't handle the case where there were to-clear values higher than any to-set value. This could result in bits not getting cleared, which could result in there being more than N bits to clear for N new bits. And that could cause really strange problems when the input slices were parts of a single larger slice, because bit positions to clear could get shoved in as possible columns in a future batch. --- fragment.go | 6 +- fragment_internal_test.go | 132 +++++++++++++++++++++++--------------- 2 files changed, 85 insertions(+), 53 deletions(-) diff --git a/fragment.go b/fragment.go index 9214a0eef..269827387 100644 --- a/fragment.go +++ b/fragment.go @@ -2416,7 +2416,6 @@ func (f *fragment) importPositions(tx Tx, set, clear []uint64, rowSet map[uint64 // TODO benchmark Add/RemoveN behavior with sorted/unsorted positions // Note: AddN() avoids writing to the op-log. While Add() does. changedN, err := tx.Add(f.index(), f.field(), f.view(), f.shard, !doBatched, set...) - if err != nil { return errors.Wrap(err, "adding positions") } @@ -2503,6 +2502,8 @@ func unclearSets(toSet, toClear []uint64) []uint64 { cv = toClear[cn] } } + copy(toClear[n:], toClear[cn:]) + n += len(toClear[cn:]) return toClear[:n] } @@ -2541,13 +2542,14 @@ func (f *fragment) bulkImportMutex(tx Tx, rowIDs, columnIDs []uint64) error { rowSet[rowID] = struct{}{} pos, err := f.pos(rowID, columnID) if err != nil { - return err + return errors.Wrap(err, fmt.Sprintf("finding pos for row %d, col %d", rowID, columnID)) } // positions are sorted by columns, but not by absolute // position. we might want them sorted, though. if pos < prev { unsorted = true } + prev = pos rowIDs[i] = pos } toSet := rowIDs diff --git a/fragment_internal_test.go b/fragment_internal_test.go index fe95664c3..6b79be177 100644 --- a/fragment_internal_test.go +++ b/fragment_internal_test.go @@ -3325,6 +3325,7 @@ func benchmarkRowsMaybeWritable(b *testing.B, writable bool) { b.Fatalf("error committing sample data: %v", err) } tx = idx.holder.txf.NewTx(Txo{Write: writable, Index: idx, Fragment: frag, Shard: 0}) + defer tx.Rollback() } testCases = append(testCases, txFrag{ rows: rows, @@ -5536,29 +5537,35 @@ func notBlueGreenTest(t *testing.T) { var mutexSamplesPrepared sync.Once -func requireMutexSampleData() { - mutexSamplesPrepared.Do(prepareMutexSampleData) +func requireMutexSampleData(tb testing.TB) { + mutexSamplesPrepared.Do(func() { prepareMutexSampleData(tb) }) } // a few mutex tests want common largeish pools of mutex data type mutexSampleData struct { name string - colIDs, rowIDs []uint64 + colIDs, rowIDs [2][]uint64 } -func (m *mutexSampleData) scratchSpace(cols, rows []uint64) ([]uint64, []uint64) { - if cap(cols) < len(m.colIDs) { - cols = make([]uint64, len(m.colIDs)) +// scratchSpace copies the values over corresponding entries in slices, +// reusing existing storage when possible. +func (m *mutexSampleData) scratchSpace(idx int, cols, rows []uint64) ([]uint64, []uint64) { + colIDs := m.colIDs[idx] + rowIDs := m.rowIDs[idx] + + if cap(cols) < len(colIDs) { + cols = make([]uint64, len(colIDs)) } else { - cols = cols[:len(m.colIDs)] + cols = cols[:len(colIDs)] } - copy(cols, m.colIDs) - if cap(rows) < len(m.rowIDs) { - rows = make([]uint64, len(m.rowIDs)) + copy(cols, colIDs) + + if cap(rows) < len(rowIDs) { + rows = make([]uint64, len(rowIDs)) } else { - rows = rows[:len(m.rowIDs)] + rows = rows[:len(rowIDs)] } - copy(rows, m.rowIDs) + copy(rows, rowIDs) return cols, rows } @@ -5624,9 +5631,12 @@ var mutexCaches = []string{ "none", } -const mutexSampleDataSize = ShardWidth +const mutexSampleDataSize = ShardWidth << 1 -func prepareMutexSampleData() { +// prepareMutexSampleData creates two sets of data for each density and +// number of rows, so that we can test performance when overwriting also. +func prepareMutexSampleData(tb testing.TB) { + myrand := rand.New(rand.NewSource(9)) for _, d := range mutexDensities { for _, s := range mutexSizes { rng := newMutexSampleRange(d.density, s.rows) @@ -5640,36 +5650,41 @@ func prepareMutexSampleData() { // // So for density 16, we compute spacing of 1, then // draw random numbers in [0,1), and add 1 to them. - spacing := ((int64(1) << (16 - d.density)) * 2) - 1 + spacing := ((1 << (16 - d.density)) * 2) - 1 rows := (int64(1) << s.rows) - expected := int64(mutexSampleDataSize) + expected := mutexSampleDataSize if (ShardWidth / spacing) < mutexSampleDataSize { expected = (ShardWidth / spacing) * 2 if expected < 2 { expected = 2 } } - data := &mutexSampleData{ - name: d.name + "/" + s.name, - colIDs: make([]uint64, mutexSampleDataSize), - rowIDs: make([]uint64, mutexSampleDataSize), - } - for i := int64(0); i < expected; i++ { - col += uint64(rand.Int63n(spacing)) + 1 + colIDs := make([]uint64, mutexSampleDataSize) + rowIDs := make([]uint64, mutexSampleDataSize) + data := &mutexSampleData{name: d.name + "/" + s.name} + prev := uint64(0) + generated := 0 + for i := 0; i < expected; i++ { + col += uint64(myrand.Int63n(int64(spacing))) + 1 // can only import one fragment at a time, // though! - if col >= ShardWidth { - expected = i - break + if col/ShardWidth > prev { + data.colIDs[prev] = colIDs[generated:i:i] + data.rowIDs[prev] = rowIDs[generated:i:i] + generated = i + prev = col / ShardWidth + if int(prev) >= len(data.colIDs) { + break + } } - row := uint64(rand.Int63n(rows)) - data.colIDs[i] = col - data.rowIDs[i] = row + row := uint64(myrand.Int63n(rows)) + colIDs[i] = col % ShardWidth + rowIDs[i] = row + } + if int(prev) < len(data.colIDs) { + data.colIDs[prev] = colIDs[generated:expected:expected] + data.rowIDs[prev] = rowIDs[generated:expected:expected] } - // if we came up short, because we overran the shard - // size, we reduced expected above. - data.colIDs = data.colIDs[:expected] - data.rowIDs = data.rowIDs[:expected] sampleMutexData[rng] = data } } @@ -5678,7 +5693,7 @@ func prepareMutexSampleData() { var importBatchSizes = []int{65536} func TestImportMutexSampleData(t *testing.T) { - requireMutexSampleData() + requireMutexSampleData(t) var scratchCols []uint64 var scratchRows []uint64 for rng, data := range sampleMutexData { @@ -5686,7 +5701,7 @@ func TestImportMutexSampleData(t *testing.T) { if rng.rows() > 256 { continue } - scratchCols, scratchRows = data.scratchSpace(scratchCols, scratchRows) + scratchCols, scratchRows = data.scratchSpace(0, scratchCols, scratchRows) t.Run(data.name, func(t *testing.T) { for _, batchSize := range importBatchSizes { t.Run(fmt.Sprintf("%d", batchSize), func(t *testing.T) { @@ -5699,7 +5714,7 @@ func TestImportMutexSampleData(t *testing.T) { if len(scratchCols) < max { max = len(scratchCols) } - err = f.bulkImport(tx, scratchRows[i:max], scratchCols[i:max], &ImportOptions{}) + err = f.bulkImport(tx, scratchRows[i:max:max], scratchCols[i:max:max], &ImportOptions{}) if err != nil { t.Fatalf("bulk importing ids [%d:%d]: %v", i, max, err) } @@ -5708,9 +5723,9 @@ func TestImportMutexSampleData(t *testing.T) { for k := uint32(0); k < rng.rows(); k++ { count += f.mustRow(tx, uint64(k)).Count() } - if int(count) != len(data.colIDs) { + if int(count) != len(data.colIDs[0]) { t.Fatalf("for %d rows, %d density: expected %d results, got %d", - rng.rows(), rng.density(), len(data.colIDs), count) + rng.rows(), rng.density(), len(data.colIDs[0]), count) } }) } @@ -5719,29 +5734,44 @@ func TestImportMutexSampleData(t *testing.T) { } func BenchmarkImportMutexSampleData(b *testing.B) { - requireMutexSampleData() - var scratchCols []uint64 - var scratchRows []uint64 + requireMutexSampleData(b) + var cols []uint64 + var rows []uint64 var data *mutexSampleData var cache string var batchSize int - // this exists just to manage indentation below - benchmarkFragmentImports := func(b *testing.B) { - f, _, tx := mustOpenMutexFragment(b, "i", "f", viewStandard, 0, cache) - defer f.Clean(b) - scratchCols, scratchRows = data.scratchSpace(scratchCols, scratchRows) - var err error - for i := 0; i < len(scratchCols) && i < (batchSize*b.N); i += batchSize { + var frag *fragment + var tx Tx + var idx *Index + benchmarkOneFragmentImports := func(b *testing.B, i int) { + cols, rows = data.scratchSpace(i, cols, rows) + for i := 0; i < len(cols) && i < (batchSize*b.N); i += batchSize { max := i + batchSize - if len(scratchCols) < max { - max = len(scratchCols) + if len(cols) < max { + max = len(cols) } - err = f.bulkImport(tx, scratchRows[i:max], scratchCols[i:max], &ImportOptions{}) + err := frag.bulkImport(tx, rows[i:max:max], cols[i:max:max], &ImportOptions{}) if err != nil { b.Fatalf("bulk importing ids [%d:%d]: %v", i, max, err) } } } + benchmarkFragmentImports := func(b *testing.B) { + frag, idx, tx = mustOpenMutexFragment(b, "i", "f", viewStandard, 0, cache) + defer frag.Clean(b) + for i := range data.colIDs { + b.Run(fmt.Sprintf("write-%d", i), func(b *testing.B) { + benchmarkOneFragmentImports(b, i) + }) + // Then commit that write and do another one as a new Tx. + err := tx.Commit() + if err != nil { + b.Fatalf("error commiting write: %v", err) + } + tx = idx.holder.txf.NewTx(Txo{Write: writable, Index: idx, Fragment: frag, Shard: 0}) + defer tx.Rollback() + } + } for _, data = range sampleMutexData { b.Run(data.name, func(b *testing.B) { for _, batchSize = range importBatchSizes {