Merge pull request #2 from pyroscope-io/improved-sampling

Refine sampler performance

Author: kolesnikovae

examples/tracing/collect.go

@@ -2,13 +2,11 @@ package main
 
 import (
 	"flag"
-	"fmt"
 	"io"
 	"log"
 	"os"
 	"os/signal"
 	"strconv"
-	"strings"
 
 	"github.com/pyroscope-io/dotnetdiag"
 	"github.com/pyroscope-io/dotnetdiag/nettrace"
@@ -74,23 +72,10 @@ func main() {
 		case nil:
 			continue
 		case io.EOF:
-			p.Walk(treePrinter(os.Stdout))
-			log.Println("Done")
+			for name, time := range p.Samples() {
+				log.Println(name, time)
+			}
 			return
 		}
 	}
 }
-
-func treePrinter(w io.Writer) func(profiler.FrameInfo) {
-	return func(frame profiler.FrameInfo) {
-		_, _ = fmt.Fprintf(w, "%s(%v) %s\n", padding(frame.Level), frame.SampledTime, frame.Name)
-	}
-}
-
-func padding(x int) string {
-	var s strings.Builder
-	for i := 0; i < x; i++ {
-		s.WriteString("\t")
-	}
-	return s.String()
-}

nettrace/block.go

@@ -210,18 +210,24 @@ func SequencePointBlockFromObject(o Object) (*SequencePointBlock, error) {
 	return &b, p.Err()
 }
 
