subscribe_test.go

     1  // Copyright 2025 The Go Authors. All rights reserved.
     2  // Use of this source code is governed by a BSD-style
     3  // license that can be found in the LICENSE file.
     4  
     5  package trace_test
     6  
     7  import (
     8  	"bytes"
     9  	inttrace "internal/trace"
    10  	"internal/trace/testtrace"
    11  	"io"
    12  	"runtime"
    13  	"runtime/trace"
    14  	"slices"
    15  	"testing"
    16  )
    17  
    18  func TestSubscribers(t *testing.T) {
    19  	validate := func(t *testing.T, source string, tr io.Reader) {
    20  		// Prepare to read the trace snapshot.
    21  		r, err := inttrace.NewReader(tr)
    22  		if err != nil {
    23  			t.Fatalf("unexpected error creating trace reader for %s: %v", source, err)
    24  			return
    25  		}
    26  
    27  		v := testtrace.NewValidator()
    28  		// These platforms can't guarantee a monotonically increasing clock reading in a short trace.
    29  		if runtime.GOOS == "windows" || runtime.GOARCH == "wasm" {
    30  			v.SkipClockSnapshotChecks()
    31  		}
    32  		// Make sure there are Sync events: at the start and end.
    33  		var syncs []int
    34  		evs := 0
    35  		for {
    36  			ev, err := r.ReadEvent()
    37  			if err == io.EOF {
    38  				break
    39  			}
    40  			if err != nil {
    41  				t.Fatalf("unexpected error reading trace for %s: %v", source, err)
    42  			}
    43  			if err := v.Event(ev); err != nil {
    44  				t.Fatalf("event validation failed: %s", err)
    45  			}
    46  			if ev.Kind() == inttrace.EventSync {
    47  				syncs = append(syncs, evs)
    48  			}
    49  			evs++
    50  		}
    51  		ends := []int{syncs[0], syncs[len(syncs)-1]}
    52  		if wantEnds := []int{0, evs - 1}; !slices.Equal(wantEnds, ends) {
    53  			t.Errorf("expected a sync event at each end of the trace, found sync events at %d instead of %d for %s",
    54  				ends, wantEnds, source)
    55  		}
    56  	}
    57  
    58  	validateTraces := func(t *testing.T, tReader, frReader io.Reader) {
    59  		validate(t, "tracer", tReader)
    60  		validate(t, "flightRecorder", frReader)
    61  	}
    62  	startFlightRecorder := func(t *testing.T) *trace.FlightRecorder {
    63  		fr := trace.NewFlightRecorder(trace.FlightRecorderConfig{})
    64  		if err := fr.Start(); err != nil {
    65  			t.Fatalf("unexpected error creating flight recorder: %v", err)
    66  		}
    67  		return fr
    68  	}
    69  	startTrace := func(t *testing.T, w io.Writer) {
    70  		if err := trace.Start(w); err != nil {
    71  			t.Fatalf("unexpected error starting flight recorder: %v", err)
    72  		}
    73  	}
    74  	stopFlightRecorder := func(t *testing.T, fr *trace.FlightRecorder, w io.Writer) {
    75  		if _, err := fr.WriteTo(w); err != nil {
    76  			t.Fatalf("unexpected error writing trace from flight recorder: %v", err)
    77  		}
    78  		fr.Stop()
    79  	}
    80  	stopTrace := func() {
    81  		trace.Stop()
    82  	}
    83  	t.Run("start(flight)_start(trace)_stop(trace)_stop(flight)", func(t *testing.T) {
    84  		if trace.IsEnabled() {
    85  			t.Skip("skipping because trace is already enabled")
    86  		}
    87  		frBuf := new(bytes.Buffer)
    88  		tBuf := new(bytes.Buffer)
    89  		fr := startFlightRecorder(t)
    90  		defer fr.Stop()
    91  		startTrace(t, tBuf)
    92  		defer trace.Stop()
    93  		stopTrace()
    94  		stopFlightRecorder(t, fr, frBuf)
    95  		validateTraces(t, tBuf, frBuf)
    96  	})
    97  	t.Run("start(trace)_start(flight)_stop(trace)_stop(flight)", func(t *testing.T) {
    98  		if trace.IsEnabled() {
    99  			t.Skip("skipping because trace is already enabled")
   100  		}
   101  		frBuf := new(bytes.Buffer)
   102  		tBuf := new(bytes.Buffer)
   103  		startTrace(t, tBuf)
   104  		defer trace.Stop()
   105  		fr := startFlightRecorder(t)
   106  		defer fr.Stop()
   107  		stopTrace()
   108  		stopFlightRecorder(t, fr, frBuf)
   109  		validateTraces(t, tBuf, frBuf)
   110  	})
   111  	t.Run("start(flight)_stop(flight)_start(trace)_stop(trace)", func(t *testing.T) {
   112  		if trace.IsEnabled() {
   113  			t.Skip("skipping because trace is already enabled")
   114  		}
   115  		frBuf := new(bytes.Buffer)
   116  		tBuf := new(bytes.Buffer)
   117  		fr := startFlightRecorder(t)
   118  		defer fr.Stop()
   119  		stopFlightRecorder(t, fr, frBuf)
   120  		startTrace(t, tBuf)
   121  		defer trace.Stop()
   122  		stopTrace()
   123  		validateTraces(t, tBuf, frBuf)
   124  	})
   125  	t.Run("start(flight)_stop(flight)_start(trace)_stop(trace)", func(t *testing.T) {
   126  		if trace.IsEnabled() {
   127  			t.Skip("skipping because trace is already enabled")
   128  		}
   129  		frBuf := new(bytes.Buffer)
   130  		tBuf := new(bytes.Buffer)
   131  		fr := startFlightRecorder(t)
   132  		defer fr.Stop()
   133  		stopFlightRecorder(t, fr, frBuf)
   134  		startTrace(t, tBuf)
   135  		defer trace.Stop()
   136  		stopTrace()
   137  		validateTraces(t, tBuf, frBuf)
   138  	})
   139  	t.Run("start(flight)_start(trace)_stop(flight)_stop(trace)", func(t *testing.T) {
   140  		if trace.IsEnabled() {
   141  			t.Skip("skipping because trace is already enabled")
   142  		}
   143  		frBuf := new(bytes.Buffer)
   144  		tBuf := new(bytes.Buffer)
   145  		fr := startFlightRecorder(t)
   146  		defer fr.Stop()
   147  		startTrace(t, tBuf)
   148  		defer trace.Stop()
   149  		stopFlightRecorder(t, fr, frBuf)
   150  		stopTrace()
   151  		validateTraces(t, tBuf, frBuf)
   152  	})
   153  }
   154
View as plain text