Skip to content

Use lowercase g prefix for goroutine IDs #6

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
tbg merged 1 commit into from
Mar 9, 2026
Merged

Use lowercase g prefix for goroutine IDs #6

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
39 changes: 25 additions & 14 deletions main.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -4,15 +4,14 @@ package main
import (
"database/sql"
"fmt"
"io"
"os"
"os/exec"
"path/filepath"
"strings"
"syscall"
"time"

"io"

"github.com/dt/schedstat/internal/tracedb"
_ "github.com/marcboeker/go-duckdb"
"github.com/spf13/cobra"
Expand Down Expand Up @@ -360,7 +359,9 @@ type spikeInfo struct {
spikeType spikeType
}

func queryLatencySpikes(db *sql.DB, minTime int64, window, threshold time.Duration, top int) ([]spikeInfo, int, error) {
func queryLatencySpikes(
db *sql.DB, minTime int64, window, threshold time.Duration, top int,
) ([]spikeInfo, int, error) {
windowNs := window.Nanoseconds()
thresholdNs := float64(threshold.Nanoseconds())

Expand Down Expand Up @@ -400,7 +401,9 @@ func queryLatencySpikes(db *sql.DB, minTime int64, window, threshold time.Durati
return spikes, total, rows.Err()
}

func queryRunnableSpikes(db *sql.DB, minTime int64, window time.Duration, runnableThreshold, top int) ([]spikeInfo, int, error) {
func queryRunnableSpikes(
db *sql.DB, minTime int64, window time.Duration, runnableThreshold, top int,
) ([]spikeInfo, int, error) {
windowMs := window.Milliseconds()

rows, err := db.Query(`
Expand Down Expand Up @@ -453,7 +456,9 @@ func queryRunnableSpikes(db *sql.DB, minTime int64, window time.Duration, runnab
return spikes, total, rows.Err()
}

func printLatencySpikesSection(w io.Writer, spikes []spikeInfo, total int, window, threshold time.Duration) {
func printLatencySpikesSection(
w io.Writer, spikes []spikeInfo, total int, window, threshold time.Duration,
) {
if total == 0 {
return
}
Expand All @@ -472,7 +477,9 @@ func printLatencySpikesSection(w io.Writer, spikes []spikeInfo, total int, windo
}
}

func printRunnableSpikesSection(w io.Writer, spikes []spikeInfo, total int, window time.Duration, runnableThreshold int) {
func printRunnableSpikesSection(
w io.Writer, spikes []spikeInfo, total int, window time.Duration, runnableThreshold int,
) {
if total == 0 {
return
}
Expand Down Expand Up @@ -720,14 +727,16 @@ func printWorstDelays(db *sql.DB, w io.Writer, n int) error {
return err
}
stack := parseStackArray(stackStr)
fmt.Fprintf(w, "\n%d. G%d waited %s\n", rank, g, fmtDuration(durationNs))
fmt.Fprintf(w, "\n%d. g%d waited %s\n", rank, g, fmtDuration(durationNs))
fmt.Fprintf(w, " %s\n", formatStack(stack, 100))
rank++
}
return rows.Err()
}

func printSpikeDetails(db *sql.DB, w io.Writer, spikes []spikeInfo, minTime int64, window time.Duration) error {
func printSpikeDetails(
db *sql.DB, w io.Writer, spikes []spikeInfo, minTime int64, window time.Duration,
) error {
fmt.Fprintf(w, "\n--- Spike Details ---\n")

windowNs := window.Nanoseconds()
Expand Down Expand Up @@ -779,7 +788,7 @@ func printLatencySpikeDetail(db *sql.DB, w io.Writer, s spikeInfo, minTime, wind
return fmt.Errorf("worst delay query: %w", err)
}

fmt.Fprintf(w, " → G%d waited %s", g, fmtDuration(durationNs))
fmt.Fprintf(w, " → g%d waited %s", g, fmtDuration(durationNs))
if srcP.Valid {
fmt.Fprintf(w, " on P%d", srcP.Int64)
}
Expand Down Expand Up @@ -843,7 +852,7 @@ func printLatencySpikeDetail(db *sql.DB, w io.Writer, s spikeInfo, minTime, wind

if longestRunNs.Valid && longestRunNs.Float64 > 500000 { // > 500µs
stack := parseStackArray(longestRunStack.String)
fmt.Fprintf(w, " → Longest run during wait: G%d ran %s\n",
fmt.Fprintf(w, " → Longest run during wait: g%d ran %s\n",
longestRunG.Int64, fmtDuration(longestRunNs.Float64))
if len(stack) > 0 {
fmt.Fprintf(w, " %s\n", formatStack(stack, 80))
Expand All @@ -862,7 +871,9 @@ func printLatencySpikeDetail(db *sql.DB, w io.Writer, s spikeInfo, minTime, wind
return nil
}

func printRunnableSpikeDetail(db *sql.DB, w io.Writer, s spikeInfo, minTime int64, window time.Duration) error {
func printRunnableSpikeDetail(
db *sql.DB, w io.Writer, s spikeInfo, minTime int64, window time.Duration,
) error {
windowMs := window.Milliseconds()
windowStartMs := int64(s.windowNum) * windowMs
windowEndMs := windowStartMs + windowMs
Expand Down Expand Up @@ -1185,9 +1196,9 @@ func printHeavyUnblockers(db *sql.DB, w io.Writer, runnableStartNs int64) error
}
if stackStr.Valid {
stack := parseStackArray(stackStr.String)
fmt.Fprintf(w, " Heavy unblocker G%d (%d): %s\n", srcG, cnt, formatStack(stack, 100))
fmt.Fprintf(w, " Heavy unblocker g%d (%d): %s\n", srcG, cnt, formatStack(stack, 100))
} else {
fmt.Fprintf(w, " Heavy unblocker G%d (%d): (unknown stack)\n", srcG, cnt)
fmt.Fprintf(w, " Heavy unblocker g%d (%d): (unknown stack)\n", srcG, cnt)
}
}
return rows.Err()
Expand Down Expand Up @@ -1261,7 +1272,7 @@ func printTopGoroutines(db *sql.DB, w io.Writer) error {
return err
}
stack := parseStackArray(stackStr)
fmt.Fprintf(w, " G%-6d %s\n", g, formatStack(stack, 80))
fmt.Fprintf(w, " g%-6d %s\n", g, formatStack(stack, 80))
fmt.Fprintf(w, " %d schedulings, total: %s, max: %s\n",
schedCount, fmtDuration(totalWait), fmtDuration(maxWait))
}
Expand Down
40 changes: 20 additions & 20 deletions testdata/experiment-upsert1000-gateway.txt
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -25,7 +25,7 @@ Events: 1554412
--- Spike Details ---

[1] t=4200ms [latency] p99=6.85ms
→ G852 waited 9.44ms on P6
→ g852 waited 9.44ms on P6
→ Burst: 125 goroutines became runnable within ±1ms
Breakdown: 139 unblocked
Unblocked by (94): selectgo
Expand All @@ -34,10 +34,10 @@ Events: 1554412
Blocked at (94): (*streamPool[...]).newPooledStream.func2 → (*pooledStream[...]).runOnce [selectgo]
Blocked at (10): gopark [gopark]
Blocked at (9): ReadFull → ReadAtLeast → (*Reader).Read → (*conn).Read → (*netFD).Read [(*FD).Read]
Heavy unblocker G841 (4): (*writeBatch).CommitNoSyncWait → (*DB).applyInternal → (*commitPipeline).prepare
Heavy unblocker G917 (4): (*Store).enqueueRaftUpdateCheck → (*raftScheduler).enqueue1 → (*raftSchedulerShard).signal
Heavy unblocker G2228028 (4): (*LogWriter).flushPending → (*pendingSyncsWithSyncQueue).pop → (*syncQueue).pop
→ Longest run during wait: G901 ran 886.8µs
Heavy unblocker g841 (4): (*writeBatch).CommitNoSyncWait → (*DB).applyInternal → (*commitPipeline).prepare
Heavy unblocker g917 (4): (*Store).enqueueRaftUpdateCheck → (*raftScheduler).enqueue1 → (*raftSchedulerShard).signal
Heavy unblocker g2228028 (4): (*LogWriter).flushPending → (*pendingSyncsWithSyncQueue).pop → (*syncQueue).pop
→ Longest run during wait: g901 ran 886.8µs
(*DB).applyInternal → (*commitPipeline).prepare
→ Queue activity: 174 goroutines ran 176 times during the wait

Expand All @@ -49,9 +49,9 @@ Events: 1554412
Blocked at (832): (*raftScheduler).Start.func2 → (*raftSchedulerShard).worker [(*Cond).Wait]
Blocked at (152): (*DB).applyInternal → (*commitPipeline).publish [(*WaitGroup).Wait]
Blocked at (40): (*DB).applyInternal → (*commitPipeline).prepare [(*Mutex).Lock]
Heavy unblocker G1222 (389): (*Store).HandleRaftRequest → (*raftScheduler).enqueue1 → (*raftSchedulerShard).signal
Heavy unblocker G916 (160): (*Store).enqueueRaftUpdateCheck → (*raftScheduler).enqueue1 → (*raftSchedulerShard).signal
Heavy unblocker G919 (121): (*Store).enqueueRaftUpdateCheck → (*raftScheduler).enqueue1 → (*raftSchedulerShard).signal
Heavy unblocker g1222 (389): (*Store).HandleRaftRequest → (*raftScheduler).enqueue1 → (*raftSchedulerShard).signal
Heavy unblocker g916 (160): (*Store).enqueueRaftUpdateCheck → (*raftScheduler).enqueue1 → (*raftSchedulerShard).signal
Heavy unblocker g919 (121): (*Store).enqueueRaftUpdateCheck → (*raftScheduler).enqueue1 → (*raftSchedulerShard).signal
Created by (1): (*Task).applyOneBatch → (*replicaAppBatch).ApplyToStateMachine → (*baseQueue).Async

[3] t=9800ms [runnable] peak 237 runnable
Expand All @@ -62,9 +62,9 @@ Events: 1554412
Blocked at (692): (*raftScheduler).Start.func2 → (*raftSchedulerShard).worker [(*Cond).Wait]
Blocked at (86): (*DB).applyInternal → (*commitPipeline).publish [(*WaitGroup).Wait]
Blocked at (60): (*DB).applyInternal → (*commitPipeline).prepare [(*Mutex).Lock]
Heavy unblocker G916 (169): (*Store).enqueueRaftUpdateCheck → (*raftScheduler).enqueue1 → (*raftSchedulerShard).signal
Heavy unblocker G1222 (161): (*Store).HandleRaftRequest → (*raftScheduler).enqueue1 → (*raftSchedulerShard).signal
Heavy unblocker G919 (160): (*Store).enqueueRaftUpdateCheck → (*raftScheduler).enqueue1 → (*raftSchedulerShard).signal
Heavy unblocker g916 (169): (*Store).enqueueRaftUpdateCheck → (*raftScheduler).enqueue1 → (*raftSchedulerShard).signal
Heavy unblocker g1222 (161): (*Store).HandleRaftRequest → (*raftScheduler).enqueue1 → (*raftSchedulerShard).signal
Heavy unblocker g919 (160): (*Store).enqueueRaftUpdateCheck → (*raftScheduler).enqueue1 → (*raftSchedulerShard).signal
Created by (4): (*Task).applyOneBatch → (*replicaAppBatch).ApplyToStateMachine → (*baseQueue).Async

[4] t=6700ms [runnable] peak 221 runnable
Expand All @@ -75,9 +75,9 @@ Events: 1554412
Blocked at (315): (*streamPool[...]).Send → (*pooledStream[...]).Send [selectgo]
Blocked at (315): ReadAtLeast → (*transportReader).Read → (*recvBufferReader).readClient [selectgo]
Blocked at (240): (*raftScheduler).Start.func2 → (*raftSchedulerShard).worker [(*Cond).Wait]
Heavy unblocker G1012 (315): (*http2Client).handleData → (*Stream).write → (*recvBuffer).put
Heavy unblocker G1222 (79): (*Store).HandleRaftRequest → (*raftScheduler).enqueue1 → (*raftSchedulerShard).signal
Heavy unblocker G917 (49): (*Store).enqueueRaftUpdateCheck → (*raftScheduler).enqueue1 → (*raftSchedulerShard).signal
Heavy unblocker g1012 (315): (*http2Client).handleData → (*Stream).write → (*recvBuffer).put
Heavy unblocker g1222 (79): (*Store).HandleRaftRequest → (*raftScheduler).enqueue1 → (*raftSchedulerShard).signal
Heavy unblocker g917 (49): (*Store).enqueueRaftUpdateCheck → (*raftScheduler).enqueue1 → (*raftSchedulerShard).signal
Created by (352): (*txnLockGatekeeper).SendLocked → (*DistSender).sendPartialBatchAsync
Created by (3): (*Task).applyOneBatch → (*replicaAppBatch).ApplyToStateMachine → (*baseQueue).Async
Created by (1): (*Task).applyOneBatch → (*replicaAppBatch).ApplyToStateMachine → (*baseQueue).Async
Expand All @@ -90,9 +90,9 @@ Events: 1554412
Blocked at (828): (*raftScheduler).Start.func2 → (*raftSchedulerShard).worker [(*Cond).Wait]
Blocked at (184): (*DB).applyInternal → (*commitPipeline).publish [(*WaitGroup).Wait]
Blocked at (104): (*Batch).Commit → (*DB).applyInternal → (*commitPipeline).publish [(*WaitGroup).Wait]
Heavy unblocker G1222 (313): (*Store).HandleRaftRequest → (*raftScheduler).enqueue1 → (*raftSchedulerShard).signal
Heavy unblocker G918 (147): (*Store).enqueueRaftUpdateCheck → (*raftScheduler).enqueue1 → (*raftSchedulerShard).signal
Heavy unblocker G916 (131): (*Store).enqueueRaftUpdateCheck → (*raftScheduler).enqueue1 → (*raftSchedulerShard).signal
Heavy unblocker g1222 (313): (*Store).HandleRaftRequest → (*raftScheduler).enqueue1 → (*raftSchedulerShard).signal
Heavy unblocker g918 (147): (*Store).enqueueRaftUpdateCheck → (*raftScheduler).enqueue1 → (*raftSchedulerShard).signal
Heavy unblocker g916 (131): (*Store).enqueueRaftUpdateCheck → (*raftScheduler).enqueue1 → (*raftSchedulerShard).signal
Created by (4): (*Task).applyOneBatch → (*replicaAppBatch).ApplyToStateMachine → (*baseQueue).Async
Created by (2): (*Task).applyOneBatch → (*replicaAppBatch).ApplyToStateMachine → (*baseQueue).Async

Expand All @@ -104,8 +104,8 @@ Events: 1554412
Blocked at (257): (*raftScheduler).Start.func2 → (*raftSchedulerShard).worker [(*Cond).Wait]
Blocked at (222): (*DB).applyInternal → (*commitPipeline).publish [(*WaitGroup).Wait]
Blocked at (208): (*streamPool[...]).newPooledStream.func2 → (*pooledStream[...]).runOnce [selectgo]
Heavy unblocker G1222 (116): (*Store).HandleRaftRequest → (*raftScheduler).enqueue1 → (*raftSchedulerShard).signal
Heavy unblocker G916 (50): (*Store).enqueueRaftUpdateCheck → (*raftScheduler).enqueue1 → (*raftSchedulerShard).signal
Heavy unblocker G917 (34): (*Store).enqueueRaftUpdateCheck → (*raftScheduler).enqueue1 → (*raftSchedulerShard).signal
Heavy unblocker g1222 (116): (*Store).HandleRaftRequest → (*raftScheduler).enqueue1 → (*raftSchedulerShard).signal
Heavy unblocker g916 (50): (*Store).enqueueRaftUpdateCheck → (*raftScheduler).enqueue1 → (*raftSchedulerShard).signal
Heavy unblocker g917 (34): (*Store).enqueueRaftUpdateCheck → (*raftScheduler).enqueue1 → (*raftSchedulerShard).signal
Created by (28): (*Task).applyOneBatch → (*replicaAppBatch).ApplyToStateMachine → (*baseQueue).Async
Created by (3): (*Task).applyOneBatch → (*replicaAppBatch).ApplyToStateMachine → (*baseQueue).Async
Loading