src/internal/trace/v2/generation.go

   1 // Copyright 2023 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
   6
   7 import (
   8         "bufio"
   9         "bytes"
  10         "cmp"
  11         "encoding/binary"
  12         "fmt"
  13         "io"
  14         "slices"
  15         "strings"
  16
  17         "internal/trace/v2/event"
  18         "internal/trace/v2/event/go122"
  19 )
  20
  21 // generation contains all the trace data for a single
  22 // trace generation. It is purely data: it does not
  23 // track any parse state nor does it contain a cursor
  24 // into the generation.
  25 type generation struct {
  26         gen        uint64
  27         batches    map[ThreadID][]batch
  28         cpuSamples []cpuSample
  29         *evTable
  30 }
  31
  32 // spilledBatch represents a batch that was read out for the next generation,
  33 // while reading the previous one. It's passed on when parsing the next
  34 // generation.
  35 type spilledBatch struct {
  36         gen uint64
  37         *batch
  38 }
  39
  40 // readGeneration buffers and decodes the structural elements of a trace generation
  41 // out of r. spill is the first batch of the new generation (already buffered and
  42 // parsed from reading the last generation). Returns the generation and the first
  43 // batch read of the next generation, if any.
  44 func readGeneration(r *bufio.Reader, spill *spilledBatch) (*generation, *spilledBatch, error) {
  45         g := &generation{
  46                 evTable: new(evTable),
  47                 batches: make(map[ThreadID][]batch),
  48         }
  49         // Process the spilled batch.
  50         if spill != nil {
  51                 g.gen = spill.gen
  52                 if err := processBatch(g, *spill.batch); err != nil {
  53                         return nil, nil, err
  54                 }
  55                 spill = nil
  56         }
  57         // Read batches one at a time until we either hit EOF or
  58         // the next generation.
  59         for {
  60                 b, gen, err := readBatch(r)
  61                 if err == io.EOF {
  62                         break
  63                 }
  64                 if err != nil {
  65                         return nil, nil, err
  66                 }
  67                 if gen == 0 {
  68                         // 0 is a sentinel used by the runtime, so we'll never see it.
  69                         return nil, nil, fmt.Errorf("invalid generation number %d", gen)
  70                 }
  71                 if g.gen == 0 {
  72                         // Initialize gen.
  73                         g.gen = gen
  74                 }
  75                 if gen == g.gen+1 { // TODO: advance this the same way the runtime does.
  76                         spill = &spilledBatch{gen: gen, batch: &b}
  77                         break
  78                 }
  79                 if gen != g.gen {
  80                         // N.B. Fail as fast as possible if we see this. At first it
  81                         // may seem prudent to be fault-tolerant and assume we have a
  82                         // complete generation, parsing and returning that first. However,
  83                         // if the batches are mixed across generations then it's likely
  84                         // we won't be able to parse this generation correctly at all.
  85                         // Rather than return a cryptic error in that case, indicate the
  86                         // problem as soon as we see it.
  87                         return nil, nil, fmt.Errorf("generations out of order")
  88                 }
  89                 if err := processBatch(g, b); err != nil {
  90                         return nil, nil, err
  91                 }
  92         }
  93
  94         // Check some invariants.
  95         if g.freq == 0 {
  96                 return nil, nil, fmt.Errorf("no frequency event found")
  97         }
  98         for _, batches := range g.batches {
  99                 sorted := slices.IsSortedFunc(batches, func(a, b batch) int {
 100                         return cmp.Compare(a.time, b.time)
 101                 })
 102                 if !sorted {
 103                         // TODO(mknyszek): Consider just sorting here.
 104                         return nil, nil, fmt.Errorf("per-M streams are out-of-order")
 105                 }
 106         }
 107
 108         // Compactify stacks and strings for better lookup performance later.
 109         g.stacks.compactify()
 110         g.strings.compactify()
 111
 112         // Validate stacks.
 113         if err := validateStackStrings(&g.stacks, &g.strings); err != nil {
 114                 return nil, nil, err
 115         }
 116
 117         // Fix up the CPU sample timestamps, now that we have freq.
 118         for i := range g.cpuSamples {
 119                 s := &g.cpuSamples[i]
 120                 s.time = g.freq.mul(timestamp(s.time))
 121         }
 122         // Sort the CPU samples.
 123         slices.SortFunc(g.cpuSamples, func(a, b cpuSample) int {
 124                 return cmp.Compare(a.time, b.time)
 125         })
 126         return g, spill, nil
 127 }
 128
 129 // processBatch adds the batch to the generation.
 130 func processBatch(g *generation, b batch) error {
 131         switch {
 132         case b.isStringsBatch():
 133                 if err := addStrings(&g.strings, b); err != nil {
 134                         return err
 135                 }
 136         case b.isStacksBatch():
 137                 if err := addStacks(&g.stacks, b); err != nil {
 138                         return err
 139                 }
 140         case b.isCPUSamplesBatch():
 141                 samples, err := addCPUSamples(g.cpuSamples, b)
 142                 if err != nil {
 143                         return err
 144                 }
 145                 g.cpuSamples = samples
 146         case b.isFreqBatch():
 147                 freq, err := parseFreq(b)
 148                 if err != nil {
 149                         return err
 150                 }
 151                 if g.freq != 0 {
 152                         return fmt.Errorf("found multiple frequency events")
 153                 }
 154                 g.freq = freq
 155         default:
 156                 g.batches[b.m] = append(g.batches[b.m], b)
 157         }
 158         return nil
 159 }
 160
 161 // validateStackStrings makes sure all the string references in
 162 // the stack table are present in the string table.
 163 func validateStackStrings(stacks *dataTable[stackID, stack], strings *dataTable[stringID, string]) error {
 164         var err error
 165         stacks.forEach(func(id stackID, stk stack) bool {
 166                 for _, frame := range stk.frames {
 167                         _, ok := strings.get(frame.funcID)
 168                         if !ok {
 169                                 err = fmt.Errorf("found invalid func string ID %d for stack %d", frame.funcID, id)
 170                                 return false
 171                         }
 172                         _, ok = strings.get(frame.fileID)
 173                         if !ok {
 174                                 err = fmt.Errorf("found invalid file string ID %d for stack %d", frame.fileID, id)
 175                                 return false
 176                         }
 177                 }
 178                 return true
 179         })
 180         return err
 181 }
 182
 183 // addStrings takes a batch whose first byte is an EvStrings event
 184 // (indicating that the batch contains only strings) and adds each
 185 // string contained therein to the provided strings map.
 186 func addStrings(stringTable *dataTable[stringID, string], b batch) error {
 187         if !b.isStringsBatch() {
 188                 return fmt.Errorf("internal error: addStrings called on non-string batch")
 189         }
 190         r := bytes.NewReader(b.data)
 191         hdr, err := r.ReadByte() // Consume the EvStrings byte.
 192         if err != nil || event.Type(hdr) != go122.EvStrings {
 193                 return fmt.Errorf("missing strings batch header")
 194         }
 195
 196         var sb strings.Builder
 197         for r.Len() != 0 {
 198                 // Read the header.
 199                 ev, err := r.ReadByte()
 200                 if err != nil {
 201                         return err
 202                 }
 203                 if event.Type(ev) != go122.EvString {
 204                         return fmt.Errorf("expected string event, got %d", ev)
 205                 }
 206
 207                 // Read the string's ID.
 208                 id, err := binary.ReadUvarint(r)
 209                 if err != nil {
 210                         return err
 211                 }
 212
 213                 // Read the string's length.
 214                 len, err := binary.ReadUvarint(r)
 215                 if err != nil {
 216                         return err
 217                 }
 218                 if len > go122.MaxStringSize {
 219                         return fmt.Errorf("invalid string size %d, maximum is %d", len, go122.MaxStringSize)
 220                 }
 221
 222                 // Copy out the string.
 223                 n, err := io.CopyN(&sb, r, int64(len))
 224                 if n != int64(len) {
 225                         return fmt.Errorf("failed to read full string: read %d but wanted %d", n, len)
 226                 }
 227                 if err != nil {
 228                         return fmt.Errorf("copying string data: %w", err)
 229                 }
 230
 231                 // Add the string to the map.
 232                 s := sb.String()
 233                 sb.Reset()
 234                 if err := stringTable.insert(stringID(id), s); err != nil {
 235                         return err
 236                 }
 237         }
 238         return nil
 239 }
 240
 241 // addStacks takes a batch whose first byte is an EvStacks event
 242 // (indicating that the batch contains only stacks) and adds each
 243 // string contained therein to the provided stacks map.
 244 func addStacks(stackTable *dataTable[stackID, stack], b batch) error {
 245         if !b.isStacksBatch() {
 246                 return fmt.Errorf("internal error: addStacks called on non-stacks batch")
 247         }
 248         r := bytes.NewReader(b.data)
 249         hdr, err := r.ReadByte() // Consume the EvStacks byte.
 250         if err != nil || event.Type(hdr) != go122.EvStacks {
 251                 return fmt.Errorf("missing stacks batch header")
 252         }
 253
 254         for r.Len() != 0 {
 255                 // Read the header.
 256                 ev, err := r.ReadByte()
 257                 if err != nil {
 258                         return err
 259                 }
 260                 if event.Type(ev) != go122.EvStack {
 261                         return fmt.Errorf("expected stack event, got %d", ev)
 262                 }
 263
 264                 // Read the stack's ID.
 265                 id, err := binary.ReadUvarint(r)
 266                 if err != nil {
 267                         return err
 268                 }
 269
 270                 // Read how many frames are in each stack.
 271                 nFrames, err := binary.ReadUvarint(r)
 272                 if err != nil {
 273                         return err
 274                 }
 275                 if nFrames > go122.MaxFramesPerStack {
 276                         return fmt.Errorf("invalid stack size %d, maximum is %d", nFrames, go122.MaxFramesPerStack)
 277                 }
 278
 279                 // Each frame consists of 4 fields: pc, funcID (string), fileID (string), line.
 280                 frames := make([]frame, 0, nFrames)
 281                 for i := uint64(0); i < nFrames; i++ {
 282                         // Read the frame data.
 283                         pc, err := binary.ReadUvarint(r)
 284                         if err != nil {
 285                                 return fmt.Errorf("reading frame %d's PC for stack %d: %w", i+1, id, err)
 286                         }
 287                         funcID, err := binary.ReadUvarint(r)
 288                         if err != nil {
 289                                 return fmt.Errorf("reading frame %d's funcID for stack %d: %w", i+1, id, err)
 290                         }
 291                         fileID, err := binary.ReadUvarint(r)
 292                         if err != nil {
 293                                 return fmt.Errorf("reading frame %d's fileID for stack %d: %w", i+1, id, err)
 294                         }
 295                         line, err := binary.ReadUvarint(r)
 296                         if err != nil {
 297                                 return fmt.Errorf("reading frame %d's line for stack %d: %w", i+1, id, err)
 298                         }
 299                         frames = append(frames, frame{
 300                                 pc:     pc,
 301                                 funcID: stringID(funcID),
 302                                 fileID: stringID(fileID),
 303                                 line:   line,
 304                         })
 305                 }
 306
 307                 // Add the stack to the map.
 308                 if err := stackTable.insert(stackID(id), stack{frames: frames}); err != nil {
 309                         return err
 310                 }
 311         }
 312         return nil
 313 }
 314
 315 // addCPUSamples takes a batch whose first byte is an EvCPUSamples event
 316 // (indicating that the batch contains only CPU samples) and adds each
 317 // sample contained therein to the provided samples list.
 318 func addCPUSamples(samples []cpuSample, b batch) ([]cpuSample, error) {
 319         if !b.isCPUSamplesBatch() {
 320                 return nil, fmt.Errorf("internal error: addStrings called on non-string batch")
 321         }
 322         r := bytes.NewReader(b.data)
 323         hdr, err := r.ReadByte() // Consume the EvCPUSamples byte.
 324         if err != nil || event.Type(hdr) != go122.EvCPUSamples {
 325                 return nil, fmt.Errorf("missing CPU samples batch header")
 326         }
 327
 328         for r.Len() != 0 {
 329                 // Read the header.
 330                 ev, err := r.ReadByte()
 331                 if err != nil {
 332                         return nil, err
 333                 }
 334                 if event.Type(ev) != go122.EvCPUSample {
 335                         return nil, fmt.Errorf("expected CPU sample event, got %d", ev)
 336                 }
 337
 338                 // Read the sample's timestamp.
 339                 ts, err := binary.ReadUvarint(r)
 340                 if err != nil {
 341                         return nil, err
 342                 }
 343
 344                 // Read the sample's M.
 345                 m, err := binary.ReadUvarint(r)
 346                 if err != nil {
 347                         return nil, err
 348                 }
 349                 mid := ThreadID(m)
 350
 351                 // Read the sample's P.
 352                 p, err := binary.ReadUvarint(r)
 353                 if err != nil {
 354                         return nil, err
 355                 }
 356                 pid := ProcID(p)
 357
 358                 // Read the sample's G.
 359                 g, err := binary.ReadUvarint(r)
 360                 if err != nil {
 361                         return nil, err
 362                 }
 363                 goid := GoID(g)
 364                 if g == 0 {
 365                         goid = NoGoroutine
 366                 }
 367
 368                 // Read the sample's stack.
 369                 s, err := binary.ReadUvarint(r)
 370                 if err != nil {
 371                         return nil, err
 372                 }
 373
 374                 // Add the sample to the slice.
 375                 samples = append(samples, cpuSample{
 376                         schedCtx: schedCtx{
 377                                 M: mid,
 378                                 P: pid,
 379                                 G: goid,
 380                         },
 381                         time:  Time(ts), // N.B. this is really a "timestamp," not a Time.
 382                         stack: stackID(s),
 383                 })
 384         }
 385         return samples, nil
 386 }
 387
 388 // parseFreq parses out a lone EvFrequency from a batch.
 389 func parseFreq(b batch) (frequency, error) {
 390         if !b.isFreqBatch() {
 391                 return 0, fmt.Errorf("internal error: parseFreq called on non-frequency batch")
 392         }
 393         r := bytes.NewReader(b.data)
 394         r.ReadByte() // Consume the EvFrequency byte.
 395
 396         // Read the frequency. It'll come out as timestamp units per second.
 397         f, err := binary.ReadUvarint(r)
 398         if err != nil {
 399                 return 0, err
 400         }
 401         // Convert to nanoseconds per timestamp unit.
 402         return frequency(1.0 / (float64(f) / 1e9)), nil
 403 }