diff --git a/api.go b/api.go index d2c09fe85..4fca1a776 100644 --- a/api.go +++ b/api.go @@ -2188,6 +2188,138 @@ func (api *API) TranslateFieldDB(ctx context.Context, indexName, fieldName strin _, err := store.ReadFrom(rd) return err } +func (api *API) mutexCheckThisNode(ctx context.Context, qcx *Qcx, indexName string, fieldName string) (map[uint64]map[uint64][]uint64, error) { + index := api.holder.Index(indexName) + if index == nil { + return nil, newNotFoundError(ErrIndexNotFound, indexName) + } + field := index.Field(fieldName) + if field == nil { + return nil, newNotFoundError(ErrFieldNotFound, fieldName) + } + return field.MutexCheck(ctx, qcx) +} + +// mergeMutexCollisions adds collisions to an existing map if they aren't already +// present. It modifies dst. +func mergeMutexCollisions(dst, src map[uint64][]uint64) { + for k, v := range src { + if len(v) == 0 { + continue + } + existing := dst[k] + if len(existing) == 0 { + dst[k] = v + continue + } + // existing and v are both non-empty lists of collisions for this key. + // but if replication is working, both nodes should have the SAME list + // of collisions, so it's probably worth special-casing that check: + if len(existing) == len(v) { + different := false + for i := range existing { + if v[i] != existing[i] { + different = true + break + } + } + // yay, we can just ignore this + if !different { + continue + } + } + // combine... + existing = append(existing, v...) + // sort... + sort.Slice(existing, func(i, j int) bool { + return existing[i] < existing[j] + }) + // dedup. + n := 0 + prev := existing[0] + for i := 0; i < len(existing); i++ { + if existing[i] != prev { + existing[n] = existing[i] + n++ + } + prev = existing[i] + } + dst[k] = existing[:n] + } +} + +// MutexCheck checks for collisions in a given mutex field. The response is +// a map[shard]map[column]values. +func (api *API) MutexCheck(ctx context.Context, qcx *Qcx, indexName string, fieldName string, remote bool) (map[uint64]map[uint64][]uint64, error) { + if err := api.validate(apiMutexCheck); err != nil { + return nil, errors.Wrap(err, "validating api method") + } + // short path: if this is an internal remote request, only try to solve the + // question for these shards. + if remote { + out, err := api.mutexCheckThisNode(ctx, qcx, indexName, fieldName) + return out, err + } + var nodes []*Node + if !remote { + nodes = Nodes(api.cluster.nodes).Clone() + } else { + nodes = []*Node{api.cluster.nodeByID(api.server.nodeID)} + } + + /* + // request data from other nodes as well + snap := topology.NewClusterSnapshot(api.cluster.noder, api.cluster.Hasher, api.cluster.ReplicaN) + */ + eg, _ := errgroup.WithContext(ctx) + myID := api.server.nodeID + results := make([]map[uint64]map[uint64][]uint64, len(nodes)) + for i, node := range nodes { + i := i // loop variable shadowing is a war crime + if node.ID != myID { + node := node // loop variable shadowing again + eg.Go(func() (err error) { + results[i], err = api.server.defaultClient.MutexCheck(ctx, &node.URI, indexName, fieldName) + return err + }) + } else { + eg.Go(func() (err error) { + results[i], err = api.mutexCheckThisNode(ctx, qcx, indexName, fieldName) + return err + }) + } + } + err := eg.Wait() + if err != nil { + return nil, err + } + var out map[uint64]map[uint64][]uint64 + for _, nodeResults := range results { + if len(nodeResults) == 0 { + continue + } + if out == nil { + out = nodeResults + continue + } + for k, v := range nodeResults { + if len(v) == 0 { + continue + } + existing := out[k] + // results from two nodes. might be normal with replication. + if len(existing) == 0 { + out[k] = v + continue + } + // we have two maps for this shard. whee. + mergeMutexCollisions(existing, v) + // we don't store it back into out[k] because it was already + // a non-empty map, so stores to it update it. yay? + } + } + return out, nil +} type serverInfo struct { ShardWidth uint64 `json:"shardWidth"` @@ -2249,11 +2381,13 @@ const ( apiIDReserve apiIDCommit apiIDReset + apiMutexCheck //Backported from 1681 Caution ) var methodsCommon = map[apiMethod]struct{}{ apiClusterMessage: {}, apiSetCoordinator: {}, + apiMutexCheck: {}, } var methodsResizing = map[apiMethod]struct{}{ diff --git a/client.go b/client.go index 4cd410345..d04791e60 100644 --- a/client.go +++ b/client.go @@ -84,6 +84,8 @@ type InternalClient interface { GetNodeUsage(ctx context.Context, uri *URI) (map[string]NodeUsage, error) GetPastQueries(ctx context.Context, uri *URI) ([]PastQueryStatus, error) + + MutexCheck(ctx context.Context, uri *URI, index string, field string) (map[uint64]map[uint64][]uint64, error) } //=============== @@ -251,3 +253,7 @@ func (n nopInternalClient) GetNodeUsage(ctx context.Context, uri *URI) (map[stri func (n nopInternalClient) GetPastQueries(ctx context.Context, uri *URI) ([]PastQueryStatus, error) { return nil, nil } + +func (n nopInternalClient) MutexCheck(ctx context.Context, uri *URI, index, field string) (map[uint64]map[uint64][]uint64, error) { + return nil, nil +} diff --git a/field.go b/field.go index baec3a931..827e4484b 100644 --- a/field.go +++ b/field.go @@ -1229,6 +1229,23 @@ func (f *Field) Row(tx Tx, rowID uint64) (*Row, error) { } } +// mutexCheck performs a sanity-check on the available fragments for a +// field. The return is map[column]map[shard][]values for collisions only. +func (f *Field) MutexCheck(ctx context.Context, qcx *Qcx) (map[uint64]map[uint64][]uint64, error) { + if f.Type() != FieldTypeMutex { + return nil, errors.New("mutex check only valid for mutex fields") + } + f.mu.RLock() + defer f.mu.RUnlock() + standard := f.viewMap[viewStandard] + if standard == nil { + // no standard view present means we've never needed to create it, + // so it has no bits set, so it has no extra bits set. + return nil, nil + } + return standard.mutexCheck(ctx, qcx) +} + // SetBit sets a bit on a view within the field. func (f *Field) SetBit(tx Tx, rowID, colID uint64, t *time.Time) (changed bool, err error) { viewName := viewStandard diff --git a/field_internal_test.go b/field_internal_test.go index d187d5ab9..1a9dad798 100644 --- a/field_internal_test.go +++ b/field_internal_test.go @@ -27,6 +27,7 @@ import ( "github.com/pilosa/pilosa/v2/pql" "github.com/pilosa/pilosa/v2/roaring" + "github.com/pilosa/pilosa/v2/shardwidth" "github.com/pilosa/pilosa/v2/testhook" ) @@ -922,3 +923,27 @@ func TestBSIGroup_TxReopenDB(t *testing.T) { // the test: can we re-open a BSI fragment under Tx store _ = f.Reopen() } + +func CorruptAMutex(tb testing.TB, field *Field, qcx *Qcx) { + v := field.view(viewStandard) + if v == nil { + tb.Fatalf("creating view failed") + } + frags := v.allFragments() + for _, frag := range frags { + func() { + tx, finisher, err := qcx.GetTx(Txo{Write: true, Index: field.idx, Shard: frag.shard}) + defer finisher(&err) + if err != nil { + tb.Fatalf("getting tx: %v", err) + } + // set a bonus bit, bypassing the mutex handling + frag.mu.Lock() + _, err = frag.unprotectedSetBit(tx, 3, frag.shard< 0 { + return key.NeedData() + } + return key.RejectOne() +} + +func (b *BitmapMutexDupFilter) ConsiderData(key FilterKey, data *Container) FilterResult { + value, basePos := uint64(key)>>rowExponent, uint64(key&keyMask)<<16 + containerCallback(data, func(u uint16) { + pos := basePos + uint64(u) + if b.first[pos] != ^uint64(0) { + b.extra[pos+b.base] = append(b.extra[pos+b.base], value) + } else { + b.first[pos] = value + } + }) + return key.MatchOne() +} + +// Report returns the set of duplicate values identified. +func (b *BitmapMutexDupFilter) Report() map[uint64][]uint64 { + // copy values into extra, and remove them from first, so calling + // Report() again won't cause double-appends. + for k, v := range b.extra { + kpos := k % (1 << shardwidth.Exponent) + if b.first[kpos] != ^uint64(0) { + v = append(v, 0) + // prepend so the lowest value goes at the beginning + copy(v[1:], v[:]) + v[0] = b.first[kpos] + b.first[kpos] = ^uint64(0) + b.extra[k] = v + } + } + return b.extra +} + // ApplyFilterToIterator is a simplistic implementation that applies a bitmap // filter to a ContainerIterator, returning an error if it encounters an error. // diff --git a/roaring/filter_internal_test.go b/roaring/filter_internal_test.go index 0fc9c0fe0..1811acff4 100644 --- a/roaring/filter_internal_test.go +++ b/roaring/filter_internal_test.go @@ -349,3 +349,50 @@ func TestFilterWithRows(t *testing.T) { } } + +func TestMutexDupFilter(t *testing.T) { + tests := []struct{ + pairs [][2]uint64 + expect map[uint64][]uint64 + }{ + { + pairs: [][2]uint64{{0, 0}, {1, 0}, {0, 1}}, + expect: map[uint64][]uint64{0: {0, 1}}, + }, + { + pairs: [][2]uint64{{0, 0}, {1, 0}, {0, 1}, {0, 2}}, + expect: map[uint64][]uint64{0: {0, 1, 2}}, + }, + } + for num, test := range tests { + t.Run(fmt.Sprintf("case%d", num), func(t *testing.T) { + b := NewSliceBitmap() + for _, p := range test.pairs { + v := (p[1] << shardwidth.Exponent) | p[0] + b.DirectAdd(v) + } + dup := NewBitmapMutexDupFilter(0) + iter, _ := b.Containers.Iterator(0) + err := ApplyFilterToIterator(dup, iter) + if err != nil { + t.Fatalf("applying filter: %v", err) + } + expected := test.expect + got := dup.Report() + if len(expected) != len(got) { + t.Fatalf("expected %d entries in duplicate map, got %d", len(expected), len(got)) + } + for k, v := range expected { + gv := got[k] + if len(v) != len(gv) { + t.Fatalf("for id %d, expected %d (len %d), got %d (len %d)", k, v, len(v), gv, len(gv)) + } + for j := range v { + if gv[j] != v[j] { + t.Fatalf("for id %d, expected %d, got %d", k, v[j], gv[j]) + } + } + } + }) + } +} diff --git a/view.go b/view.go index 15173f5e4..0af36b71a 100644 --- a/view.go +++ b/view.go @@ -447,6 +447,50 @@ func (v *view) row(txOrig Tx, rowID uint64) (*Row, error) { } +// mutexCheck checks all available fragments for duplicate values. The return +// is map[column]map[shard][]values for collisions only. +func (v *view) mutexCheck(ctx context.Context, qcx *Qcx) (map[uint64]map[uint64][]uint64, error) { + // We don't need the context, we just want the context-awareness on the error groups. + // It would be nice if the inner functions could use this too... + eg, _ := errgroup.WithContext(ctx) + throttle := make(chan struct{}, runtime.NumCPU()) + frags := v.allFragments() + results := make([]map[uint64][]uint64, len(frags)) + for i, frag := range frags { + // local copies for the goroutine to use + i, frag := i, frag + eg.Go(func() error { + // limit simultaneous parallel goroutines associated with this + throttle <- struct{}{} + defer func() { + <-throttle + }() + tx, finisher, err := qcx.GetTx(Txo{Index: v.idx, Shard: frag.shard}) + if err != nil { + return err + } + defer finisher(&err) + results[i], err = frag.mutexCheck(tx) + if err != nil { + return err + } + return nil + }) + } + err := eg.Wait() + if err != nil { + return nil, err + } + out := map[uint64]map[uint64][]uint64{} + for i, result := range results { + if len(result) == 0 { + continue + } + out[frags[i].shard] = result + } + return out, nil +} + // setBit sets a bit within the view. func (v *view) setBit(txOrig Tx, rowID, columnID uint64) (changed bool, err error) { shard := columnID / ShardWidth