diff --git a/api.go b/api.go index d8cdef042..8d3bfcea5 100644 --- a/api.go +++ b/api.go @@ -1114,7 +1114,28 @@ func (api *API) ImportAtomicRecord(ctx context.Context, req *AtomicRecord, opts return tx.Commit() } +// This is a hide your face ugly hack, forced upon +// us by the horrible invention of function based options +// by the usually brilliant Rob Pike. - JEA +func addClearToImportOptions(opts []ImportOption) []ImportOption { + var opt ImportOptions + for _, o := range opts { + // check for side-effect of setting io.Clear; that is + // how we know it is present. + _ = o(&opt) + if opt.Clear { + // we already have the clear flag set, so nothing more to do. + return opts + } + } + // no clear flag being set, add that option now. + return append(opts, OptImportOptionsClear(true)) +} + func (api *API) Import(ctx context.Context, req *ImportRequest, opts ...ImportOption) error { + if req.Clear { + opts = addClearToImportOptions(opts) + } return api.ImportWithTx(ctx, nil, req, opts...) } @@ -1254,6 +1275,9 @@ func (api *API) ImportWithTx(ctx context.Context, tx Tx, req *ImportRequest, opt } func (api *API) ImportValue(ctx context.Context, req *ImportValueRequest, opts ...ImportOption) error { + if req.Clear { + opts = addClearToImportOptions(opts) + } return api.ImportValueWithTx(ctx, nil, req, opts...) } diff --git a/api_test.go b/api_test.go index f592afb6e..88bb65510 100644 --- a/api_test.go +++ b/api_test.go @@ -474,3 +474,127 @@ type offsetModHasher struct{} func (*offsetModHasher) Hash(key uint64, n int) int { return int(key+1) % n } + +func TestAPI_ClearFlagForImportAndImportValues(t *testing.T) { + c := test.MustRunCluster(t, 1, + []server.CommandOption{ + server.OptCommandServerOptions( + pilosa.OptServerNodeID("node0"), + pilosa.OptServerClusterHasher(&offsetModHasher{}), + pilosa.OptServerOpenTranslateReader(http.GetOpenTranslateReaderFunc(nil)), + )}, + ) + defer c.Close() + + // plan: + // 1. set a bit + // 2. clear with Import() using the ImportRequest.Clear flag + // 3. verifiy the clear is done. + // repeat for ImportValueRequest and ImportValues() + + m0 := c[0] + m0api := m0.API + + ctx := context.Background() + index := "i" + fieldAcct0 := "acct0" + + opts := pilosa.OptFieldTypeInt(-1000, 1000) + + _, err := m0api.CreateIndex(ctx, index, pilosa.IndexOptions{}) + if err != nil { + t.Fatalf("creating index: %v", err) + } + _, err = m0api.CreateField(ctx, index, fieldAcct0, opts) + if err != nil { + t.Fatalf("creating fieldAcct0: %v", err) + } + + iraField := "ira" // set field. + iraRowID := uint64(3) + _, err = m0api.CreateField(ctx, index, iraField) + if err != nil { + t.Fatalf("creating fieldIRA: %v", err) + } + + acctOwnerID := uint64(78) // ColumnID + shard := acctOwnerID / ShardWidth + acct0bal := int64(500) + + ivr0 := &pilosa.ImportValueRequest{ + Index: index, + Field: fieldAcct0, + Shard: shard, + ColumnIDs: []uint64{acctOwnerID}, + Values: []int64{acct0bal}, + } + ir0 := &pilosa.ImportRequest{ + Index: index, + Field: iraField, + Shard: shard, + ColumnIDs: []uint64{acctOwnerID}, + RowIDs: []uint64{iraRowID}, + } + + if err := m0api.Import(ctx, ir0); err != nil { + t.Fatal(err) + } + if err := m0api.ImportValue(ctx, ivr0); err != nil { + t.Fatal(err) + } + + bitIsSet := func() bool { + query := fmt.Sprintf("Row(%v=%v)", iraField, iraRowID) + res, err := m0api.Query(context.Background(), &pilosa.QueryRequest{Index: index, Query: query}) + panicOn(err) + cols := res.Results[0].(*pilosa.Row).Columns() + for i := range cols { + if cols[i] == acctOwnerID { + return true + } + } + return false + } + + if !bitIsSet() { + panic("IRA bit should have been set") + } + + queryAcct := func(m0api *pilosa.API, acctOwnerID uint64, fieldAcct0, index string) (acctBal int64) { + query := fmt.Sprintf("FieldValue(field=%v, column=%v)", fieldAcct0, acctOwnerID) + res, err := m0api.Query(context.Background(), &pilosa.QueryRequest{Index: index, Query: query}) + panicOn(err) + + if len(res.Results) == 0 { + return 0 + } + valCount := res.Results[0].(pilosa.ValCount) + return valCount.Val + } + + bal := queryAcct(m0api, acctOwnerID, fieldAcct0, index) + + if bal != acct0bal { + panic(fmt.Sprintf("expected %v, observed %v starting acct0 balance", acct0bal, bal)) + } + + // clear the bit + ir0.Clear = true + if err := m0api.Import(ctx, ir0); err != nil { + t.Fatal(err) + } + + if bitIsSet() { + panic("IRA bit should have been cleared") + } + + // clear the BSI + ivr0.Clear = true + if err := m0api.ImportValue(ctx, ivr0); err != nil { + t.Fatal(err) + } + bal = queryAcct(m0api, acctOwnerID, fieldAcct0, index) + if bal != 0 { + panic(fmt.Sprintf("expected %v, observed %v starting acct0 balance", acct0bal, 0)) + } +} diff --git a/fragment_internal_test.go b/fragment_internal_test.go index 8ef6d5909..d033a24e6 100644 --- a/fragment_internal_test.go +++ b/fragment_internal_test.go @@ -29,6 +29,7 @@ import ( "runtime" "runtime/debug" "sort" + "strings" "testing" "testing/quick" @@ -190,6 +191,8 @@ func TestFragment_RowcacheMap(t *testing.T) { // Ensure a fragment can clear a row. func TestFragment_ClearRow(t *testing.T) { + notBlueGreenTest(t) + f, idx := mustOpenFragment("i", "f", viewStandard, 0, "") _ = idx defer f.Clean(t) @@ -225,6 +228,7 @@ func TestFragment_ClearRow(t *testing.T) { // Ensure a fragment can set a row. func TestFragment_SetRow(t *testing.T) { + notBlueGreenTest(t) f, idx := mustOpenFragment("i", "f", viewStandard, 7, "") _ = idx defer f.Clean(t) @@ -5644,3 +5648,12 @@ func TestFragment_Bug_Q2DoubleDelete(t *testing.T) { t.Fatalf("expected nothing got %v", res) } } + +func notBlueGreenTest(t *testing.T) { + src := os.Getenv("PILOSA_TXSRC") + if strings.Contains(src, "_") { + if strings.Contains(src, "roaring") { + t.Skip("skip under blue green with roaring") + } + } +} diff --git a/http/translator.go b/http/translator.go index f95067f70..0722dcbf9 100644 --- a/http/translator.go +++ b/http/translator.go @@ -22,19 +22,26 @@ import ( "io" "io/ioutil" "net/http" + "sync" "github.com/pilosa/pilosa/v2" "github.com/pilosa/pilosa/v2/logger" ) func GetOpenTranslateReaderFunc(client *http.Client) pilosa.OpenTranslateReaderFunc { + return GetOpenTranslateReaderWithLockerFunc(client, nopLocker{}) +} + +func GetOpenTranslateReaderWithLockerFunc(client *http.Client, locker sync.Locker) pilosa.OpenTranslateReaderFunc { return func(ctx context.Context, nodeURL string, offsets pilosa.TranslateOffsetMap) (pilosa.TranslateEntryReader, error) { - return openTranslateReader(ctx, nodeURL, offsets, client) + return openTranslateReader(ctx, nodeURL, offsets, client, locker) } } -func openTranslateReader(ctx context.Context, nodeURL string, offsets pilosa.TranslateOffsetMap, client *http.Client) (pilosa.TranslateEntryReader, error) { +func openTranslateReader(ctx context.Context, nodeURL string, offsets pilosa.TranslateOffsetMap, client *http.Client, locker sync.Locker) (pilosa.TranslateEntryReader, error) { r := NewTranslateEntryReader(ctx, client) + r.locker = locker + r.URL = nodeURL + "/internal/translate/data" r.Offsets = offsets if err := r.Open(); err != nil { @@ -43,9 +50,16 @@ func openTranslateReader(ctx context.Context, nodeURL string, offsets pilosa.Tra return r, nil } +type nopLocker struct{} + +func (nopLocker) Lock() {} +func (nopLocker) Unlock() {} + // TranslateEntryReader represents an implementation of pilosa.TranslateEntryReader. // It consolidates all index & field translate entries into a single reader. type TranslateEntryReader struct { + locker sync.Locker + ctx context.Context cancel func() @@ -70,7 +84,7 @@ func NewTranslateEntryReader(ctx context.Context, client *http.Client) *Translat if client == nil { client = http.DefaultClient } - r := &TranslateEntryReader{HTTPClient: client, Logger: logger.NopLogger} + r := &TranslateEntryReader{locker: nopLocker{}, HTTPClient: client, Logger: logger.NopLogger} r.ctx, r.cancel = context.WithCancel(ctx) return r } @@ -116,7 +130,10 @@ func (r *TranslateEntryReader) Close() error { r.cancel() } if r.body != nil { - return r.body.Close() + r.locker.Lock() + err := r.body.Close() + r.locker.Unlock() + return err } return nil } @@ -124,5 +141,8 @@ func (r *TranslateEntryReader) Close() error { // ReadEntry reads the next entry from the stream into entry. // Returns io.EOF at the end of the stream. func (r *TranslateEntryReader) ReadEntry(entry *pilosa.TranslateEntry) error { + r.locker.Lock() + defer r.locker.Unlock() + return r.dec.Decode(&entry) } diff --git a/http/translator_test.go b/http/translator_test.go index be6db423a..8d4a9362d 100644 --- a/http/translator_test.go +++ b/http/translator_test.go @@ -17,6 +17,7 @@ package http_test import ( "context" "fmt" + "sync" "testing" "time" @@ -151,3 +152,84 @@ func TestTranslateStore_EntryReader(t *testing.T) { }) */ } + +func benchmarkSetup(b *testing.B, ctx context.Context, key string, nkeys int) (string, pilosa.TranslateOffsetMap, func()) { + b.Helper() + + cluster := test.MustRunCluster(b, 1) + primary := cluster[0] + + idx := primary.MustCreateIndex(b, "i", pilosa.IndexOptions{}) + fld := primary.MustCreateField(b, idx.Name(), "f", pilosa.OptFieldKeys()) + offset := make(pilosa.TranslateOffsetMap) + offset.SetIndexPartitionOffset(idx.Name(), 0, 1) + offset.SetFieldOffset(idx.Name(), fld.Name(), 1) + + // Set data on the primary node. + for k := 0; k < nkeys; k++ { + if _, err := primary.API.Query(ctx, &pilosa.QueryRequest{ + Index: idx.Name(), + Query: fmt.Sprintf(`Set(%d, %s="%s%[1]d")`, k, fld.Name(), key), + }); err != nil { + b.Fatalf("quering api: %+v", err) + } + } + + return primary.URL(), offset, func() { + b.Helper() + + if err := primary.API.DeleteIndex(ctx, idx.Name()); err != nil { + panic(err) + } + if err := cluster.Close(); err != nil { + panic(err) + } + } +} + +func benchmarkReadEntry(b *testing.B, r pilosa.TranslateEntryReader, key string, nkeys int) { + var entry pilosa.TranslateEntry + for k := 0; k < nkeys; k++ { + if err := r.ReadEntry(&entry); err != nil { + b.Fatalf("reading entry: %+v", err) + } + if entry.Key != fmt.Sprintf("%s%d", key, k) { + b.Fatalf("got: %s, expected: %s%d", entry.Key, key, k) + } + } +} + +const ( + key = "foo" + nkeys = 1000 +) + +func BenchmarkReadEntryNoMutex(b *testing.B) { + ctx := context.Background() + url, offset, teardown := benchmarkSetup(b, ctx, key, nkeys) + defer teardown() + + for n := 0; n < b.N; n++ { + r, err := http.GetOpenTranslateReaderFunc(nil)(ctx, url, offset) + if err != nil { + b.Fatalf("opening translate reader: %+v", err) + } + benchmarkReadEntry(b, r, key, nkeys) + r.Close() + } +} + +func BenchmarkReadEntryWithMutex(b *testing.B) { + ctx := context.Background() + url, offset, teardown := benchmarkSetup(b, ctx, key, nkeys) + defer teardown() + + for n := 0; n < b.N; n++ { + r, err := http.GetOpenTranslateReaderWithLockerFunc(nil, &sync.Mutex{})(ctx, url, offset) + if err != nil { + b.Fatalf("opening translate reader: %+v", err) + } + benchmarkReadEntry(b, r, key, nkeys) + r.Close() + } +} diff --git a/lmdb.go b/lmdb.go index 9e0d30ed9..b67436abf 100644 --- a/lmdb.go +++ b/lmdb.go @@ -147,9 +147,9 @@ func (r *lmdbRegistrar) openLMDBWrapper(path0 string) (*LMDBWrapper, error) { flags = flags | lmdb.WriteMap | // Use a writable memory map. - lmdb.NoMetaSync | // Don't fsync metapage after commit. - lmdb.NoSync | // Don't fsync after commit. - lmdb.MapAsync | // Flush asynchronously when using the WriteMap flag. + //lmdb.NoMetaSync | // Don't fsync metapage after commit. + //lmdb.NoSync | // Don't fsync after commit. + //lmdb.MapAsync | // Flush asynchronously when using the WriteMap flag. lmdb.NoMemInit // Disable LMDB memory initialization err = env.Open(path, flags, 0644) @@ -344,7 +344,7 @@ func (tx *LMDBTx) Type() string { } func (tx *LMDBTx) UseRowCache() bool { - return false + return true } // Pointer gives us a memory address for the underlying transaction for debugging. diff --git a/pg/server_test.go b/pg/server_test.go index 4903fdb60..9eda44bcf 100644 --- a/pg/server_test.go +++ b/pg/server_test.go @@ -59,7 +59,7 @@ func TestStartupInvalidLength(t *testing.T) { res := testing.Benchmark(func(b *testing.B) { connect, shutdown, err := pgtest.ServeMem(&pg.Server{ MaxStartupSize: 1024, - Logger: logger.NewLogfLogger(t), + Logger: logger.NopLogger, }) if err != nil { t.Fatalf("starting in-memory postgres server: %v", err) diff --git a/server/handler_test.go b/server/handler_test.go index 31954eff8..ccd91bdcc 100644 --- a/server/handler_test.go +++ b/server/handler_test.go @@ -27,6 +27,7 @@ import ( "net/http/httptest" "reflect" "strings" + "sync" "testing" "time" @@ -1300,7 +1301,7 @@ func TestCluster_TranslateStore(t *testing.T) { cluster[0] = test.NewCommandNode(true, server.OptCommandServerOptions( pilosa.OptServerOpenTranslateStore(boltdb.OpenTranslateStore), - pilosa.OptServerOpenTranslateReader(http.GetOpenTranslateReaderFunc(nil)), + pilosa.OptServerOpenTranslateReader(http.GetOpenTranslateReaderWithLockerFunc(nil, &sync.Mutex{})), ), ) cluster[0].Config.Gossip.Port = "0" @@ -1329,7 +1330,7 @@ func TestClusterTranslator(t *testing.T) { cluster[1] = test.NewCommandNode(false, server.OptCommandServerOptions( pilosa.OptServerOpenTranslateStore(boltdb.OpenTranslateStore), - pilosa.OptServerOpenTranslateReader(http.GetOpenTranslateReaderFunc(nil)), + pilosa.OptServerOpenTranslateReader(http.GetOpenTranslateReaderWithLockerFunc(nil, &sync.Mutex{})), ), ) cluster[1].Config.Gossip.Port = "0" diff --git a/server/server.go b/server/server.go index 5578cf35f..e74d460e2 100644 --- a/server/server.go +++ b/server/server.go @@ -31,6 +31,7 @@ import ( "os/signal" "runtime" "strconv" + "sync" "syscall" "time" @@ -389,7 +390,7 @@ func (m *Command) SetupServer() error { pilosa.OptServerDiagnosticsInterval(diagnosticsInterval), pilosa.OptServerExecutorPoolSize(m.Config.WorkerPoolSize), pilosa.OptServerOpenTranslateStore(boltdb.OpenTranslateStore), - pilosa.OptServerOpenTranslateReader(http.GetOpenTranslateReaderFunc(c)), + pilosa.OptServerOpenTranslateReader(http.GetOpenTranslateReaderWithLockerFunc(c, &sync.Mutex{})), pilosa.OptServerLogger(m.logger), pilosa.OptServerAttrStoreFunc(boltdb.NewAttrStore), pilosa.OptServerSystemInfo(gopsutil.NewSystemInfo()), diff --git a/tx_test.go b/tx_test.go index 300e3444b..ae5e0b984 100644 --- a/tx_test.go +++ b/tx_test.go @@ -59,7 +59,10 @@ func queryBalances(m0api *pilosa.API, acctOwnerID uint64, fldAcct0, fldAcct1, in return } func skipForRoaring(t *testing.T) { - if strings.Contains(os.Getenv("PILOSA_TXSRC"), "roaring") { + src := os.Getenv("PILOSA_TXSRC") + // once txfactory.go DefaultTxsrc != RoaringTxn, this + // will break, of course. Take out the src == "" below. + if src == "" || strings.Contains(src, "roaring") { t.Skip("skip if roaring pseudo-txn involved -- won't show transactional rollback") } }