diff --git a/internal/pdh/collector.go b/internal/pdh/collector.go index ae02bc70..3869ec21 100644 --- a/internal/pdh/collector.go +++ b/internal/pdh/collector.go @@ -427,6 +427,8 @@ type collectState struct { // valid records for every row and counter whether the counter had a valid value. valid []bool + + names instanceNameCache } // collect replaces the content of dst with one sample. @@ -443,15 +445,11 @@ func (c *Collector[T]) collect(dst *[]T, state *collectState) (err error) { *dst = (*dst)[:0:0] - rows := rowSet[T]{ - c: c, - dst: dst, - index: map[instanceKey]int{}, - nameCache: map[string]string{}, - occurrences: map[string]int{}, - seen: map[string]int{}, - valid: state.valid[:0], - } + rows := newRowSet(c, dst, state) + + // Also after an error or panic, so the cache stays bounded while samples keep failing. + // A name that the sample did not reach is decoded again by the next sample. + defer state.names.removeUnseen() for counterIndex := range c.counters { for _, instance := range c.counters[counterIndex].Instances { @@ -571,7 +569,7 @@ type rowSet[T any] struct { c *Collector[T] dst *[]T index map[instanceKey]int - nameCache map[string]string + nameCache *instanceNameCache // occurrences includes invalid items to reserve all literal names and keep occurrence positions. occurrences map[string]int // seen counts names in the current counter array and is cleared between arrays. @@ -580,11 +578,25 @@ type rowSet[T any] struct { valid []bool } +func newRowSet[T any](c *Collector[T], dst *[]T, state *collectState) rowSet[T] { + state.names.startSample() + + return rowSet[T]{ + c: c, + dst: dst, + index: map[instanceKey]int{}, + nameCache: &state.names, + occurrences: map[string]int{}, + seen: map[string]int{}, + valid: state.valid[:0], + } +} + // row returns the index of the row for the instance named by szName, // appending the row if needed. ok is false if the item has no valid data or // its instance is not collected. func (r *rowSet[T]) row(counter *Counter, szName *uint16, status uint32) (int, bool) { - instanceName := decodeInstanceName(r.nameCache, szName) + instanceName := r.nameCache.decode(szName) if instanceName == "" || instanceName == "*" { instanceName = InstanceEmpty } @@ -781,11 +793,32 @@ func formatCounterPath(object, instance, counterName string) string { return counterPath } -// decodeInstanceName returns the instance name p points to. -// Decoded names are cached by their raw UTF-16 content and not by p, because p points -// into the item buffer, which is overwritten by every Get*CounterArray call. With explicit -// instances, every call writes its single name at the same address. -func decodeInstanceName(cache map[string]string, p *uint16) string { +// instanceNameCache maps the raw UTF-16 content of instance names to their decoded names. +// Names are cached by content and not by pointer, because the pointers point into the item +// buffer, which is overwritten by every Get*CounterArray call. With explicit instances, every +// call writes its single name at the same address. +// +// The cache is kept between samples, so a name is decoded once while its instance exists. +// Names that a sample did not decode are removed at its end, because instance names that +// contain process IDs, like those of GPU Engine, change all the time. +type instanceNameCache struct { + names map[string]*cachedInstanceName + sample uint64 +} + +type cachedInstanceName struct { + name string + // sample is the last sample that contained the name. + sample uint64 +} + +// startSample starts a new sample. +func (cache *instanceNameCache) startSample() { + cache.sample++ +} + +// decode returns the instance name p points to. +func (cache *instanceNameCache) decode(p *uint16) string { if p == nil { return "" } @@ -798,16 +831,31 @@ func decodeInstanceName(cache map[string]string, p *uint16) string { raw := unsafe.Slice((*byte)(unsafe.Pointer(p)), n*2) // The compiler does not allocate for the string conversion in a map lookup. - if name, ok := cache[string(raw)]; ok { - return name + if cached, ok := cache.names[string(raw)]; ok { + cached.sample = cache.sample + + return cached.name + } + + if cache.names == nil { + cache.names = map[string]*cachedInstanceName{} } name := windows.UTF16ToString(unsafe.Slice(p, n)) - cache[string(raw)] = name + cache.names[string(raw)] = &cachedInstanceName{name: name, sample: cache.sample} return name } +// removeUnseen removes the names that the current sample did not decode. +func (cache *instanceNameCache) removeUnseen() { + for raw, cached := range cache.names { + if cached.sample != cache.sample { + delete(cache.names, raw) + } + } +} + func isKnownCounterDataError(err error) bool { var pdhErr *Error diff --git a/internal/pdh/collector_internal_test.go b/internal/pdh/collector_internal_test.go index 742f526a..7a64d51d 100644 --- a/internal/pdh/collector_internal_test.go +++ b/internal/pdh/collector_internal_test.go @@ -73,14 +73,7 @@ func fillRows(t *testing.T, c *Collector[rowSetValues]) []rowSetValues { var dst []rowSetValues - rows := rowSet[rowSetValues]{ - c: c, - dst: &dst, - index: map[instanceKey]int{}, - nameCache: map[string]string{}, - occurrences: map[string]int{}, - seen: map[string]int{}, - } + rows := newRowSet(c, &dst, &collectState{}) for _, name := range []string{"complete", "partial"} { row, ok := rows.row(&c.counters[0], windows.StringToUTF16Ptr(name), CstatusValidData) @@ -233,14 +226,7 @@ func TestRowSetDuplicateInstances(t *testing.T) { var dst []rowSetValues - rows := rowSet[rowSetValues]{ - c: c, - dst: &dst, - index: map[instanceKey]int{}, - nameCache: map[string]string{}, - occurrences: map[string]int{}, - seen: map[string]int{}, - } + rows := newRowSet(c, &dst, &collectState{}) addDuplicateTestItems( &rows, @@ -420,3 +406,79 @@ func TestCollectRecoversPanic(t *testing.T) { require.NoError(t, c.Collect(&dst)) require.NotEmpty(t, dst) } + +// TestCollectKeepsInstanceNames checks that Collect reuses decoded names between samples and +// removes the cached names that a failed sample did not reach. +func TestCollectKeepsInstanceNames(t *testing.T) { + t.Parallel() + + c, err := NewCollector[processThreads](slog.New(slog.DiscardHandler), CounterTypeRaw, "Process", InstancesAll) + require.NoError(t, err) + + t.Cleanup(c.Close) + + systemName := func(rows []processThreads) string { + i := slices.IndexFunc(rows, func(row processThreads) bool { return row.Name == "System" }) + require.NotEqual(t, -1, i, "missing instance System") + + return rows[i].Name + } + + var first, second []processThreads + + require.NoError(t, c.Collect(&first)) + require.NoError(t, c.Collect(&second)) + require.Same(t, unsafe.StringData(systemName(first)), unsafe.StringData(systemName(second)), "name decoded again") + + // Assertions run without holding the lock, because a failed assertion would leave it locked for Close. + cachedNames := func(fieldIndex int) int { + c.mu.Lock() + defer c.mu.Unlock() + + c.counters[0].FieldIndexValue = fieldIndex + + return len(c.state.names.names) + } + + // Point the counter at a field that does not exist, so that the first item panics. + require.Greater(t, cachedNames(42), 1) + require.Error(t, c.Collect(&first)) + require.Equal(t, 1, cachedNames(1), "names not reached by the failed sample kept") + + require.NoError(t, c.Collect(&first)) + require.Equal(t, systemName(second), systemName(first)) +} + +func TestInstanceNameCache(t *testing.T) { + t.Parallel() + + var cache instanceNameCache + + // PDH writes the names of every counter array into the same buffer. + buf := make([]uint16, 8) + write := func(name string) *uint16 { + copy(buf, windows.StringToUTF16(name)) + + return &buf[0] + } + + cache.startSample() + + first := cache.decode(write("a")) + require.Equal(t, "a", first) + require.Equal(t, "bb", cache.decode(write("bb"))) + require.Empty(t, cache.decode(nil)) + + cache.removeUnseen() + require.Len(t, cache.names, 2) + + cache.startSample() + + second := cache.decode(write("a")) + require.Equal(t, "a", second) + require.Same(t, unsafe.StringData(first), unsafe.StringData(second), "name decoded again") + + cache.removeUnseen() + require.Len(t, cache.names, 1, "name of a vanished instance kept") + require.Contains(t, cache.names, "a\x00", "keyed by the UTF-16LE content") +} diff --git a/internal/pdh/collector_rowset_bench_test.go b/internal/pdh/collector_rowset_bench_test.go new file mode 100644 index 00000000..c19d82ad --- /dev/null +++ b/internal/pdh/collector_rowset_bench_test.go @@ -0,0 +1,64 @@ +// SPDX-License-Identifier: Apache-2.0 +// +// Copyright The Prometheus Authors +// Licensed under the Apache License, Version 2.0 (the "License"); +// you may not use this file except in compliance with the License. +// You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, +// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +// See the License for the specific language governing permissions and +// limitations under the License. + +//go:build windows + +package pdh + +import ( + "fmt" + "runtime" + "testing" + "unsafe" + + "golang.org/x/sys/windows" +) + +// BenchmarkRowSetInstanceNames measures repeated samples of an object with +// many instances, like GPU Engine, without querying PDH. +func BenchmarkRowSetInstanceNames(b *testing.B) { + const instances = 500 + + c := newRowSetTestCollector(true) + + items := make([]RawCounterItem, instances) + for i := range items { + items[i] = RawCounterItem{ + SzName: windows.StringToUTF16Ptr(fmt.Sprintf("pid_%d_luid_0x00000000_0x0000D1A5_phys_0_eng_%d_engtype_3D", 1000+i, i%8)), + RawValue: RawCounter{CStatus: CstatusValidData, FirstValue: int64(i)}, + } + } + + buf := unsafe.Slice((*byte)(unsafe.Pointer(unsafe.SliceData(items))), len(items)*int(unsafe.Sizeof(RawCounterItem{}))) + state := &collectState{} + + var dst []rowSetValues + + b.ReportAllocs() + + for b.Loop() { + dst = dst[:0:0] + rows := newRowSet(c, &dst, state) + + rows.addRawItems(0, buf, instances) + rows.addRawItems(1, buf, instances) + rows.finish() + state.names.removeUnseen() + + state.valid = rows.valid + } + + runtime.KeepAlive(items) +}