6558c1f fix(app): stop the watchdog probes from leaking and outliving the session · neferwl

M internal/adapters/watchdog/watchdog.go +6 -2


@@ -9,6 +9,7 @@ import (
9	9		"os"
10	10		"path/filepath"
11	11		"runtime/pprof"
	12	+	"sync"
12	13		"time"
13	14
14	15		"github.com/bnema/zerowrap"

@@ -31,8 +32,11 @@ type Options struct {
31	32
32	33		// Run probes each loop probes returns until ctx ends; the set may change
33	34		// between ticks (outputs come and go). On a stall it writes one dump per
34		-	// stall: a loop must answer again before it is dumped again.
	35	+	// stall: a loop must answer again before it is dumped again. Every Ping
	36	+	// must return once ctx ends: Run waits for the probes in flight.
35	37		func Run(ctx context.Context, probes func() []Probe, o Options) {
	38	+	var inflight sync.WaitGroup
	39	+	defer inflight.Wait()
36	40		type state struct {
37	41		answer chan struct{} // the probe in flight, nil when none
38	42		since time.Time

@@ -75,7 +79,7 @@ func Run(ctx context.Context, probes func() []Probe, o Options) {
75	79		}
76	80		answer := make(chan struct{})
77	81		s.answer, s.since = answer, now
78		-	go func() { p.Ping(); close(answer) }()
	82	+	inflight.Go(func() { p.Ping(); close(answer) })
79	83		}
80	84		// A loop that left (an unplugged output) is forgotten; its
81	85		// probe goroutine ends with the loop's context.

 	"os"
 	"path/filepath"
 	"runtime/pprof"
+	"sync"
 	"time"
 	"github.com/bnema/zerowrap"
 // Run probes each loop probes returns until ctx ends; the set may change
 // between ticks (outputs come and go). On a stall it writes one dump per
-// stall: a loop must answer again before it is dumped again.
+// stall: a loop must answer again before it is dumped again. Every Ping
+// must return once ctx ends: Run waits for the probes in flight.
 func Run(ctx context.Context, probes func() []Probe, o Options) {
+	var inflight sync.WaitGroup
+	defer inflight.Wait()
 	type state struct {
 		answer  chan struct{} // the probe in flight, nil when none
 		since   time.Time
 				}
 				answer := make(chan struct{})
 				s.answer, s.since = answer, now
-				go func() { p.Ping(); close(answer) }()
+				inflight.Go(func() { p.Ping(); close(answer) })
 			}
 			// A loop that left (an unplugged output) is forgotten; its
 			// probe goroutine ends with the loop's context.

M internal/adapters/watchdog/watchdog_test.go +57 -46


@@ -5,8 +5,9 @@ import (
5	5		"os"
6	6		"path/filepath"
7	7		"strings"
8		-	"sync"
	8	+	"sync/atomic"
9	9		"testing"
	10	+	"testing/synctest"
10	11		"time"
11	12
12	13		"github.com/bnema/zerowrap"

@@ -14,53 +15,63 @@ import (
14	15
15	16		// stuckOwnerLoop blocks like an owner goroutine that never reaches its
16	17		// select again; the dump must name it.
17		-	func stuckOwnerLoop(release <-chan struct{}) { <-release }
	18	+	func stuckOwnerLoop(ctx context.Context, release <-chan struct{}) {
	19	+	select {
	20	+	case <-release:
	21	+	case <-ctx.Done():
	22	+	}
	23	+	}
18	24
19	25		func TestStalledLoopIsDumpedOnce(t *testing.T) {
20	26		dir := t.TempDir()
21		-	release := make(chan struct{})
22		-	var mu sync.Mutex
23		-	answered := 0
24		-	probes := func() []Probe {
25		-	return []Probe{
26		-	{Name: "stuck", Ping: func() { stuckOwnerLoop(release) }},
27		-	{Name: "fine", Ping: func() { mu.Lock(); answered++; mu.Unlock() }},
	27	+	synctest.Test(t, func(t *testing.T) {
	28	+	ctx, cancel := context.WithCancel(context.Background())
	29	+	release := make(chan struct{})
	30	+	var answered atomic.Int32
	31	+	probes := func() []Probe {
	32	+	return []Probe{
	33	+	{Name: "stuck", Ping: func() { stuckOwnerLoop(ctx, release) }},
	34	+	{Name: "fine", Ping: func() { answered.Add(1) }},
	35	+	}
28	36		}
29		-	}
30		-	ctx, cancel := context.WithCancel(context.Background())
31		-	done := make(chan struct{})
32		-	go func() {
33		-	defer close(done)
34		-	Run(ctx, probes, Options{Dir: dir, Every: 5 * time.Millisecond, Timeout: 30 * time.Millisecond, Log: zerowrap.Default()})
35		-	}()
36		-	deadline := time.Now().Add(2 * time.Second)
37		-	var dumps []string
38		-	for len(dumps) == 0 && time.Now().Before(deadline) {
39		-	time.Sleep(5 * time.Millisecond)
40		-	dumps, _ = filepath.Glob(filepath.Join(dir, "stall-*-stuck.txt"))
41		-	}
42		-	if len(dumps) != 1 {
43		-	t.Fatalf("dumps %v", dumps)
44		-	}
45		-	// Still stuck: no second dump.
46		-	time.Sleep(100 * time.Millisecond)
47		-	if again, _ := filepath.Glob(filepath.Join(dir, "stall-*.txt")); len(again) != 1 {
48		-	t.Fatalf("dumps after the first %v", again)
49		-	}
50		-	data, err := os.ReadFile(dumps[0])
51		-	if err != nil {
52		-	t.Fatal(err)
53		-	}
54		-	if !strings.Contains(string(data), "stuckOwnerLoop") {
55		-	t.Fatalf("dump does not name the stuck function:\n%s", data)
56		-	}
57		-	mu.Lock()
58		-	n := answered
59		-	mu.Unlock()
60		-	if n < 2 {
61		-	t.Fatalf("a healthy loop was probed %d times", n)
62		-	}
63		-	close(release)
64		-	cancel()
65		-	<-done
	37	+	done := make(chan struct{})
	38	+	go func() {
	39	+	defer close(done)
	40	+	Run(ctx, probes, Options{Dir: dir, Every: 10 * time.Millisecond, Timeout: 50 * time.Millisecond, Log: zerowrap.Default()})
	41	+	}()
	42	+	dumps := func() []string {
	43	+	m, _ := filepath.Glob(filepath.Join(dir, "stall-*.txt"))
	44	+	return m
	45	+	}
	46	+	time.Sleep(40 * time.Millisecond)
	47	+	synctest.Wait()
	48	+	if d := dumps(); len(d) != 0 {
	49	+	t.Fatalf("dumped before the timeout: %v", d)
	50	+	}
	51	+	time.Sleep(30 * time.Millisecond)
	52	+	synctest.Wait()
	53	+	d := dumps()
	54	+	if len(d) != 1 \|\| !strings.HasSuffix(d[0], "-stuck.txt") {
	55	+	t.Fatalf("dumps %v", d)
	56	+	}
	57	+	// Still stuck: no second dump.
	58	+	time.Sleep(time.Second)
	59	+	synctest.Wait()
	60	+	if again := dumps(); len(again) != 1 {
	61	+	t.Fatalf("dumps after the first %v", again)
	62	+	}
	63	+	data, err := os.ReadFile(d[0])
	64	+	if err != nil {
	65	+	t.Fatal(err)
	66	+	}
	67	+	if !strings.Contains(string(data), "stuckOwnerLoop") {
	68	+	t.Fatalf("dump does not name the stuck function:\n%s", data)
	69	+	}
	70	+	if n := answered.Load(); n < 50 {
	71	+	t.Fatalf("a healthy loop was probed %d times in 1 s", n)
	72	+	}
	73	+	// Run waits for its probes: a stuck one ends with ctx.
	74	+	cancel()
	75	+	<-done
	76	+	})
66	77		}


@@ -5,8 +5,9 @@ import (
5		"os"	5		"os"
6		"path/filepath"	6		"path/filepath"
7		"strings"	7		"strings"
8	-	"sync"	8	+	"sync/atomic"
9		"testing"	9		"testing"
			10	+	"testing/synctest"
10		"time"	11		"time"
11			12
12		"github.com/bnema/zerowrap"	13		"github.com/bnema/zerowrap"

@@ -14,53 +15,63 @@ import (
14			15
15		// stuckOwnerLoop blocks like an owner goroutine that never reaches its	16		// stuckOwnerLoop blocks like an owner goroutine that never reaches its
16		// select again; the dump must name it.	17		// select again; the dump must name it.
17	-	func stuckOwnerLoop(release <-chan struct{}) { <-release }	18	+	func stuckOwnerLoop(ctx context.Context, release <-chan struct{}) {
			19	+	select {
			20	+	case <-release:
			21	+	case <-ctx.Done():
			22	+	}
			23	+	}
18			24
19		func TestStalledLoopIsDumpedOnce(t *testing.T) {	25		func TestStalledLoopIsDumpedOnce(t *testing.T) {
20		dir := t.TempDir()	26		dir := t.TempDir()
21	-	release := make(chan struct{})	27	+	synctest.Test(t, func(t *testing.T) {
22	-	var mu sync.Mutex	28	+	ctx, cancel := context.WithCancel(context.Background())
23	-	answered := 0	29	+	release := make(chan struct{})
24	-	probes := func() []Probe {	30	+	var answered atomic.Int32
25	-	return []Probe{	31	+	probes := func() []Probe {
26	-	{Name: "stuck", Ping: func() { stuckOwnerLoop(release) }},	32	+	return []Probe{
27	-	{Name: "fine", Ping: func() { mu.Lock(); answered++; mu.Unlock() }},	33	+	{Name: "stuck", Ping: func() { stuckOwnerLoop(ctx, release) }},
			34	+	{Name: "fine", Ping: func() { answered.Add(1) }},
			35	+	}
28		}	36		}
29	-	}	37	+	done := make(chan struct{})
30	-	ctx, cancel := context.WithCancel(context.Background())	38	+	go func() {
31	-	done := make(chan struct{})	39	+	defer close(done)
32	-	go func() {	40	+	Run(ctx, probes, Options{Dir: dir, Every: 10 * time.Millisecond, Timeout: 50 * time.Millisecond, Log: zerowrap.Default()})
33	-	defer close(done)	41	+	}()
34	-	Run(ctx, probes, Options{Dir: dir, Every: 5 * time.Millisecond, Timeout: 30 * time.Millisecond, Log: zerowrap.Default()})	42	+	dumps := func() []string {
35	-	}()	43	+	m, _ := filepath.Glob(filepath.Join(dir, "stall-*.txt"))
36	-	deadline := time.Now().Add(2 * time.Second)	44	+	return m
37	-	var dumps []string	45	+	}
38	-	for len(dumps) == 0 && time.Now().Before(deadline) {	46	+	time.Sleep(40 * time.Millisecond)
39	-	time.Sleep(5 * time.Millisecond)	47	+	synctest.Wait()
40	-	dumps, _ = filepath.Glob(filepath.Join(dir, "stall-*-stuck.txt"))	48	+	if d := dumps(); len(d) != 0 {
41	-	}	49	+	t.Fatalf("dumped before the timeout: %v", d)
42	-	if len(dumps) != 1 {	50	+	}
43	-	t.Fatalf("dumps %v", dumps)	51	+	time.Sleep(30 * time.Millisecond)
44	-	}	52	+	synctest.Wait()
45	-	// Still stuck: no second dump.	53	+	d := dumps()
46	-	time.Sleep(100 * time.Millisecond)	54	+	if len(d) != 1 \|\| !strings.HasSuffix(d[0], "-stuck.txt") {
47	-	if again, _ := filepath.Glob(filepath.Join(dir, "stall-*.txt")); len(again) != 1 {	55	+	t.Fatalf("dumps %v", d)
48	-	t.Fatalf("dumps after the first %v", again)	56	+	}
49	-	}	57	+	// Still stuck: no second dump.
50	-	data, err := os.ReadFile(dumps[0])	58	+	time.Sleep(time.Second)
51	-	if err != nil {	59	+	synctest.Wait()
52	-	t.Fatal(err)	60	+	if again := dumps(); len(again) != 1 {
53	-	}	61	+	t.Fatalf("dumps after the first %v", again)
54	-	if !strings.Contains(string(data), "stuckOwnerLoop") {	62	+	}
55	-	t.Fatalf("dump does not name the stuck function:\n%s", data)	63	+	data, err := os.ReadFile(d[0])
56	-	}	64	+	if err != nil {
57	-	mu.Lock()	65	+	t.Fatal(err)
58	-	n := answered	66	+	}
59	-	mu.Unlock()	67	+	if !strings.Contains(string(data), "stuckOwnerLoop") {
60	-	if n < 2 {	68	+	t.Fatalf("dump does not name the stuck function:\n%s", data)
61	-	t.Fatalf("a healthy loop was probed %d times", n)	69	+	}
62	-	}	70	+	if n := answered.Load(); n < 50 {
63	-	close(release)	71	+	t.Fatalf("a healthy loop was probed %d times in 1 s", n)
64	-	cancel()	72	+	}
65	-	<-done	73	+	// Run waits for its probes: a stuck one ends with ctx.
			74	+	cancel()
			75	+	<-done
			76	+	})
66		}	77		}

M internal/adapters/wayland/probe.go +13 -3

@@ -1,5 +1,15 @@
1	1		package wayland
2	2
3		-	// Ping returns once the display loop ran an empty task, or once the display
4		-	// stopped. The watchdog uses it to detect a stuck display loop.
5		-	func (s *Server) Ping() { s.display.Do(func() {}) }
	3	+	import "context"
	4	+
	5	+	// Ping returns once the display loop ran an empty task, the display
	6	+	// stopped or ctx ended. The watchdog uses it to detect a stuck display
	7	+	// loop; a stuck loop leaves only the inner Do goroutine behind.
	8	+	func (s *Server) Ping(ctx context.Context) {
	9	+	done := make(chan struct{})
	10	+	go func() { s.display.Do(func() {}); close(done) }()
	11	+	select {
	12	+	case <-done:
	13	+	case <-ctx.Done():
	14	+	}
	15	+	}

@@ -1,5 +1,15 @@
1		package wayland	1		package wayland
2			2
3	-	// Ping returns once the display loop ran an empty task, or once the display	3	+	import "context"
4	-	// stopped. The watchdog uses it to detect a stuck display loop.	4	+
5	-	func (s *Server) Ping() { s.display.Do(func() {}) }	5	+	// Ping returns once the display loop ran an empty task, the display
			6	+	// stopped or ctx ended. The watchdog uses it to detect a stuck display
			7	+	// loop; a stuck loop leaves only the inner Do goroutine behind.
			8	+	func (s *Server) Ping(ctx context.Context) {
			9	+	done := make(chan struct{})
			10	+	go func() { s.display.Do(func() {}); close(done) }()
			11	+	select {
			12	+	case <-done:
			13	+	case <-ctx.Done():
			14	+	}
			15	+	}

M internal/app/drm.go +8 -11


@@ -122,14 +122,14 @@ type drmBackend struct {
122	122		preparers map[seatDisablePreparer]struct{}
123	123		// probes are the running output loops the watchdog pings, by
124	124		// connector, guarded by mu.
125		-	probes map[string]func(context.Context)
	125	+	probes map[string]func()
126	126		}
127	127
128		-	func (b *drmBackend) addProbe(name string, ping func(context.Context)) {
	128	+	func (b *drmBackend) addProbe(name string, ping func()) {
129	129		b.mu.Lock()
130	130		defer b.mu.Unlock()
131	131		if b.probes == nil {
132		-	b.probes = map[string]func(context.Context){}
	132	+	b.probes = map[string]func(){}
133	133		}
134	134		b.probes[name] = ping
135	135		}

@@ -141,12 +141,12 @@ func (b *drmBackend) removeProbe(name string) {
141	141		}
142	142
143	143		// watchdogProbes lists the running output loops for the watchdog.
144		-	func (b *drmBackend) watchdogProbes(ctx context.Context) []watchdog.Probe {
	144	+	func (b *drmBackend) watchdogProbes() []watchdog.Probe {
145	145		b.mu.Lock()
146	146		defer b.mu.Unlock()
147	147		out := make([]watchdog.Probe, 0, len(b.probes))
148	148		for name, ping := range b.probes {
149		-	out = append(out, watchdog.Probe{Name: "output-" + name, Ping: func() { ping(ctx) }})
	149	+	out = append(out, watchdog.Probe{Name: "output-" + name, Ping: ping})
150	150		}
151	151		return out
152	152		}

@@ -520,12 +520,9 @@ func (b *drmBackend) runOutputs(ctx context.Context, want func(ports.Config) drm
520	520		defer o.Close()
521	521		b.addPreparer(o)
522	522		defer b.removePreparer(o) // before Close: it handles no request any more
523		-	b.addProbe(name, func(pctx context.Context) {
524		-	pctx, stop := context.WithCancel(pctx)
525		-	defer stop()
526		-	context.AfterFunc(octx, stop) // the probe ends with this output
527		-	o.Ping(pctx)
528		-	})
	523	+	// octx derives from the session: the probe ends with this
	524	+	// output or the session.
	525	+	b.addProbe(name, func() { o.Ping(octx) })
529	526		defer b.removeProbe(name)
530	527		if realtime {
531	528		runtime.LockOSThread()

 	preparers map[seatDisablePreparer]struct{}
 	// probes are the running output loops the watchdog pings, by
 	// connector, guarded by mu.
-	probes map[string]func(context.Context)
+	probes map[string]func()
 }
-func (b *drmBackend) addProbe(name string, ping func(context.Context)) {
+func (b *drmBackend) addProbe(name string, ping func()) {
 	b.mu.Lock()
 	defer b.mu.Unlock()
 	if b.probes == nil {
-		b.probes = map[string]func(context.Context){}
+		b.probes = map[string]func(){}
 	}
 	b.probes[name] = ping
 }
 }
 // watchdogProbes lists the running output loops for the watchdog.
-func (b *drmBackend) watchdogProbes(ctx context.Context) []watchdog.Probe {
+func (b *drmBackend) watchdogProbes() []watchdog.Probe {
 	b.mu.Lock()
 	defer b.mu.Unlock()
 	out := make([]watchdog.Probe, 0, len(b.probes))
 	for name, ping := range b.probes {
-		out = append(out, watchdog.Probe{Name: "output-" + name, Ping: func() { ping(ctx) }})
+		out = append(out, watchdog.Probe{Name: "output-" + name, Ping: ping})
 	}
 	return out
 }
 						defer o.Close()
 						b.addPreparer(o)
 						defer b.removePreparer(o) // before Close: it handles no request any more
-						b.addProbe(name, func(pctx context.Context) {
+						// octx derives from the session: the probe ends with this
-							pctx, stop := context.WithCancel(pctx)
+						// output or the session.
-							defer stop()
+						b.addProbe(name, func() { o.Ping(octx) })
-							context.AfterFunc(octx, stop) // the probe ends with this output
-							o.Ping(pctx)
-						})
 						defer b.removeProbe(name)
 						if realtime {
 							runtime.LockOSThread()

M internal/app/session.go +3 -3


@@ -309,14 +309,14 @@ func (s *session) run(ctx context.Context, inject func(chan<- ports.InputEvent))
309	309		defer workers.Done()
310	310		watchdog.Run(ctx, func() []watchdog.Probe {
311	311		probes := []watchdog.Probe{
312		-	{Name: "wayland", Ping: server.Ping},
	312	+	{Name: "wayland", Ping: func() { server.Ping(ctx) }},
313	313		{Name: "core", Ping: func() { c.Ping(ctx) }},
314	314		}
315	315		if hw != nil {
316		-	probes = append(probes, hw.watchdogProbes(ctx)...)
	316	+	probes = append(probes, hw.watchdogProbes()...)
317	317		}
318	318		return probes
319		-	}, watchdog.Options{Dir: dir, Every: 500 * time.Millisecond, Timeout: 2 * time.Second, Log: logging.For(ctx, "app")})
	319	+	}, watchdog.Options{Dir: dir, Every: time.Second, Timeout: 3 * time.Second, Log: logging.For(ctx, "app")})
320	320		}()
321	321		}
322	322		go func() {


@@ -309,14 +309,14 @@ func (s *session) run(ctx context.Context, inject func(chan<- ports.InputEvent))
309		defer workers.Done()	309		defer workers.Done()
310		watchdog.Run(ctx, func() []watchdog.Probe {	310		watchdog.Run(ctx, func() []watchdog.Probe {
311		probes := []watchdog.Probe{	311		probes := []watchdog.Probe{
312	-	{Name: "wayland", Ping: server.Ping},	312	+	{Name: "wayland", Ping: func() { server.Ping(ctx) }},
313		{Name: "core", Ping: func() { c.Ping(ctx) }},	313		{Name: "core", Ping: func() { c.Ping(ctx) }},
314		}	314		}
315		if hw != nil {	315		if hw != nil {
316	-	probes = append(probes, hw.watchdogProbes(ctx)...)	316	+	probes = append(probes, hw.watchdogProbes()...)
317		}	317		}
318		return probes	318		return probes
319	-	}, watchdog.Options{Dir: dir, Every: 500 * time.Millisecond, Timeout: 2 * time.Second, Log: logging.For(ctx, "app")})	319	+	}, watchdog.Options{Dir: dir, Every: time.Second, Timeout: 3 * time.Second, Log: logging.For(ctx, "app")})
320		}()	320		}()
321		}	321		}
322		go func() {	322		go func() {