test(module): a blocked stdin write is released when the pump stops
- commit
- 487baa8b2c2efad3f01c9b7e0231e12c296204f1
- author
- brice · · 2026-10-09 03:59 UTC
- parent
- 718054e
- signature
- GPG Verified
CD88F9E564E0102646FB6077A1A07C7C7C3E349D· arch-gpg Verified for the committer email
1 file · +42 -0
M internal/module/module_test.go +42 -0
| @@ -227,6 +227,48 @@ func TestSendAllocs(t *testing.T) { | |||
| 227 | 227 | } | |
| 228 | 228 | } | |
| 229 | 229 | ||
| 230 | + | // A script that never reads its stdin blocks the writer once the pipe is | |
| 231 | + | // full; stop must still return, because closing the pipe ends the Write. | |
| 232 | + | func TestPumpInputStopUnblocksAWrite(t *testing.T) { | |
| 233 | + | r, w, err := os.Pipe() // r is never read | |
| 234 | + | if err != nil { | |
| 235 | + | t.Fatal(err) | |
| 236 | + | } | |
| 237 | + | defer r.Close() | |
| 238 | + | m, _ := newTest() | |
| 239 | + | m.Interactive = true | |
| 240 | + | stop := m.pumpInput(w) | |
| 241 | + | ||
| 242 | + | // Push well over the 64 KiB a pipe holds, until the queue stays full: then | |
| 243 | + | // the writer is stuck in Write. | |
| 244 | + | line := []byte(strings.Repeat("x", 100)) | |
| 245 | + | sent := 0 | |
| 246 | + | deadline := time.Now().Add(5 * time.Second) | |
| 247 | + | for full := 0; full < 50; { | |
| 248 | + | if time.Now().After(deadline) { | |
| 249 | + | t.Fatalf("the writer never blocked after %d lines", sent) | |
| 250 | + | } | |
| 251 | + | if m.Send(line) { | |
| 252 | + | sent++ | |
| 253 | + | full = 0 | |
| 254 | + | } else { | |
| 255 | + | full++ | |
| 256 | + | time.Sleep(time.Millisecond) | |
| 257 | + | } | |
| 258 | + | } | |
| 259 | + | if sent*101 <= 64<<10 { | |
| 260 | + | t.Fatalf("only %d bytes went in; the pipe was not full", sent*101) | |
| 261 | + | } | |
| 262 | + | ||
| 263 | + | done := make(chan struct{}) | |
| 264 | + | go func() { stop(); close(done) }() | |
| 265 | + | select { | |
| 266 | + | case <-done: | |
| 267 | + | case <-time.After(time.Second): | |
| 268 | + | t.Fatal("stop did not return: the blocked Write was not released") | |
| 269 | + | } | |
| 270 | + | } | |
| 271 | + | ||
| 230 | 272 | func TestGenCountsStarts(t *testing.T) { | |
| 231 | 273 | wake := make(chan struct{}, 1) | |
| 232 | 274 | m := New("t", Left, "printf 'hi\\n'", wake, slog.New(slog.DiscardHandler)) | |
| @@ -227,6 +227,48 @@ func TestSendAllocs(t *testing.T) { | |||||
| 227 | } | 227 | } | ||
| 228 | } | 228 | } | ||
| 229 | 229 | ||||
| 230 | + | // A script that never reads its stdin blocks the writer once the pipe is | |||
| 231 | + | // full; stop must still return, because closing the pipe ends the Write. | |||
| 232 | + | func TestPumpInputStopUnblocksAWrite(t *testing.T) { | |||
| 233 | + | r, w, err := os.Pipe() // r is never read | |||
| 234 | + | if err != nil { | |||
| 235 | + | t.Fatal(err) | |||
| 236 | + | } | |||
| 237 | + | defer r.Close() | |||
| 238 | + | m, _ := newTest() | |||
| 239 | + | m.Interactive = true | |||
| 240 | + | stop := m.pumpInput(w) | |||
| 241 | + | ||||
| 242 | + | // Push well over the 64 KiB a pipe holds, until the queue stays full: then | |||
| 243 | + | // the writer is stuck in Write. | |||
| 244 | + | line := []byte(strings.Repeat("x", 100)) | |||
| 245 | + | sent := 0 | |||
| 246 | + | deadline := time.Now().Add(5 * time.Second) | |||
| 247 | + | for full := 0; full < 50; { | |||
| 248 | + | if time.Now().After(deadline) { | |||
| 249 | + | t.Fatalf("the writer never blocked after %d lines", sent) | |||
| 250 | + | } | |||
| 251 | + | if m.Send(line) { | |||
| 252 | + | sent++ | |||
| 253 | + | full = 0 | |||
| 254 | + | } else { | |||
| 255 | + | full++ | |||
| 256 | + | time.Sleep(time.Millisecond) | |||
| 257 | + | } | |||
| 258 | + | } | |||
| 259 | + | if sent*101 <= 64<<10 { | |||
| 260 | + | t.Fatalf("only %d bytes went in; the pipe was not full", sent*101) | |||
| 261 | + | } | |||
| 262 | + | ||||
| 263 | + | done := make(chan struct{}) | |||
| 264 | + | go func() { stop(); close(done) }() | |||
| 265 | + | select { | |||
| 266 | + | case <-done: | |||
| 267 | + | case <-time.After(time.Second): | |||
| 268 | + | t.Fatal("stop did not return: the blocked Write was not released") | |||
| 269 | + | } | |||
| 270 | + | } | |||
| 271 | + | ||||
| 230 | func TestGenCountsStarts(t *testing.T) { | 272 | func TestGenCountsStarts(t *testing.T) { | ||
| 231 | wake := make(chan struct{}, 1) | 273 | wake := make(chan struct{}, 1) | ||
| 232 | m := New("t", Left, "printf 'hi\\n'", wake, slog.New(slog.DiscardHandler)) | 274 | m := New("t", Left, "printf 'hi\\n'", wake, slog.New(slog.DiscardHandler)) | ||