From bd33e373937014159af58e77c3660eae3cfdbd1e Mon Sep 17 00:00:00 2001 From: Alexis de Treglode Date: Tue, 8 Sep 2026 22:01:57 -0700 Subject: [PATCH 1/3] Lock shared maps that raced under concurrent reads. Statelet column names, collector parent indexes, predicate registry, and session cache JSON could fatal with concurrent map iteration or write. --- service/executor/expand/data_unit.go | 7 ++++ service/executor/expand/data_unit_test.go | 35 +++++++++++++++++ service/session/cache.go | 10 ++++- service/session/cache_test.go | 39 +++++++++++++++++++ view/cache.go | 4 +- view/collector.go | 33 +++++++++++++++- view/collector_concurrent_test.go | 41 ++++++++++++++++++++ view/extension/predicates.go | 4 ++ view/extension/predicates_test.go | 35 +++++++++++++++++ view/state.go | 23 ++++++++--- view/state/types.go | 4 +- view/state/types_test.go | 47 +++++++++++++++++++++++ view/state_test.go | 33 ++++++++++++++++ 13 files changed, 305 insertions(+), 10 deletions(-) create mode 100644 service/executor/expand/data_unit_test.go create mode 100644 service/session/cache_test.go create mode 100644 view/collector_concurrent_test.go create mode 100644 view/state/types_test.go diff --git a/service/executor/expand/data_unit.go b/service/executor/expand/data_unit.go index 2f39ac0b2..ef26e56e5 100644 --- a/service/executor/expand/data_unit.go +++ b/service/executor/expand/data_unit.go @@ -173,6 +173,8 @@ func (c *DataUnit) Next() (interface{}, error) { } func (c *DataUnit) ensureSliceIndex() { + c.mu.Lock() + defer c.mu.Unlock() if c.sliceIndex != nil { return } @@ -181,6 +183,11 @@ func (c *DataUnit) ensureSliceIndex() { } func (c *DataUnit) xunsafeSlice(valueType reflect.Type) *xunsafe.Slice { + c.mu.Lock() + defer c.mu.Unlock() + if c.sliceIndex == nil { + c.sliceIndex = map[reflect.Type]*xunsafe.Slice{} + } slice, ok := c.sliceIndex[valueType] if !ok { slice = xunsafe.NewSlice(reflect.SliceOf(valueType)) diff --git a/service/executor/expand/data_unit_test.go b/service/executor/expand/data_unit_test.go new file mode 100644 index 000000000..a82a0d252 --- /dev/null +++ b/service/executor/expand/data_unit_test.go @@ -0,0 +1,35 @@ +package expand + +import ( + "reflect" + "sync" + "sync/atomic" + "testing" + "time" +) + +// TestDataUnitConcurrentSliceIndex reproduces concurrent map write on +// DataUnit.sliceIndex. xunsafeSlice writes the map with no lock. +func TestDataUnitConcurrentSliceIndex(t *testing.T) { + deadline := time.Now().Add(3 * time.Second) + unit := &DataUnit{} + unit.ensureSliceIndex() + var next atomic.Uint64 + var waitGroup sync.WaitGroup + waitGroup.Add(5) + + run := func(fn func()) { + defer waitGroup.Done() + for time.Now().Before(deadline) { + fn() + } + } + + for worker := 0; worker < 5; worker++ { + go run(func() { + n := int(next.Add(1)%64) + 1 + _ = unit.xunsafeSlice(reflect.ArrayOf(n, reflect.TypeOf(byte(0)))) + }) + } + waitGroup.Wait() +} diff --git a/service/session/cache.go b/service/session/cache.go index fa5a60f08..b0a193210 100644 --- a/service/session/cache.go +++ b/service/session/cache.go @@ -24,10 +24,18 @@ func (c *cache) lookup(parameter *state.Parameter) (interface{}, bool) { } func (s *Session) MarshalJSON() ([]byte, error) { - return json.Marshal(s.cache.values) + s.cache.RWMutex.RLock() + snapshot := make(map[string]interface{}, len(s.cache.values)) + for key, value := range s.cache.values { + snapshot[key] = value + } + s.cache.RWMutex.RUnlock() + return json.Marshal(snapshot) } func (s *Session) Unmarshal(parameters state.Parameters, data []byte) error { + s.cache.RWMutex.Lock() + defer s.cache.RWMutex.Unlock() err := json.Unmarshal(data, &s.cache.values) if err != nil { return err diff --git a/service/session/cache_test.go b/service/session/cache_test.go new file mode 100644 index 000000000..d23f8020b --- /dev/null +++ b/service/session/cache_test.go @@ -0,0 +1,39 @@ +package session + +import ( + "fmt" + "sync" + "sync/atomic" + "testing" + "time" + + "github.com/viant/datly/view/state" +) + +// TestSessionMarshalJSONConcurrentCachePut reproduces concurrent map +// iteration and write on session cache.values. MarshalJSON ranges the map +// while put writes it. +func TestSessionMarshalJSONConcurrentCachePut(t *testing.T) { + deadline := time.Now().Add(3 * time.Second) + s := &Session{cache: newCache()} + var next atomic.Uint64 + var waitGroup sync.WaitGroup + waitGroup.Add(5) + + run := func(fn func()) { + defer waitGroup.Done() + for time.Now().Before(deadline) { + fn() + } + } + + go run(func() { + s.cache.put(&state.Parameter{Name: fmt.Sprintf("p%d", next.Add(1))}, 1) + }) + for worker := 0; worker < 4; worker++ { + go run(func() { + _, _ = s.MarshalJSON() + }) + } + waitGroup.Wait() +} diff --git a/view/cache.go b/view/cache.go index 51a4e1683..884c1c5c0 100644 --- a/view/cache.go +++ b/view/cache.go @@ -1002,6 +1002,8 @@ func (c *Cache) applyWarmupFieldNames(selector *Statelet, fieldNames []string) { if selector == nil || c.owner == nil || len(fieldNames) == 0 { return } + selector.columnNamesMu.Lock() + defer selector.columnNamesMu.Unlock() if selector._columnNames == nil { selector._columnNames = map[string]bool{} } @@ -1019,7 +1021,7 @@ func (c *Cache) applyWarmupFieldNames(selector *Statelet, fieldNames []string) { if outputName == "" { outputName = columnName } - if selector.Has(columnName) || selector.Has(outputName) { + if selector._columnNames[columnName] || selector._columnNames[outputName] { continue } selector._columnNames[columnName] = true diff --git a/view/collector.go b/view/collector.go index 9e3bc24fe..841d22572 100644 --- a/view/collector.go +++ b/view/collector.go @@ -26,6 +26,8 @@ type compositeKey string type Collector struct { Id string mutex sync.Mutex + indexOnce sync.Once + indexMu *sync.Mutex parent *Collector destValue reflect.Value appender *xunsafe.Appender @@ -154,6 +156,7 @@ func (r *Collector) Clone() *Collector { slicePtrValue.Elem().Set(dest) return &Collector{ Id: uuid.New().String(), + indexMu: r.indexMu, parent: r.parent, destValue: slicePtrValue, appender: r.slice.Appender(xunsafe.ValuePointer(&slicePtrValue)), @@ -187,6 +190,7 @@ func (r *Collector) Lock() *sync.Mutex { // Resolve resolved unmapped column func (r *Collector) Resolve(column io.Column) func(ptr unsafe.Pointer) interface{} { + r.lockIndex() buffer, ok := r.values[column.Name()] if !ok { localSlice := make([]interface{}, 0) @@ -201,6 +205,7 @@ func (r *Collector) Resolve(column io.Column) func(ptr unsafe.Pointer) interface scanType = reflect.TypeOf(0) } r.types[column.Name()] = xunsafe.NewType(scanType) + r.unlockIndex() return func(ptr unsafe.Pointer) interface{} { var valuePtr interface{} switch kind { @@ -228,6 +233,8 @@ func (r *Collector) Resolve(column io.Column) func(ptr unsafe.Pointer) interface // parentValuesPositions returns positions in the parent main slice by given column name // After first use, it is not possible to index new resolved column indexes by Resolve method func (r *Collector) parentValuesPositions(ns string, columnName string) map[interface{}][]int { + r.parent.lockIndex() + defer r.parent.unlockIndex() columnValues, ok := r.parent.valuePosition[ns] if !ok { columnValues = map[string]map[interface{}][]int{} @@ -242,6 +249,8 @@ func (r *Collector) parentValuesPositions(ns string, columnName string) map[inte } func (r *Collector) parentCompositePositions(relation *Relation) map[compositeKey][]int { + r.parent.lockIndex() + defer r.parent.unlockIndex() signature := relationCompositeSignature(relation.On) result, ok := r.parent.compositeValuePosition[signature] if !ok || len(result) == 0 { @@ -251,6 +260,23 @@ func (r *Collector) parentCompositePositions(relation *Relation) map[compositeKe return result } +func (r *Collector) indexLocker() *sync.Mutex { + r.indexOnce.Do(func() { + if r.indexMu == nil { + r.indexMu = &sync.Mutex{} + } + }) + return r.indexMu +} + +func (r *Collector) lockIndex() { + r.indexLocker().Lock() +} + +func (r *Collector) unlockIndex() { + r.indexLocker().Unlock() +} + // NewCollector creates a collector func NewCollector(slice *xunsafe.Slice, view *View, dest interface{}, viewMetaHandler viewSummaryHandlerFn, readAll bool) *Collector { ensuredDest := ensureDest(dest, view) @@ -258,6 +284,7 @@ func NewCollector(slice *xunsafe.Slice, view *View, dest interface{}, viewMetaHa wg.Add(1) return &Collector{ Id: uuid.New().String(), + indexMu: &sync.Mutex{}, destValue: reflect.ValueOf(ensuredDest), valuePosition: make(map[string]map[string]map[interface{}][]int), compositeValuePosition: make(map[string]map[compositeKey][]int), @@ -376,6 +403,8 @@ func (r *Collector) valueIndexer(ctx context.Context, visitorRelations []*Relati } func (r *Collector) indexCompositeValueByRel(ptr unsafe.Pointer, rel *Relation, counter int) { + r.lockIndex() + defer r.unlockIndex() signature := relationCompositeSignature(rel.On) index := r.compositeValuePosition[signature] if index == nil { @@ -424,7 +453,8 @@ func (r *Collector) indexValueByRel(fieldValue interface{}, rel *Relation, count // 6c0d0 func (r *Collector) indexValueToPosition(rel *Relation, fieldValue interface{}, counter int) { - + r.lockIndex() + defer r.unlockIndex() for _, item := range rel.On { columnValues, ok := r.valuePosition[item.Namespace] if !ok { @@ -805,6 +835,7 @@ func (r *Collector) Relations(selector *Statelet) ([]*Collector, error) { } result[counter] = &Collector{ Id: uuid.New().String(), + indexMu: &sync.Mutex{}, parent: r, viewMetaHandler: aHandler, destValue: destPtr, diff --git a/view/collector_concurrent_test.go b/view/collector_concurrent_test.go new file mode 100644 index 000000000..b31d2312a --- /dev/null +++ b/view/collector_concurrent_test.go @@ -0,0 +1,41 @@ +package view + +import ( + "fmt" + "reflect" + "sync" + "sync/atomic" + "testing" + "time" + + "github.com/viant/datly/view/state" +) + +// TestCollectorConcurrentParentValuePositions reproduces concurrent map +// iteration and write on Collector.valuePosition. Child collectors write the +// parent map while another goroutine ranges it. +func TestCollectorConcurrentParentValuePositions(t *testing.T) { + deadline := time.Now().Add(3 * time.Second) + parentView := &View{Schema: state.NewSchema(reflect.TypeOf([]*compositeParentRow{}))} + parentDest := []*compositeParentRow{} + parent := NewCollector(parentView.Schema.Slice(), parentView, &parentDest, nil, false) + child := &Collector{parent: parent} + var next atomic.Uint64 + var waitGroup sync.WaitGroup + waitGroup.Add(5) + + run := func(fn func()) { + defer waitGroup.Done() + for time.Now().Before(deadline) { + fn() + } + } + + for worker := 0; worker < 5; worker++ { + go run(func() { + ns := fmt.Sprintf("ns%d", next.Add(1)) + _ = child.parentValuesPositions(ns, "col") + }) + } + waitGroup.Wait() +} diff --git a/view/extension/predicates.go b/view/extension/predicates.go index bf39c75dc..b8d8467b8 100644 --- a/view/extension/predicates.go +++ b/view/extension/predicates.go @@ -75,7 +75,9 @@ type ( ) func (r *PredicateRegistry) Lookup(name string) (*Predicate, error) { + r.Lock() result, ok := r.registry[name] + r.Unlock() if ok { return result, nil } @@ -94,6 +96,8 @@ func (r *PredicateRegistry) Scope() *PredicateRegistry { } func (r *PredicateRegistry) Add(template *predicate.Template) { + r.Lock() + defer r.Unlock() r.registry[template.Name] = &Predicate{ Template: template, } diff --git a/view/extension/predicates_test.go b/view/extension/predicates_test.go index 52fdac590..990aaae66 100644 --- a/view/extension/predicates_test.go +++ b/view/extension/predicates_test.go @@ -1,8 +1,14 @@ package extension import ( + "fmt" "strings" + "sync" + "sync/atomic" "testing" + "time" + + "github.com/viant/xdatly/predicate" ) func TestNewDurationPredicate_DoesNotUseLogicalOrInVelty(t *testing.T) { @@ -34,3 +40,32 @@ func TestDefaultExtensionRegistersNopPredicate(t *testing.T) { t.Fatalf("unexpected registered nop predicate: %#v", predicate) } } + +// TestPredicateRegistryConcurrentLookupAndAdd reproduces concurrent map access +// on PredicateRegistry.registry. Lookup reads while Add writes, and the +// embedded mutex is unused. +func TestPredicateRegistryConcurrentLookupAndAdd(t *testing.T) { + deadline := time.Now().Add(3 * time.Second) + registry := NewPredicates() + var next atomic.Uint64 + var waitGroup sync.WaitGroup + waitGroup.Add(5) + + run := func(fn func()) { + defer waitGroup.Done() + for time.Now().Before(deadline) { + fn() + } + } + + go run(func() { + name := fmt.Sprintf("pred%d", next.Add(1)) + registry.Add(&predicate.Template{Name: name, Source: "1 = 1"}) + }) + for worker := 0; worker < 4; worker++ { + go run(func() { + _, _ = registry.Lookup(fmt.Sprintf("pred%d", next.Add(1))) + }) + } + waitGroup.Wait() +} diff --git a/view/state.go b/view/state.go index 6bc6b6e79..60ab32c03 100644 --- a/view/state.go +++ b/view/state.go @@ -23,6 +23,7 @@ type ( state.QuerySelector QuerySettings filtersMu sync.Mutex + columnNamesMu sync.Mutex initialized bool WarmupNoLimit bool _columnNames map[string]bool @@ -46,21 +47,30 @@ func (s *Statelet) Init(aView *View) { if s.initialized { return } + s.columnNamesMu.Lock() s._columnNames = Names(s.Columns).Index() + s.columnNamesMu.Unlock() } // Has checks if Field is present in Template.Columns func (s *Statelet) Has(field string) bool { + s.columnNamesMu.Lock() + defer s.columnNamesMu.Unlock() _, ok := s._columnNames[field] return ok } func (s *Statelet) Add(fieldName string, isHolder bool) { toLower := strings.ToLower(fieldName) + s.columnNamesMu.Lock() + defer s.columnNamesMu.Unlock() if _, ok := s._columnNames[toLower]; ok { return } + if s._columnNames == nil { + s._columnNames = map[string]bool{} + } s._columnNames[toLower] = true s._columnNames[fieldName] = true @@ -74,6 +84,8 @@ func (s *Statelet) Add(fieldName string, isHolder bool) { } func (s *Statelet) SetColumns(columns []string) { + s.columnNamesMu.Lock() + defer s.columnNamesMu.Unlock() s.Columns = append([]string(nil), columns...) s._columnNames = Names(s.Columns).Index() } @@ -187,6 +199,7 @@ func (s *Statelet) CloneForSummary() *Statelet { Ignore: s.Ignore, } + s.columnNamesMu.Lock() if s._columnNames != nil { ret._columnNames = make(map[string]bool, len(s._columnNames)) for k, v := range s._columnNames { @@ -195,17 +208,17 @@ func (s *Statelet) CloneForSummary() *Statelet { } else { ret._columnNames = map[string]bool{} } - - s.filtersMu.Lock() - ret.Filters = append(predicate.Filters(nil), s.Filters...) - s.filtersMu.Unlock() - if len(s.Fields) > 0 { ret.Fields = append([]string(nil), s.Fields...) } if len(s.Columns) > 0 { ret.Columns = append([]string(nil), s.Columns...) } + s.columnNamesMu.Unlock() + + s.filtersMu.Lock() + ret.Filters = append(predicate.Filters(nil), s.Filters...) + s.filtersMu.Unlock() return ret } diff --git a/view/state/types.go b/view/state/types.go index f301290ea..40adf574e 100644 --- a/view/state/types.go +++ b/view/state/types.go @@ -11,12 +11,12 @@ type Types struct { } func (c *Types) Lookup(p reflect.Type) (*Type, bool) { + c.RWMutex.RLock() + defer c.RWMutex.RUnlock() if len(c.types) == 0 { return nil, false } - c.RWMutex.RLock() ret, ok := c.types[p] - c.RWMutex.RUnlock() return ret, ok } diff --git a/view/state/types_test.go b/view/state/types_test.go new file mode 100644 index 000000000..78235b37a --- /dev/null +++ b/view/state/types_test.go @@ -0,0 +1,47 @@ +package state + +import ( + "fmt" + "reflect" + "sync" + "sync/atomic" + "testing" + "time" +) + +func uniqueType(n uint64) reflect.Type { + return reflect.StructOf([]reflect.StructField{{ + Name: "F", + Type: reflect.TypeOf(0), + Tag: reflect.StructTag(fmt.Sprintf(`json:"f%d"`, n)), + }}) +} + +// TestTypesConcurrentLookupAndPut reproduces concurrent map access on Types.types. +// Lookup reads the map (including an unlocked len) while Put writes it. +func TestTypesConcurrentLookupAndPut(t *testing.T) { + deadline := time.Now().Add(3 * time.Second) + registry := NewTypes() + var next atomic.Uint64 + var waitGroup sync.WaitGroup + waitGroup.Add(8) + + run := func(fn func()) { + defer waitGroup.Done() + for time.Now().Before(deadline) { + fn() + } + } + + for writer := 0; writer < 4; writer++ { + go run(func() { + registry.Put(&Type{Schema: &Schema{rType: uniqueType(next.Add(1))}}) + }) + } + for worker := 0; worker < 4; worker++ { + go run(func() { + _, _ = registry.Lookup(uniqueType(next.Add(1))) + }) + } + waitGroup.Wait() +} diff --git a/view/state_test.go b/view/state_test.go index 51814ee09..bbfdd23e2 100644 --- a/view/state_test.go +++ b/view/state_test.go @@ -1,8 +1,11 @@ package view import ( + "fmt" "sync" + "sync/atomic" "testing" + "time" "github.com/viant/datly/view/state/predicate" ) @@ -35,3 +38,33 @@ func TestStateletCloneForSummaryConcurrentFilters(t *testing.T) { waitGroup.Wait() } + +// TestStateletCloneForSummaryConcurrentColumnNames reproduces the production +// fatal "concurrent map iteration and map write" on Statelet._columnNames. +// CloneForSummary ranges the map while Add writes it. Surviving the 3s stress +// window is the assertion that the race is gone. +func TestStateletCloneForSummaryConcurrentColumnNames(t *testing.T) { + deadline := time.Now().Add(3 * time.Second) + statelet := NewStatelet() + var next atomic.Uint64 + var waitGroup sync.WaitGroup + waitGroup.Add(5) + + run := func(fn func()) { + defer waitGroup.Done() + for time.Now().Before(deadline) { + fn() + } + } + + // One writer so we do not crash on concurrent Add lookups first. + go run(func() { + statelet.Add(fmt.Sprintf("Field%d", next.Add(1)), true) + }) + for worker := 0; worker < 4; worker++ { + go run(func() { + _ = statelet.CloneForSummary() + }) + } + waitGroup.Wait() +} From d104f61be6bd785773737b89cd3cd0006529c846 Mon Sep 17 00:00:00 2001 From: Alexis de Treglode Date: Thu, 10 Sep 2026 11:08:19 -0700 Subject: [PATCH 2/3] Clone selector slices under lock during summary copy. Copying the embedded QuerySelector before columnNamesMu raced with Add on Fields and Columns; ViewMetaHandler also read valuePosition without the collector index lock. --- view/collector.go | 31 +++++++++++++++++-------------- view/state.go | 11 ++++++++++- 2 files changed, 27 insertions(+), 15 deletions(-) diff --git a/view/collector.go b/view/collector.go index 841d22572..5ebe9edab 100644 --- a/view/collector.go +++ b/view/collector.go @@ -880,24 +880,27 @@ func (r *Collector) ViewMetaHandler(rel *Relation) (func(viewMeta interface{}) e return nil, fmt.Errorf("not found holder field %v at %v", templateMeta.Name, templateMeta.Schema.Type().String()) } - var valuesPosition map[interface{}][]int return func(viewMeta interface{}) error { + viewMetaPtr := xunsafe.AsPointer(viewMeta) + if viewMetaPtr == nil { + return nil + } + value := io.NormalizeKey(metaChildKeyField.Value(viewMetaPtr)) + for _, item := range rel.On { - if valuesPosition == nil { - if r.valuePosition[item.Namespace] == nil { - r.valuePosition[item.Namespace] = map[string]map[interface{}][]int{} - } - valuesPosition = r.valuePosition[item.Namespace][item.Column] + r.lockIndex() + namespaceIndex := r.valuePosition[item.Namespace] + if namespaceIndex == nil { + namespaceIndex = map[string]map[interface{}][]int{} + r.valuePosition[item.Namespace] = namespaceIndex } - - viewMetaPtr := xunsafe.AsPointer(viewMeta) - if viewMetaPtr == nil { - return nil + columnIndex := namespaceIndex[item.Column] + var positions []int + if found, ok := columnIndex[value]; ok { + positions = append([]int(nil), found...) } - - value := io.NormalizeKey(metaChildKeyField.Value(viewMetaPtr)) - positions, ok := valuesPosition[value] - if !ok { + r.unlockIndex() + if len(positions) == 0 { return nil } diff --git a/view/state.go b/view/state.go index 60ab32c03..5fefe343b 100644 --- a/view/state.go +++ b/view/state.go @@ -191,12 +191,21 @@ func (s *Statelet) CloneForSummary() *Statelet { DatabaseFormat: s.DatabaseFormat, OutputFormat: s.OutputFormat, Template: s.Template, - QuerySelector: s.QuerySelector, QuerySettings: s.QuerySettings, initialized: s.initialized, WarmupNoLimit: s.WarmupNoLimit, result: s.result, Ignore: s.Ignore, + QuerySelector: state.QuerySelector{ + OrderBy: s.OrderBy, + Offset: s.Offset, + Limit: s.Limit, + Page: s.Page, + Criteria: s.Criteria, + }, + } + if len(s.Placeholders) > 0 { + ret.Placeholders = append([]interface{}(nil), s.Placeholders...) } s.columnNamesMu.Lock() From ec56fb20af47c396aa61b10fb8ae0836a04923b2 Mon Sep 17 00:00:00 2001 From: Alexis de Treglode Date: Thu, 10 Sep 2026 11:55:37 -0700 Subject: [PATCH 3/3] Copy collector parent-index positions under lock. Returning the live valuePosition maps after unlock left child relation matching reading while indexValueToPosition still wrote. --- view/collector.go | 105 ++++++++++++++++++------------ view/collector_concurrent_test.go | 28 ++++++-- 2 files changed, 86 insertions(+), 47 deletions(-) diff --git a/view/collector.go b/view/collector.go index 5ebe9edab..8012f5860 100644 --- a/view/collector.go +++ b/view/collector.go @@ -230,11 +230,8 @@ func (r *Collector) Resolve(column io.Column) func(ptr unsafe.Pointer) interface } } -// parentValuesPositions returns positions in the parent main slice by given column name -// After first use, it is not possible to index new resolved column indexes by Resolve method -func (r *Collector) parentValuesPositions(ns string, columnName string) map[interface{}][]int { - r.parent.lockIndex() - defer r.parent.unlockIndex() +// parentColumnIndex returns the live parent column index. Caller must hold the parent index lock. +func (r *Collector) parentColumnIndex(ns string, columnName string) map[interface{}][]int { columnValues, ok := r.parent.valuePosition[ns] if !ok { columnValues = map[string]map[interface{}][]int{} @@ -248,9 +245,33 @@ func (r *Collector) parentValuesPositions(ns string, columnName string) map[inte return result } -func (r *Collector) parentCompositePositions(relation *Relation) map[compositeKey][]int { +// parentPositionsFor copies parent slice positions for a key while the parent index lock is held. +func (r *Collector) parentPositionsFor(ns string, columnName string, key interface{}) []int { + r.parent.lockIndex() + defer r.parent.unlockIndex() + found, ok := r.parentColumnIndex(ns, columnName)[key] + if !ok { + return nil + } + return append([]int(nil), found...) +} + +// parentPositionKeys copies parent index keys for a column while the parent index lock is held. +func (r *Collector) parentPositionKeys(ns string, columnName string) []interface{} { r.parent.lockIndex() defer r.parent.unlockIndex() + index := r.parentColumnIndex(ns, columnName) + if len(index) == 0 { + return nil + } + keys := make([]interface{}, 0, len(index)) + for key := range index { + keys = append(keys, key) + } + return keys +} + +func (r *Collector) parentCompositeIndex(relation *Relation) map[compositeKey][]int { signature := relationCompositeSignature(relation.On) result, ok := r.parent.compositeValuePosition[signature] if !ok || len(result) == 0 { @@ -260,6 +281,17 @@ func (r *Collector) parentCompositePositions(relation *Relation) map[compositeKe return result } +// parentCompositePositionsFor copies parent slice positions for a composite key under the parent index lock. +func (r *Collector) parentCompositePositionsFor(relation *Relation, key compositeKey) []int { + r.parent.lockIndex() + defer r.parent.unlockIndex() + found, ok := r.parentCompositeIndex(relation)[key] + if !ok { + return nil + } + return append([]int(nil), found...) +} + func (r *Collector) indexLocker() *sync.Mutex { r.indexOnce.Do(func() { if r.indexMu == nil { @@ -487,8 +519,8 @@ func (r *Collector) visitorOne(relation *Relation) func(value interface{}) error } keyParts = append(keyParts, io.NormalizeKey(link.xField.Interface(xunsafe.AsPointer(owner)))) } - positions, ok := r.parentCompositePositions(relation)[buildCompositeKey(keyParts)] - if !ok { + positions := r.parentCompositePositionsFor(relation, buildCompositeKey(keyParts)) + if len(positions) == 0 { return nil } for _, index := range positions { @@ -505,9 +537,8 @@ func (r *Collector) visitorOne(relation *Relation) func(value interface{}) error aKey = io.NormalizeKey(aKey) parentLink := relation.On[j] - valuePosition := r.parentValuesPositions(parentLink.Namespace, parentLink.Column) - positions, ok := valuePosition[aKey] - if !ok { + positions := r.parentPositionsFor(parentLink.Namespace, parentLink.Column, aKey) + if len(positions) == 0 { return nil } for _, index := range positions { @@ -549,10 +580,9 @@ func (r *Collector) ParentRow(relation *Relation) func(value interface{}) (inter } else { key = xType.Deref((*values)[r.manyCounter]) } - valuePosition := r.parentValuesPositions(namespace, column) key = io.NormalizeKey(key) - positions, ok := valuePosition[key] - if !ok { + positions := r.parentPositionsFor(namespace, column, key) + if len(positions) == 0 { return nil, fmt.Errorf(`key "%v" is not found`, key) } if len(positions) > 1 { @@ -580,8 +610,8 @@ func (r *Collector) ParentRow(relation *Relation) func(value interface{}) (inter } keyParts = append(keyParts, io.NormalizeKey(key)) } - positions, ok := r.parentCompositePositions(relation)[buildCompositeKey(keyParts)] - if !ok { + positions := r.parentCompositePositionsFor(relation, buildCompositeKey(keyParts)) + if len(positions) == 0 { return nil, fmt.Errorf(`composite key "%v" is not found`, keyParts) } if len(positions) > 1 { @@ -603,10 +633,9 @@ func (r *Collector) ParentRow(relation *Relation) func(value interface{}) (inter } else { key = xType.Deref((*values)[r.manyCounter]) } - valuePosition := r.parentValuesPositions(relation.On[i].Namespace, relation.On[i].Column) key = io.NormalizeKey(key) - positions, ok := valuePosition[key] - if !ok { + positions := r.parentPositionsFor(relation.On[i].Namespace, relation.On[i].Column, key) + if len(positions) == 0 { return nil, fmt.Errorf(`key "%v" is not found`, key) } if len(positions) > 1 { @@ -645,8 +674,8 @@ func (r *Collector) visitorMany(relation *Relation) func(value interface{}) erro } keyParts = append(keyParts, io.NormalizeKey(key)) } - positions, ok := r.parentCompositePositions(relation)[buildCompositeKey(keyParts)] - if !ok { + positions := r.parentCompositePositionsFor(relation, buildCompositeKey(keyParts)) + if len(positions) == 0 { return nil } for _, index := range positions { @@ -675,10 +704,9 @@ func (r *Collector) visitorMany(relation *Relation) func(value interface{}) erro key = xType.Deref((*values)[r.manyCounter]) r.manyCounter++ } - valuePosition := r.parentValuesPositions(relation.On[i].Namespace, relation.On[i].Column) key = io.NormalizeKey(key) - positions, ok := valuePosition[key] - if !ok { + positions := r.parentPositionsFor(relation.On[i].Namespace, relation.On[i].Column, key) + if len(positions) == 0 { return nil } for _, index := range positions { @@ -1042,16 +1070,14 @@ func (r *Collector) mergeToParent() { holderField := r.relation.holderField parentSlice := r.parent.slice parentDestPtr := xunsafe.AsPointer(r.parent.DestPtr()) - valuePositions := r.parentCompositePositions(r.relation) - for i := 0; i < r.slice.Len(destPtr); i++ { value := r.slice.ValuePointerAt(destPtr, i) keyParts := make([]interface{}, 0, len(links)) for _, link := range links { keyParts = append(keyParts, io.NormalizeKey(link.xField.Value(xunsafe.AsPointer(value)))) } - positions, ok := valuePositions[buildCompositeKey(keyParts)] - if !ok { + positions := r.parentCompositePositionsFor(r.relation, buildCompositeKey(keyParts)) + if len(positions) == 0 { continue } for _, position := range positions { @@ -1072,18 +1098,19 @@ func (r *Collector) mergeToParent() { } for i, link := range links { - valuePositions := r.parentValuesPositions(r.relation.On[i].Namespace, r.relation.On[i].Column) destPtr := xunsafe.AsPointer(r.DestPtr()) holderField := r.relation.holderField parentSlice := r.parent.slice parentDestPtr := xunsafe.AsPointer(r.parent.DestPtr()) + namespace := r.relation.On[i].Namespace + column := r.relation.On[i].Column field := link.xField for i := 0; i < r.slice.Len(destPtr); i++ { value := r.slice.ValuePointerAt(destPtr, i) key := io.NormalizeKey(field.Value(xunsafe.AsPointer(value))) - positions, ok := valuePositions[key] - if !ok { + positions := r.parentPositionsFor(namespace, column, key) + if len(positions) == 0 { continue } @@ -1126,16 +1153,12 @@ func (r *Collector) ParentPlaceholders() ([]interface{}, [][]interface{}, []stri valueSets = append(valueSets, normalizeValues(field.Value(xunsafe.AsPointer(parent)))) continue } - positions := r.parentValuesPositions(link.Namespace, link.Column) - if len(positions) == 0 { + keys := r.parentPositionKeys(link.Namespace, link.Column) + if len(keys) == 0 { valueSets = nil break } - values := make([]interface{}, 0, len(positions)) - for key := range positions { - values = append(values, key) - } - valueSets = append(valueSets, values) + valueSets = append(valueSets, keys) } for _, row := range compositeRows(valueSets) { key := buildCompositeKey(row) @@ -1212,10 +1235,10 @@ outer: continue } - positions := r.parentValuesPositions(r.relation.On[k].Namespace, r.relation.On[k].Column) - result := make([]interface{}, len(positions)) + keys := r.parentPositionKeys(r.relation.On[k].Namespace, r.relation.On[k].Column) + result := make([]interface{}, len(keys)) counter := 0 - for key := range positions { + for _, key := range keys { result[counter] = key counter++ } diff --git a/view/collector_concurrent_test.go b/view/collector_concurrent_test.go index b31d2312a..df72ec8d2 100644 --- a/view/collector_concurrent_test.go +++ b/view/collector_concurrent_test.go @@ -12,17 +12,21 @@ import ( ) // TestCollectorConcurrentParentValuePositions reproduces concurrent map -// iteration and write on Collector.valuePosition. Child collectors write the -// parent map while another goroutine ranges it. +// iteration and write on Collector.valuePosition. Writers mutate the parent +// index while readers copy positions for a specific key. func TestCollectorConcurrentParentValuePositions(t *testing.T) { deadline := time.Now().Add(3 * time.Second) parentView := &View{Schema: state.NewSchema(reflect.TypeOf([]*compositeParentRow{}))} parentDest := []*compositeParentRow{} parent := NewCollector(parentView.Schema.Slice(), parentView, &parentDest, nil, false) + parent.valuePosition["ns"] = map[string]map[interface{}][]int{ + "col": {}, + } child := &Collector{parent: parent} + rel := &Relation{On: Links{{Namespace: "ns", Column: "col"}}} var next atomic.Uint64 var waitGroup sync.WaitGroup - waitGroup.Add(5) + waitGroup.Add(6) run := func(fn func()) { defer waitGroup.Done() @@ -31,11 +35,23 @@ func TestCollectorConcurrentParentValuePositions(t *testing.T) { } } - for worker := 0; worker < 5; worker++ { + for worker := 0; worker < 3; worker++ { + go run(func() { + key := int(next.Add(1)) + parent.indexValueToPosition(rel, key, key%8) + }) + } + for worker := 0; worker < 2; worker++ { go run(func() { - ns := fmt.Sprintf("ns%d", next.Add(1)) - _ = child.parentValuesPositions(ns, "col") + key := int(next.Load()) + positions := child.parentPositionsFor("ns", "col", key) + _ = len(positions) }) } + go run(func() { + ns := fmt.Sprintf("ns%d", next.Add(1)) + _ = child.parentPositionsFor(ns, "col", nil) + _ = child.parentPositionKeys(ns, "col") + }) waitGroup.Wait() }