-func (s Stack) InstructionPointers64() []uint64 {
-	n := make([]uint64, len(s.Data)/8)
+func (s Stack) InstructionPointers(size int32) []uint64 {
+	var address func([]byte, int) uint64
+	if size == 8 {
+		address = address64
+	} else {
+		address = address32
+	}
+	n := make([]uint64, len(s.Data)/int(size))
 	for i := 0; i < len(n); i++ {
-		n[i] = binary.LittleEndian.Uint64(s.Data[i*8 : (i+1)*8])
+		n[len(n)-1-i] = address(s.Data, i)
 	}
 	return n
 }
 
-func (s Stack) InstructionPointers32() []uint64 {
-	n := make([]uint64, len(s.Data)/4)
-	for i := 0; i < len(n); i++ {
-		n[i] = uint64(binary.LittleEndian.Uint32(s.Data[i*4 : (i+1)*4]))
-	}
-	return n
+func address64(b []byte, i int) uint64 {
+	return binary.LittleEndian.Uint64(b[i*8 : (i+1)*8])
+}
+
+func address32(b []byte, i int) uint64 {
+	return uint64(binary.LittleEndian.Uint32(b[i*4 : (i+1)*4]))
 }

nettrace/nettrace_test.go

@@ -7,21 +7,36 @@ import (
 	"os"
 	"runtime/debug"
 	"sort"
-	"strings"
 	"testing"
 	"time"
 
 	"github.com/pyroscope-io/dotnetdiag/nettrace"
 	"github.com/pyroscope-io/dotnetdiag/nettrace/profiler"
 )
 
-func TestStream(t *testing.T) {
-	sample, err := os.Open("testdata/golden.nettrace")
+func TestNetTraceDecoding(t *testing.T) {
+	t.Run(".Net 5.0 SampleProfiler Web app", func(t *testing.T) {
+		requireEqual(t,
+			"testdata/dotnet-5.0-SampleProfiler-webapp.golden.nettrace",
+			"testdata/dotnet-5.0-SampleProfiler-webapp.txt")
+	})
+
+	t.Run(".Net 5.0 SampleProfiler Simple single thread app", func(t *testing.T) {
+		requireEqual(t,
+			"testdata/dotnet-5.0-SampleProfiler-single-thread.golden.nettrace",
+			"testdata/dotnet-5.0-SampleProfiler-single-thread.txt")
+	})
+}
+
+func requireEqual(t *testing.T, sample, expected string) {
+	t.Helper()
+
+	s, err := os.Open(sample)
 	requireNoError(t, err)
-	expected, err := os.ReadFile("testdata/expected.nettrace")
+	e, err := os.ReadFile(expected)
 	requireNoError(t, err)
 
-	stream := nettrace.NewStream(sample)
+	stream := nettrace.NewStream(s)
 	trace, err := stream.Open()
 	requireNoError(t, err)
 
@@ -39,12 +54,10 @@ func TestStream(t *testing.T) {
 		case nil:
 			continue
 		case io.EOF:
-			r := newRenderer()
-			p.Walk(r.visitor)
 			var b bytes.Buffer
-			r.dumpFlat(&b)
-			if b.String() != string(expected) {
-				t.Fatalf("Unexpected output")
+			dump(&b, p.Samples())
+			if b.String() != string(e) {
+				t.Fatal("output mismatch")
 			}
 			return
 		}
@@ -58,60 +71,13 @@ func requireNoError(t *testing.T, err error) {
 	}
 }
 
-type renderer struct {
-	out   map[string]time.Duration
-	names []string
-	val   time.Duration
-	prev  int
-}
-
-func newRenderer() *renderer {
-	return &renderer{out: make(map[string]time.Duration)}
-}
-
-func (r *renderer) visitor(frame profiler.FrameInfo) {
-	if frame.Level > r.prev || (frame.Level == 0 && r.prev == 0) {
-		r.names = append(r.names, frame.Name)
-	} else {
-		r.complete()
-		if frame.Level == 0 {
-			r.names = []string{frame.Name}
-		} else {
-			r.names = append(r.names[:frame.Level], frame.Name)
-		}
-	}
-	r.val = frame.SampledTime
-	r.prev = frame.Level
-}
-
-func (r *renderer) complete() {
-	if len(r.names) > 0 {
-		r.out[strings.Join(r.names, ";")] += r.val
-	}
-}
-
-func (r *renderer) dumpFlat(w io.Writer) {
-	r.complete()
-	s := make([]string, 0, len(r.out))
-	for k, v := range r.out {
-		s = append(s, fmt.Sprint(k, " ", v.Nanoseconds()))
-	}
-	sort.Strings(s)
-	for _, x := range s {
-		_, _ = fmt.Fprintln(w, x)
+func dump(w io.Writer, samples map[string]time.Duration) {
+	names := make([]string, 0, len(samples))
+	for k := range samples {
+		names = append(names, k)
 	}
-}
-
-func (r *renderer) dumpTree(w io.Writer) func(profiler.FrameInfo) {
-	return func(frame profiler.FrameInfo) {
-		_, _ = fmt.Fprintf(w, "%s(%v) %s\n", padding(frame.Level), frame.SampledTime, frame.Name)
-	}
-}
-
-func padding(x int) string {
-	var s strings.Builder
-	for i := 0; i < x; i++ {
-		s.WriteString("\t")
+	sort.Strings(names)
+	for _, n := range names {
+		_, _ = fmt.Fprintln(w, n, samples[n].Nanoseconds())
 	}
-	return s.String()
 }

nettrace/profiler/profiler.go

@@ -4,6 +4,7 @@ import (
 	"container/heap"
 	"encoding/binary"
 	"fmt"
+	"strings"
 	"time"
 
 	"github.com/pyroscope-io/dotnetdiag/nettrace"
@@ -17,40 +18,43 @@ type SampleProfiler struct {
 	stacks  map[int32][]uint64
 	threads map[int64]*thread
 
-	samples samples
+	events  events
+	samples []sample
+}
+
+type sample struct {
+	stack []uint64
+	val   int64
 }
 
 type FrameInfo struct {
-	ThreadID    int64
-	Level       int
 	SampledTime time.Duration
-	Addr        uint64
 	Name        string
 }
 
-type sample struct {
+type event struct {
 	typ          clrThreadSampleType
 	threadID     int64
 	stackID      int32
 	timestamp    int64
 	relativeTime int64
 }
 
-type samples []*sample
+type events []*event
 
-func (s samples) Len() int { return len(s) }
+func (e events) Len() int { return len(e) }
 
-func (s samples) Less(i, j int) bool { return s[i].timestamp < s[j].timestamp }
+func (e events) Less(i, j int) bool { return e[i].timestamp < e[j].timestamp }
 
-func (s samples) Swap(i, j int) { s[i], s[j] = s[j], s[i] }
+func (e events) Swap(i, j int) { e[i], e[j] = e[j], e[i] }
 
-func (s *samples) Push(x interface{}) { *s = append(*s, x.(*sample)) }
+func (e *events) Push(x interface{}) { *e = append(*e, x.(*event)) }
 
-func (s *samples) Pop() interface{} {
-	old := *s
+func (e *events) Pop() interface{} {
+	old := *e
 	n := len(old)
 	x := old[n-1]
-	*s = old[0 : n-1]
+	*e = old[0 : n-1]
 	return x
 }
 
@@ -80,28 +84,16 @@ func NewSampleProfiler(trace *nettrace.Trace) *SampleProfiler {
 	}
 }
 
-func (s *SampleProfiler) Walk(fn func(FrameInfo)) {
-	for tid := range s.threads {
-		s.WalkThread(tid, fn)
-	}
-}
-
-func (s *SampleProfiler) WalkThread(threadID int64, fn func(FrameInfo)) {
-	t, ok := s.threads[threadID]
-	if !ok {
-		return
+func (s *SampleProfiler) Samples() map[string]time.Duration {
+	samples := make(map[string]time.Duration)
+	for _, x := range s.samples {
+		name := make([]string, len(x.stack))
+		for i := range x.stack {
+			name[i] = s.sym.resolve(x.stack[i])
+		}
+		samples[strings.Join(name, ";")] += time.Duration(x.val * -1)
 	}
-	t.walk(func(i int, n *frame) {
-		addr, _ := s.sym.resolveAddress(n.addr)
-		name, _ := s.sym.resolveString(addr)
-		fn(FrameInfo{
-			ThreadID:    threadID,
-			Addr:        n.addr,
-			Name:        name,
-			SampledTime: time.Duration(n.sampledTime),
-			Level:       i,
-		})
-	})
+	return samples
 }
 
 func (s *SampleProfiler) EventHandler(e *nettrace.Blob) error {
@@ -134,35 +126,35 @@ func (s *SampleProfiler) MetadataHandler(md *nettrace.Metadata) error {
 
 func (s *SampleProfiler) StackBlockHandler(sb *nettrace.StackBlock) error {
 	for _, stack := range sb.Stacks {
-		s.addStack(stack)
+		s.stacks[stack.ID] = stack.InstructionPointers(s.trace.PointerSize)
 	}
 	return nil
 }
 
 func (s *SampleProfiler) SequencePointBlockHandler(*nettrace.SequencePointBlock) error {
-	for s.samples.Len() != 0 {
-		x := heap.Pop(&s.samples).(*sample)
-		s.thread(x.threadID).addSample(x.typ, x.relativeTime, s.stacks[x.stackID])
+	for s.events.Len() != 0 {
+		x := heap.Pop(&s.events).(*event)
+		s.thread(x.threadID).addSample(x.typ, x.relativeTime, x.stackID)
+	}
+	for _, t := range s.threads {
+		for k, v := range t.samples {
+			s.samples = append(s.samples, sample{
+				stack: s.stacks[k],
+				val:   v,
+			})
+		}
+		t.samples = make(map[int32]int64)
 	}
 	s.stacks = make(map[int32][]uint64)
 	return nil
 }
 
-func (s *SampleProfiler) addStack(x nettrace.Stack) {
-	if s.trace.PointerSize == 8 {
-		s.stacks[x.ID] = x.InstructionPointers64()
-		return
-	}
-	s.stacks[x.ID] = x.InstructionPointers32()
-	return
-}
-
 func (s *SampleProfiler) addSample(e *nettrace.Blob) error {
 	var d clrThreadSampleTraceData
 	if err := binary.Read(e.Payload, binary.LittleEndian, &d); err != nil {
 		return err
 	}
-	heap.Push(&s.samples, &sample{
+	heap.Push(&s.events, &event{
 		typ:          d.Type,
 		threadID:     e.Header.ThreadID,
 		stackID:      e.Header.StackID,
@@ -177,7 +169,7 @@ func (s *SampleProfiler) thread(tid int64) *thread {
 	if ok {
 		return t
 	}
-	t = new(thread)
+	t = &thread{samples: make(map[int32]int64)}
 	s.threads[tid] = t
 	return t
 }