scriptoriumadmin/featurebase
1
0
Fork 0
mirror of https://github.com/featurebasedb/featurebase.git synced 2026-09-15 16:51:03 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions

make details optional and support limits on mutex checks

Browse source 
We support query parameters for details (default false) which
request additional data, and for a limit (default 0/MaxInt32)
on number of results returned to limit the amount of spam
produced if there's a lot of results. The simpler default
output should reduce load and runtime significantly, and the
ability to specify limits makes it easier to get reasonably
small responses.

There's some context support here, but the underlying filters
don't take contexts or check for them, which is probably
a flaw but might be a bit large to correct for this.
Despite being large, this set of changes is actually
fairly well contained within the mutex-checking code.
This commit is contained in:
Seebs 2021-09-08 11:59:47 -05:00
parent 484fcd8cf2
commit 26d38c0ee0
10 changed files with 248 additions and 60 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

110
api.go
View file

@ -2653,7 +2653,7 @@ func (api *API) RestoreShard(ctx context.Context, indexName string, shard uint64
return nil
}
func (api *API) mutexCheckThisNode(ctx context.Context, qcx *Qcx, indexName string, fieldName string) (map[uint64]map[uint64][]uint64, error) {
func (api *API) mutexCheckThisNode(ctx context.Context, qcx *Qcx, indexName string, fieldName string, details bool, limit int) (map[uint64]map[uint64][]uint64, error) {
index := api.holder.Index(indexName)
if index == nil {
return nil, newNotFoundError(ErrIndexNotFound, indexName)
@ -2662,7 +2662,26 @@ func (api *API) mutexCheckThisNode(ctx context.Context, qcx *Qcx, indexName stri
if field == nil {
return nil, newNotFoundError(ErrFieldNotFound, fieldName)
}
return field.MutexCheck(ctx, qcx)
results, err := field.MutexCheck(ctx, qcx, details, limit)
if err != nil {
return nil, err
}
if limit != 0 && len(results) > limit {
toDel := len(results) - limit
// yes, Go allows you to delete keys you've already seen while
// iterating a map. The spec says that if a value not-yet-reached
// is deleted during iteration, it may or may not appear; this
// carries the implication that deleting things during map iteration
// is safe.
for k := range results {
delete(results, k)
toDel--
if toDel == 0 {
break
}
}
}
return results, err
}
// mergeIDLists merges a list of numeric IDs into another list, removing
@ -2707,21 +2726,25 @@ func mergeKeyLists(dst []string, src []string) []string {
// MutexCheckNode checks for collisions in a given mutex field. The response is
// a map[shard]map[column]values, not translated.
func (api *API) MutexCheckNode(ctx context.Context, qcx *Qcx, indexName string, fieldName string) (map[uint64]map[uint64][]uint64, error) {
func (api *API) MutexCheckNode(ctx context.Context, qcx *Qcx, indexName string, fieldName string, details bool, limit int) (map[uint64]map[uint64][]uint64, error) {
if err := api.validate(apiMutexCheck); err != nil {
return nil, errors.Wrap(err, "validating api method")
}
return api.mutexCheckThisNode(ctx, qcx, indexName, fieldName)
return api.mutexCheckThisNode(ctx, qcx, indexName, fieldName, details, limit)
}
// MutexCheck checks a named field for mutex violations, returning a
// map of record IDs to values for records that have multiple values in the
// field. The return will be one of:
// details true:
// map[uint64][]uint64 // unkeyed index, unkeyed field
// map[uint64][]string // unkeyed index, keyed field
// map[string][]uint64 // keyed index, unkeyed field
// map[string][]string // keyed index, keyed field
func (api *API) MutexCheck(ctx context.Context, qcx *Qcx, indexName string, fieldName string) (result interface{}, err error) {
// details false:
// []uint64 // unkeyed index
// []string // keyed index
func (api *API) MutexCheck(ctx context.Context, qcx *Qcx, indexName string, fieldName string, details bool, limit int) (result interface{}, err error) {
if err = api.validate(apiMutexCheck); err != nil {
return nil, errors.Wrap(err, "validating api method")
}
@ -2746,12 +2769,12 @@ func (api *API) MutexCheck(ctx context.Context, qcx *Qcx, indexName string, fiel
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)
results[i], err = api.server.defaultClient.MutexCheck(ctx, &node.URI, indexName, fieldName, details, limit)
return err
})
} else {
eg.Go(func() (err error) {
results[i], err = api.mutexCheckThisNode(ctx, qcx, indexName, fieldName)
results[i], err = api.mutexCheckThisNode(ctx, qcx, indexName, fieldName, details, limit)
return err
})
}
@ -2760,12 +2783,18 @@ func (api *API) MutexCheck(ctx context.Context, qcx *Qcx, indexName string, fiel
if err != nil {
return nil, err
}
// Set this arbitrarily large so we don't have to be hand-checking for 0
// throughout.
if limit == 0 {
limit = math.MaxInt32
}
// We now have a series of maps from shards to maps of record IDs to
// values. But wait! Either the field, or the index, might be using keys,
// and want those translated. So we have to translate those. We'll create
// some tables.
useIndexKeys := index.Keys()
useFieldKeys := field.Keys()
// If we're not doing details, we won't translate field keys even if we could.
useFieldKeys := field.Keys() && details
var indexKeys = map[uint64]string{}
var fieldKeys = map[uint64]string{}
var indexIDs []uint64
@ -2778,8 +2807,9 @@ func (api *API) MutexCheck(ctx context.Context, qcx *Qcx, indexName string, fiel
// but what we can do is make a function which works with that map type
// given the raw integer values, and is a closure with an already-created
// map which has already been stashed in `result`. Because maps are
// reference-y, this should actually work.
var process func(uint64, []uint64)
// reference-y, this should actually work. This function returns true if
// it's hit the limit for length of results.
var process func(uint64, []uint64) bool
if useIndexKeys || useFieldKeys {
for _, nodeResults := range results {
for _, shardResults := range nodeResults {
@ -2843,11 +2873,27 @@ func (api *API) MutexCheck(ctx context.Context, qcx *Qcx, indexName string, fiel
// define the process functions. separated from above code just to make
// it easier to follow/compare them.
if useIndexKeys {
if useFieldKeys {
if !details {
outMap := make(map[uint64]struct{})
outStrings := []string{}
// unlike a map, the slice won't get updated-in-place, so we have
// to assign to result after we're done
defer func() {
result = outStrings
}()
process = func(recordID uint64, valueIDs []uint64) bool {
if _, ok := outMap[recordID]; ok {
return len(outMap) >= limit
}
outMap[recordID] = struct{}{}
outStrings = append(outStrings, indexKeys[recordID])
return len(outMap) >= limit
}
} else if useFieldKeys {
outMap := make(map[string][]string)
var valueKeys []string
result = outMap
process = func(recordID uint64, valueIDs []uint64) {
process = func(recordID uint64, valueIDs []uint64) bool {
valueKeys = valueKeys[:0]
for _, id := range valueIDs {
valueKeys = append(valueKeys, fieldKeys[id])
@ -2855,32 +2901,50 @@ func (api *API) MutexCheck(ctx context.Context, qcx *Qcx, indexName string, fiel
record := indexKeys[recordID]
if existing, ok := outMap[record]; ok {
outMap[record] = mergeKeyLists(existing, valueKeys)
} else {
} else if len(outMap) < limit {
// The append is so we can reuse this buffer safely,
// which matters if there's replication, because many
// cases won't need to copy the buffer, they'll just
// copy individual things from it.
outMap[record] = append([]string{}, valueKeys...)
}
return len(outMap) >= limit
}
} else {
outMap := make(map[string][]uint64)
result = outMap
process = func(recordID uint64, values []uint64) {
process = func(recordID uint64, values []uint64) bool {
record := indexKeys[recordID]
if existing, ok := outMap[record]; ok {
outMap[record] = mergeIDLists(existing, values)
} else {
} else if len(outMap) < limit {
outMap[record] = values
}
return len(outMap) >= limit
}
}
} else {
if useFieldKeys {
if !details {
outMap := make(map[uint64]struct{})
outIDs := []uint64{}
// unlike a map, the slice won't get updated-in-place, so we have
// to assign to result after we're done
defer func() {
result = outIDs
}()
process = func(recordID uint64, valueIDs []uint64) bool {
if _, ok := outMap[recordID]; ok {
return len(outMap) >= limit
}
outMap[recordID] = struct{}{}
outIDs = append(outIDs, recordID)
return len(outMap) >= limit
}
} else if useFieldKeys {
outMap := make(map[uint64][]string)
var valueKeys []string
result = outMap
process = func(record uint64, valueIDs []uint64) {
process = func(record uint64, valueIDs []uint64) bool {
valueKeys = valueKeys[:0]
for _, id := range valueIDs {
valueKeys = append(valueKeys, fieldKeys[id])
@ -2894,20 +2958,26 @@ func (api *API) MutexCheck(ctx context.Context, qcx *Qcx, indexName string, fiel
// copy individual things from it.
outMap[record] = append([]string{}, valueKeys...)
}
return len(outMap) >= limit
}
} else {
outMap := make(map[uint64][]uint64)
result = outMap
process = func(record uint64, values []uint64) {
process = func(record uint64, values []uint64) bool {
if existing, ok := outMap[record]; ok {
outMap[record] = mergeIDLists(existing, values)
} else {
outMap[record] = values
}
return len(outMap) >= limit
}
}
}
// if you specify a limit, and you have *different* errors on different
// nodes, we will not check all of the nodes. otherwise there's no practical
// way to get the primary benefit of specifying a limit.
processing:
for _, nodeResults := range results {
if len(nodeResults) == 0 {
continue
@ -2917,7 +2987,9 @@ func (api *API) MutexCheck(ctx context.Context, qcx *Qcx, indexName string, fiel
continue
}
for record, values := range v {
process(record, values)
if process(record, values) {
break processing
}
}
}
}

56
api_test.go
View file

@ -966,10 +966,6 @@ func TestAPI_MutexCheck(t *testing.T) {
qcx := m0.API.Txf().NewQcx()
defer qcx.Abort()
results, err := m0.API.MutexCheck(ctx, qcx, indexData.indexName, fieldData.fieldName)
if err != nil {
t.Fatalf("checking mutexes: %v", err)
}
// first two shards of each group of 4 should have a collision in
// position 1
expected := map[uint64]bool{
@ -980,6 +976,12 @@ func TestAPI_MutexCheck(t *testing.T) {
(8 << shardwidth.Exponent) + 1: true,
(9 << shardwidth.Exponent) + 1: true,
}
results, err := m0.API.MutexCheck(ctx, qcx, indexData.indexName, fieldData.fieldName, true, 0)
if err != nil {
t.Fatalf("checking mutexes: %v", err)
}
if keyedField {
mapped, ok := results.(map[uint64][]string)
if !ok {
@ -1014,9 +1016,29 @@ func TestAPI_MutexCheck(t *testing.T) {
}
}
if seen != len(expected) {
t.Fatalf("expected exactly %d records to have collisions", len(expected))
t.Fatalf("expected exactly %d records to have collisions, got %d", len(expected), seen)
}
}
// and let's try with no details and a limit of 3...
results, err = m0.API.MutexCheck(ctx, qcx, indexData.indexName, fieldData.fieldName, false, 3)
if err != nil {
t.Fatalf("checking mutexes: %v", err)
}
mapped, ok := results.([]uint64)
if !ok {
t.Fatalf("expected []uint64, got %T", results)
}
seen := 0
for _, k := range mapped {
seen++
if !expected[k] {
t.Fatalf("expected all collisions to be position 1 in shards (s %% 4 in [0,1]), got %d", k)
}
}
if seen != 3 {
t.Fatalf("expected results limited to 3, got %d", seen)
}
})
}
indexData = indexes[true]
@ -1111,7 +1133,7 @@ func TestAPI_MutexCheck(t *testing.T) {
qcx := m0.API.Txf().NewQcx()
defer qcx.Abort()
results, err := m0.API.MutexCheck(ctx, qcx, indexData.indexName, fieldData.fieldName)
results, err := m0.API.MutexCheck(ctx, qcx, indexData.indexName, fieldData.fieldName, true, 0)
if err != nil {
t.Fatalf("checking mutexes: %v", err)
}
@ -1155,6 +1177,28 @@ func TestAPI_MutexCheck(t *testing.T) {
t.Fatalf("expected exactly %d records to have collisions, got %d", len(expected), seen)
}
}
results, err = m0.API.MutexCheck(ctx, qcx, indexData.indexName, fieldData.fieldName, false, 3)
if err != nil {
t.Fatalf("checking mutexes: %v", err)
}
// this just sorta comes out this way with our hashing; these are
// the things which were in position 1 of their shards, and did
// not have a value which happens to map to 3.
mapped, ok := results.([]string)
if !ok {
t.Fatalf("expected []string, got %T", results)
}
seen := 0
for _, k := range mapped {
seen++
if _, ok := expected[k]; !ok {
t.Fatalf("unexpected collision on key %q", k)
}
}
if seen != 3 {
t.Fatalf("expected results limited to 3, got %d", len(expected))
}
})
}
}

4
client.go
View file

@ -80,7 +80,7 @@ type InternalClient interface {
RetrieveTranslatePartitionFromURI(ctx context.Context, index string, partition int, uri pnet.URI) (io.ReadCloser, error)
ImportRoaring(ctx context.Context, uri *pnet.URI, index, field string, shard uint64, remote bool, req *ImportRoaringRequest) error
ShardReader(ctx context.Context, index string, shard uint64) (io.ReadCloser, error)
MutexCheck(ctx context.Context, uri *pnet.URI, index string, field string) (map[uint64]map[uint64][]uint64, error)
MutexCheck(ctx context.Context, uri *pnet.URI, index string, field string, details bool, limit int) (map[uint64]map[uint64][]uint64, error)
IDAllocDataReader(ctx context.Context) (io.ReadCloser, error)
IndexTranslateDataReader(ctx context.Context, index string, partitionID int) (io.ReadCloser, error)
@ -214,7 +214,7 @@ func (n nopInternalClient) ImportRoaring(ctx context.Context, uri *pnet.URI, ind
return nil
}
func (n nopInternalClient) MutexCheck(ctx context.Context, uri *pnet.URI, index, field string) (map[uint64]map[uint64][]uint64, error) {
func (n nopInternalClient) MutexCheck(ctx context.Context, uri *pnet.URI, index, field string, details bool, limit int) (map[uint64]map[uint64][]uint64, error) {
return nil, nil
}

4
field.go
View file

@ -1094,7 +1094,7 @@ 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) {
func (f *Field) MutexCheck(ctx context.Context, qcx *Qcx, details bool, limit int) (map[uint64]map[uint64][]uint64, error) {
if f.Type() != FieldTypeMutex {
return nil, errors.New("mutex check only valid for mutex fields")
}
@ -1106,7 +1106,7 @@ func (f *Field) MutexCheck(ctx context.Context, qcx *Qcx) (map[uint64]map[uint64
// so it has no bits set, so it has no extra bits set.
return nil, nil
}
return standard.mutexCheck(ctx, qcx)
return standard.mutexCheck(ctx, qcx, details, limit)
}
// SetBit sets a bit on a view within the field.

4
fragment.go
View file

@ -601,8 +601,8 @@ func (f *fragment) closeStorage() error {
// mutexCheck checks for any entries in fragment which violate the mutex
// property of having only one value set for a given column ID.
func (f *fragment) mutexCheck(tx Tx) (map[uint64][]uint64, error) {
dup := roaring.NewBitmapMutexDupFilter(f.shard << shardwidth.Exponent)
func (f *fragment) mutexCheck(tx Tx, details bool, limit int) (map[uint64][]uint64, error) {
dup := roaring.NewBitmapMutexDupFilter(f.shard<<shardwidth.Exponent, details, limit)
err := tx.ApplyFilter(f.index(), f.field(), f.view(), f.shard, 0, dup)
if err != nil {
return nil, err

6
http/client.go
View file

@ -287,11 +287,13 @@ func (c *InternalClient) IngestOperations(ctx context.Context, uri *pnet.URI, in
// MutexCheck uses the mutex-check endpoint to request mutex collision data
// from a single node. It produces per-shard results, and does not translate
// them.
func (c *InternalClient) MutexCheck(ctx context.Context, uri *pnet.URI, indexName string, fieldName string) (map[uint64]map[uint64][]uint64, error) {
func (c *InternalClient) MutexCheck(ctx context.Context, uri *pnet.URI, indexName string, fieldName string, details bool, limit int) (map[uint64]map[uint64][]uint64, error) {
if uri == nil {
uri = c.defaultURI
}
u := uri.Path(fmt.Sprintf("/internal/index/%s/field/%s/mutex-check", indexName, fieldName))
// This is not actually a "Path", but reworking this to support queries
// is messier than I have resources to pursue just now.
u := uri.Path(fmt.Sprintf("/internal/index/%s/field/%s/mutex-check?details=%t&limit=%d", indexName, fieldName, details, limit))
req, err := http.NewRequest("GET", u, nil)
if err != nil {
return nil, errors.Wrap(err, "creating request")

26
http/handler.go
View file

@ -2765,9 +2765,20 @@ func (h *Handler) handleGetMutexCheck(w http.ResponseWriter, r *http.Request) {
// Get index and field type to determine how to handle the
// import data.
indexName, fieldName := mux.Vars(r)["index"], mux.Vars(r)["field"]
q := r.URL.Query()
limit := 0
details := q.Get("details") == "true"
limitStr := q.Get("limit")
if limitStr != "" {
var err error
limit, err = strconv.Atoi(limitStr)
if err != nil {
http.Error(w, "limit must be numeric", http.StatusBadRequest)
}
}
qcx := h.api.Txf().NewQcx()
defer qcx.Abort()
out, err := h.api.MutexCheck(r.Context(), qcx, indexName, fieldName)
out, err := h.api.MutexCheck(r.Context(), qcx, indexName, fieldName, details, limit)
if err != nil {
http.Error(w, err.Error(), http.StatusInternalServerError)
return
@ -2791,9 +2802,20 @@ func (h *Handler) handleInternalGetMutexCheck(w http.ResponseWriter, r *http.Req
// Get index and field type to determine how to handle the
// import data.
indexName, fieldName := mux.Vars(r)["index"], mux.Vars(r)["field"]
q := r.URL.Query()
limit := 0
details := q.Get("details") == "true"
limitStr := q.Get("limit")
if limitStr != "" {
var err error
limit, err = strconv.Atoi(limitStr)
if err != nil {
http.Error(w, "limit must be numeric", http.StatusBadRequest)
}
}
qcx := h.api.Txf().NewQcx()
defer qcx.Abort()
out, err := h.api.MutexCheckNode(r.Context(), qcx, indexName, fieldName)
out, err := h.api.MutexCheckNode(r.Context(), qcx, indexName, fieldName, details, limit)
if err != nil {
http.Error(w, err.Error(), http.StatusInternalServerError)
return

72
roaring/filter.go
View file

@ -767,18 +767,29 @@ func NewBitmapRangeFilter(min, max FilterKey, keyCallback func(FilterKey, int32)
// The slice is local-coordinates (first column 0), but the map is global
// coordinates (first column is whatever base was).
type BitmapMutexDupFilter struct {
base uint64 // the offset of 0 for this, used to accommodate shard offsets
extra map[uint64][]uint64 // extra values observed
first []uint64 // first values observed
base uint64 // the offset of 0 for this, used to accommodate shard offsets
extra map[uint64][]uint64 // extra values observed
first []uint64 // first values observed
details bool
limit int
done bool // if we have a limit, and we've hit it...
highKey FilterKey // ... we can stop after this many containers.
}
var _ BitmapFilter = &BitmapMutexDupFilter{}
func NewBitmapMutexDupFilter(base uint64) *BitmapMutexDupFilter {
func NewBitmapMutexDupFilter(base uint64, details bool, limit int) *BitmapMutexDupFilter {
filter := &BitmapMutexDupFilter{
base: base,
extra: map[uint64][]uint64{},
first: make([]uint64, 1<<shardwidth.Exponent),
base: base,
extra: map[uint64][]uint64{},
first: make([]uint64, 1<<shardwidth.Exponent),
details: details,
limit: limit,
}
if filter.limit == 0 {
// A limit of 0 is not a limit; set limit higher than possible number of
// values we could have.
filter.limit = 2 << shardwidth.Exponent
}
for i := range filter.first {
filter.first[i] = ^uint64(0)
@ -798,27 +809,52 @@ func (b *BitmapMutexDupFilter) ConsiderData(key FilterKey, data *Container) Filt
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)
if b.details {
b.extra[pos+b.base] = append(b.extra[pos+b.base], value)
} else {
// no details, just annotate that it exists
b.extra[pos+b.base] = []uint64{}
}
} else {
b.first[pos] = value
}
})
if len(b.extra) >= b.limit {
if !b.done {
// we note which container we found the last value we needed in.
// We may still go over the limit, but we won't look at any *more*
// containers in this row.
//
// We can't just abort early because the records we already found
// could have more values.
b.done = true
b.highKey = key & keyMask
return key.RejectRow()
}
if (key & keyMask) >= b.highKey {
return key.RejectRow()
}
}
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
// Report() again won't cause double-appends. We only have to do
// this if we've been asked for details; otherwise the list of
// known positions is sufficient.
if b.details {
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

10
roaring/filter_internal_test.go
View file

@ -350,16 +350,16 @@ func TestFilterWithRows(t *testing.T) {
}
func TestMutexDupFilter(t *testing.T) {
tests := []struct{
pairs [][2]uint64
tests := []struct {
pairs [][2]uint64
expect map[uint64][]uint64
}{
{
pairs: [][2]uint64{{0, 0}, {1, 0}, {0, 1}},
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}},
pairs: [][2]uint64{{0, 0}, {1, 0}, {0, 1}, {0, 2}},
expect: map[uint64][]uint64{0: {0, 1, 2}},
},
}
@ -370,7 +370,7 @@ func TestMutexDupFilter(t *testing.T) {
v := (p[1] << shardwidth.Exponent) | p[0]
b.DirectAdd(v)
}
dup := NewBitmapMutexDupFilter(0)
dup := NewBitmapMutexDupFilter(0, true, 9)
iter, _ := b.Containers.Iterator(0)
err := ApplyFilterToIterator(dup, iter)
if err != nil {

16
view.go
View file

@ -17,6 +17,7 @@ package pilosa
import (
"context"
"fmt"
"math"
"os"
"path/filepath"
"runtime"
@ -444,7 +445,7 @@ 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) {
func (v *view) mutexCheck(ctx context.Context, qcx *Qcx, details bool, limit int) (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)
@ -465,7 +466,7 @@ func (v *view) mutexCheck(ctx context.Context, qcx *Qcx) (map[uint64]map[uint64]
return err
}
defer finisher(&err)
results[i], err = frag.mutexCheck(tx)
results[i], err = frag.mutexCheck(tx, details, limit)
if err != nil {
return err
}
@ -477,11 +478,22 @@ func (v *view) mutexCheck(ctx context.Context, qcx *Qcx) (map[uint64]map[uint64]
return nil, err
}
out := map[uint64]map[uint64][]uint64{}
// We would use MaxInt here, but it's new with go 1.17. In practice if
// you have 2 billion duplicates you're sorta screwed anyway.
if limit == 0 {
limit = math.MaxInt32
}
count := 0
for i, result := range results {
if len(result) == 0 {
continue
}
out[frags[i].shard] = result
count += len(result)
// if we have enough, stop
if count > limit {
break
}
}
return out, nil
}