feat(module): feed interactive modules a stdin of pointer lines
- commit
- 6692ae1a0c9589b8d00f24110826b6778e6a3680
- author
- brice · · 2026-10-09 03:34 UTC
- parent
- 6010507
- signature
- GPG Verified
CD88F9E564E0102646FB6077A1A07C7C7C3E349D· arch-gpg Verified for the committer email
2 files · +177 -1
M internal/module/module.go +82 -1
| @@ -46,22 +46,56 @@ type Module struct { | |||
| 46 | 46 | Exec string | |
| 47 | 47 | // Env is added to the script's environment, as KEY=value entries. | |
| 48 | 48 | Env []string | |
| 49 | + | // Interactive gives the script a stdin that carries Send lines. Set it | |
| 50 | + | // before Start. | |
| 51 | + | Interactive bool | |
| 49 | 52 | ||
| 50 | 53 | mu sync.Mutex | |
| 51 | 54 | pending []byte // latest complete frame | |
| 52 | 55 | dirty bool | |
| 53 | 56 | failed bool // the script is not running; show an error marker | |
| 54 | 57 | ||
| 58 | + | input chan event // lines for the script's stdin | |
| 59 | + | ||
| 55 | 60 | wake chan<- struct{} | |
| 56 | 61 | log *slog.Logger | |
| 57 | 62 | ||
| 58 | 63 | done sync.WaitGroup // Start adds one; Run releases it when its process is gone | |
| 59 | 64 | } | |
| 60 | 65 | ||
| 66 | + | // maxLine is the longest line Send accepts, without its newline. | |
| 67 | + | const maxLine = 127 | |
| 68 | + | ||
| 69 | + | // event is one stdin line, newline included. | |
| 70 | + | type event struct { | |
| 71 | + | b [maxLine + 1]byte | |
| 72 | + | n uint8 | |
| 73 | + | } | |
| 74 | + | ||
| 61 | 75 | // New creates a module. wake receives a non-blocking signal after each frame. | |
| 62 | 76 | func New(name string, zone Zone, command string, wake chan<- struct{}, log *slog.Logger) *Module { | |
| 63 | 77 | return &Module{Name: name, Zone: zone, Exec: command, wake: wake, log: log, | |
| 64 | - | pending: make([]byte, 0, MaxFrame)} | |
| 78 | + | pending: make([]byte, 0, MaxFrame), input: make(chan event, 32)} | |
| 79 | + | } | |
| 80 | + | ||
| 81 | + | // Send queues line (without a newline) for the script's stdin. It never | |
| 82 | + | // blocks and does not allocate. It reports false when the module is not | |
| 83 | + | // interactive, the line is longer than 127 bytes, or 32 lines are already | |
| 84 | + | // waiting (the script is not reading). | |
| 85 | + | func (m *Module) Send(line []byte) bool { | |
| 86 | + | if !m.Interactive || len(line) > maxLine { | |
| 87 | + | return false | |
| 88 | + | } | |
| 89 | + | var ev event | |
| 90 | + | n := copy(ev.b[:], line) | |
| 91 | + | ev.b[n] = '\n' | |
| 92 | + | ev.n = uint8(n + 1) | |
| 93 | + | select { | |
| 94 | + | case m.input <- ev: | |
| 95 | + | return true | |
| 96 | + | default: | |
| 97 | + | return false | |
| 98 | + | } | |
| 65 | 99 | } | |
| 66 | 100 | ||
| 67 | 101 | // Take copies the pending frame into dst (reusing its capacity) and clears the | |
| @@ -162,10 +196,31 @@ func (m *Module) runOnce(ctx context.Context) error { | |||
| 162 | 196 | if err != nil { | |
| 163 | 197 | return err | |
| 164 | 198 | } | |
| 199 | + | var in io.WriteCloser | |
| 200 | + | if m.Interactive { | |
| 201 | + | if in, err = cmd.StdinPipe(); err != nil { | |
| 202 | + | return err | |
| 203 | + | } | |
| 204 | + | // Events queued while no script was running are stale. | |
| 205 | + | for drained := false; !drained; { | |
| 206 | + | select { | |
| 207 | + | case <-m.input: | |
| 208 | + | default: | |
| 209 | + | drained = true | |
| 210 | + | } | |
| 211 | + | } | |
| 212 | + | } | |
| 165 | 213 | if err = cmd.Start(); err != nil { | |
| 166 | 214 | return err | |
| 167 | 215 | } | |
| 216 | + | var stopWriter func() | |
| 217 | + | if in != nil { | |
| 218 | + | stopWriter = m.pumpInput(in) | |
| 219 | + | } | |
| 168 | 220 | m.readFrames(out) | |
| 221 | + | if stopWriter != nil { | |
| 222 | + | stopWriter() | |
| 223 | + | } | |
| 169 | 224 | err = cmd.Wait() | |
| 170 | 225 | // The script may have started children that kept running (a "while" loop | |
| 171 | 226 | // in a subshell, a "sleep"). They share the process group: end them too. | |
| @@ -173,6 +228,32 @@ func (m *Module) runOnce(ctx context.Context) error { | |||
| 173 | 228 | return err | |
| 174 | 229 | } | |
| 175 | 230 | ||
| 231 | + | // pumpInput writes queued events to w until the returned stop function is | |
| 232 | + | // called or a write fails. stop waits for the writer and closes w. | |
| 233 | + | func (m *Module) pumpInput(w io.WriteCloser) (stop func()) { | |
| 234 | + | done := make(chan struct{}) | |
| 235 | + | exited := make(chan struct{}) | |
| 236 | + | go func() { | |
| 237 | + | defer close(exited) | |
| 238 | + | for { | |
| 239 | + | select { | |
| 240 | + | case ev := <-m.input: | |
| 241 | + | if _, err := w.Write(ev.b[:ev.n]); err != nil { | |
| 242 | + | return | |
| 243 | + | } | |
| 244 | + | case <-done: | |
| 245 | + | return | |
| 246 | + | } | |
| 247 | + | } | |
| 248 | + | }() | |
| 249 | + | return func() { | |
| 250 | + | close(done) | |
| 251 | + | // A write blocked on a script that does not read ends when w closes. | |
| 252 | + | _ = w.Close() | |
| 253 | + | <-exited | |
| 254 | + | } | |
| 255 | + | } | |
| 256 | + | ||
| 176 | 257 | // readFrames splits r into frames and publishes each one. | |
| 177 | 258 | func (m *Module) readFrames(r io.Reader) { | |
| 178 | 259 | buf := make([]byte, readSize) | |
| @@ -46,22 +46,56 @@ type Module struct { | |||||
| 46 | Exec string | 46 | Exec string | ||
| 47 | // Env is added to the script's environment, as KEY=value entries. | 47 | // Env is added to the script's environment, as KEY=value entries. | ||
| 48 | Env []string | 48 | Env []string | ||
| 49 | + | // Interactive gives the script a stdin that carries Send lines. Set it | |||
| 50 | + | // before Start. | |||
| 51 | + | Interactive bool | |||
| 49 | 52 | ||||
| 50 | mu sync.Mutex | 53 | mu sync.Mutex | ||
| 51 | pending []byte // latest complete frame | 54 | pending []byte // latest complete frame | ||
| 52 | dirty bool | 55 | dirty bool | ||
| 53 | failed bool // the script is not running; show an error marker | 56 | failed bool // the script is not running; show an error marker | ||
| 54 | 57 | ||||
| 58 | + | input chan event // lines for the script's stdin | |||
| 59 | + | ||||
| 55 | wake chan<- struct{} | 60 | wake chan<- struct{} | ||
| 56 | log *slog.Logger | 61 | log *slog.Logger | ||
| 57 | 62 | ||||
| 58 | done sync.WaitGroup // Start adds one; Run releases it when its process is gone | 63 | done sync.WaitGroup // Start adds one; Run releases it when its process is gone | ||
| 59 | } | 64 | } | ||
| 60 | 65 | ||||
| 66 | + | // maxLine is the longest line Send accepts, without its newline. | |||
| 67 | + | const maxLine = 127 | |||
| 68 | + | ||||
| 69 | + | // event is one stdin line, newline included. | |||
| 70 | + | type event struct { | |||
| 71 | + | b [maxLine + 1]byte | |||
| 72 | + | n uint8 | |||
| 73 | + | } | |||
| 74 | + | ||||
| 61 | // New creates a module. wake receives a non-blocking signal after each frame. | 75 | // New creates a module. wake receives a non-blocking signal after each frame. | ||
| 62 | func New(name string, zone Zone, command string, wake chan<- struct{}, log *slog.Logger) *Module { | 76 | func New(name string, zone Zone, command string, wake chan<- struct{}, log *slog.Logger) *Module { | ||
| 63 | return &Module{Name: name, Zone: zone, Exec: command, wake: wake, log: log, | 77 | return &Module{Name: name, Zone: zone, Exec: command, wake: wake, log: log, | ||
| 64 | - | pending: make([]byte, 0, MaxFrame)} | 78 | + | pending: make([]byte, 0, MaxFrame), input: make(chan event, 32)} |
| 79 | + | } | |||
| 80 | + | ||||
| 81 | + | // Send queues line (without a newline) for the script's stdin. It never | |||
| 82 | + | // blocks and does not allocate. It reports false when the module is not | |||
| 83 | + | // interactive, the line is longer than 127 bytes, or 32 lines are already | |||
| 84 | + | // waiting (the script is not reading). | |||
| 85 | + | func (m *Module) Send(line []byte) bool { | |||
| 86 | + | if !m.Interactive || len(line) > maxLine { | |||
| 87 | + | return false | |||
| 88 | + | } | |||
| 89 | + | var ev event | |||
| 90 | + | n := copy(ev.b[:], line) | |||
| 91 | + | ev.b[n] = '\n' | |||
| 92 | + | ev.n = uint8(n + 1) | |||
| 93 | + | select { | |||
| 94 | + | case m.input <- ev: | |||
| 95 | + | return true | |||
| 96 | + | default: | |||
| 97 | + | return false | |||
| 98 | + | } | |||
| 65 | } | 99 | } | ||
| 66 | 100 | ||||
| 67 | // Take copies the pending frame into dst (reusing its capacity) and clears the | 101 | // Take copies the pending frame into dst (reusing its capacity) and clears the | ||
| @@ -162,10 +196,31 @@ func (m *Module) runOnce(ctx context.Context) error { | |||||
| 162 | if err != nil { | 196 | if err != nil { | ||
| 163 | return err | 197 | return err | ||
| 164 | } | 198 | } | ||
| 199 | + | var in io.WriteCloser | |||
| 200 | + | if m.Interactive { | |||
| 201 | + | if in, err = cmd.StdinPipe(); err != nil { | |||
| 202 | + | return err | |||
| 203 | + | } | |||
| 204 | + | // Events queued while no script was running are stale. | |||
| 205 | + | for drained := false; !drained; { | |||
| 206 | + | select { | |||
| 207 | + | case <-m.input: | |||
| 208 | + | default: | |||
| 209 | + | drained = true | |||
| 210 | + | } | |||
| 211 | + | } | |||
| 212 | + | } | |||
| 165 | if err = cmd.Start(); err != nil { | 213 | if err = cmd.Start(); err != nil { | ||
| 166 | return err | 214 | return err | ||
| 167 | } | 215 | } | ||
| 216 | + | var stopWriter func() | |||
| 217 | + | if in != nil { | |||
| 218 | + | stopWriter = m.pumpInput(in) | |||
| 219 | + | } | |||
| 168 | m.readFrames(out) | 220 | m.readFrames(out) | ||
| 221 | + | if stopWriter != nil { | |||
| 222 | + | stopWriter() | |||
| 223 | + | } | |||
| 169 | err = cmd.Wait() | 224 | err = cmd.Wait() | ||
| 170 | // The script may have started children that kept running (a "while" loop | 225 | // The script may have started children that kept running (a "while" loop | ||
| 171 | // in a subshell, a "sleep"). They share the process group: end them too. | 226 | // in a subshell, a "sleep"). They share the process group: end them too. | ||
| @@ -173,6 +228,32 @@ func (m *Module) runOnce(ctx context.Context) error { | |||||
| 173 | return err | 228 | return err | ||
| 174 | } | 229 | } | ||
| 175 | 230 | ||||
| 231 | + | // pumpInput writes queued events to w until the returned stop function is | |||
| 232 | + | // called or a write fails. stop waits for the writer and closes w. | |||
| 233 | + | func (m *Module) pumpInput(w io.WriteCloser) (stop func()) { | |||
| 234 | + | done := make(chan struct{}) | |||
| 235 | + | exited := make(chan struct{}) | |||
| 236 | + | go func() { | |||
| 237 | + | defer close(exited) | |||
| 238 | + | for { | |||
| 239 | + | select { | |||
| 240 | + | case ev := <-m.input: | |||
| 241 | + | if _, err := w.Write(ev.b[:ev.n]); err != nil { | |||
| 242 | + | return | |||
| 243 | + | } | |||
| 244 | + | case <-done: | |||
| 245 | + | return | |||
| 246 | + | } | |||
| 247 | + | } | |||
| 248 | + | }() | |||
| 249 | + | return func() { | |||
| 250 | + | close(done) | |||
| 251 | + | // A write blocked on a script that does not read ends when w closes. | |||
| 252 | + | _ = w.Close() | |||
| 253 | + | <-exited | |||
| 254 | + | } | |||
| 255 | + | } | |||
| 256 | + | ||||
| 176 | // readFrames splits r into frames and publishes each one. | 257 | // readFrames splits r into frames and publishes each one. | ||
| 177 | func (m *Module) readFrames(r io.Reader) { | 258 | func (m *Module) readFrames(r io.Reader) { | ||
| 178 | buf := make([]byte, readSize) | 259 | buf := make([]byte, readSize) | ||
M internal/module/module_test.go +95 -0
| @@ -131,3 +131,98 @@ func TestScriptDir(t *testing.T) { | |||
| 131 | 131 | } | |
| 132 | 132 | } | |
| 133 | 133 | } | |
| 134 | + | ||
| 135 | + | func TestSendReachesTheScript(t *testing.T) { | |
| 136 | + | wake := make(chan struct{}, 1) | |
| 137 | + | m := New("t", Left, `while read l; do printf '%s\n' "$l"; done`, wake, slog.New(slog.DiscardHandler)) | |
| 138 | + | m.Interactive = true | |
| 139 | + | ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second) | |
| 140 | + | defer cancel() | |
| 141 | + | m.Start(ctx) | |
| 142 | + | defer m.Wait(5 * time.Second) | |
| 143 | + | defer cancel() | |
| 144 | + | deadline := time.After(5 * time.Second) | |
| 145 | + | for { | |
| 146 | + | // The script may start after the first Send: resend until it answers. | |
| 147 | + | m.Send([]byte("hover 3")) | |
| 148 | + | select { | |
| 149 | + | case <-wake: | |
| 150 | + | if b, ch, failed := m.Take(nil); ch && !failed && string(b) == "hover 3" { | |
| 151 | + | return | |
| 152 | + | } | |
| 153 | + | case <-time.After(50 * time.Millisecond): | |
| 154 | + | case <-deadline: | |
| 155 | + | t.Fatal("the script never echoed the line") | |
| 156 | + | } | |
| 157 | + | } | |
| 158 | + | } | |
| 159 | + | ||
| 160 | + | func TestSendRefusals(t *testing.T) { | |
| 161 | + | m, _ := newTest() | |
| 162 | + | if m.Send([]byte("hover 1")) { | |
| 163 | + | t.Fatal("a non-interactive module must refuse Send") | |
| 164 | + | } | |
| 165 | + | m.Interactive = true | |
| 166 | + | if m.Send(make([]byte, 128)) { | |
| 167 | + | t.Fatal("a line over 127 bytes must be refused") | |
| 168 | + | } | |
| 169 | + | if !m.Send(make([]byte, 127)) { | |
| 170 | + | t.Fatal("a 127-byte line must be accepted") | |
| 171 | + | } | |
| 172 | + | } | |
| 173 | + | ||
| 174 | + | func TestSendNeverBlocks(t *testing.T) { | |
| 175 | + | m, _ := newTest() | |
| 176 | + | m.Interactive = true | |
| 177 | + | done := make(chan int) | |
| 178 | + | go func() { | |
| 179 | + | ok := 0 | |
| 180 | + | for i := 0; i < 33; i++ { | |
| 181 | + | if m.Send([]byte("hover 1")) { | |
| 182 | + | ok++ | |
| 183 | + | } | |
| 184 | + | } | |
| 185 | + | done <- ok | |
| 186 | + | }() | |
| 187 | + | select { | |
| 188 | + | case ok := <-done: | |
| 189 | + | if ok != 32 { | |
| 190 | + | t.Fatalf("%d sends accepted, want 32 (the queue size)", ok) | |
| 191 | + | } | |
| 192 | + | case <-time.After(2 * time.Second): | |
| 193 | + | t.Fatal("Send blocked on a full queue") | |
| 194 | + | } | |
| 195 | + | } | |
| 196 | + | ||
| 197 | + | // A script that never reads its stdin must not hold the module back. | |
| 198 | + | func TestUnreadStdinDoesNotBlockStop(t *testing.T) { | |
| 199 | + | m := New("t", Left, "echo up; sleep 30", make(chan struct{}, 1), slog.New(slog.DiscardHandler)) | |
| 200 | + | m.Interactive = true | |
| 201 | + | ctx, cancel := context.WithCancel(context.Background()) | |
| 202 | + | m.Start(ctx) | |
| 203 | + | for i := 0; i < 40; i++ { | |
| 204 | + | m.Send([]byte(strings.Repeat("x", 100))) | |
| 205 | + | time.Sleep(5 * time.Millisecond) | |
| 206 | + | } | |
| 207 | + | cancel() | |
| 208 | + | start := time.Now() | |
| 209 | + | m.Wait(5 * time.Second) | |
| 210 | + | if time.Since(start) > 4*time.Second { | |
| 211 | + | t.Fatal("stopping took too long") | |
| 212 | + | } | |
| 213 | + | } | |
| 214 | + | ||
| 215 | + | func TestSendAllocs(t *testing.T) { | |
| 216 | + | if racecheck.Enabled { | |
| 217 | + | t.Skip("the race detector allocates") | |
| 218 | + | } | |
| 219 | + | m, _ := newTest() | |
| 220 | + | m.Interactive = true | |
| 221 | + | line := []byte("hover 12") | |
| 222 | + | if got := testing.AllocsPerRun(100, func() { | |
| 223 | + | m.Send(line) | |
| 224 | + | <-m.input | |
| 225 | + | }); got > 0 { | |
| 226 | + | t.Errorf("Send allocates %.1f objects; want 0", got) | |
| 227 | + | } | |
| 228 | + | } | |
| @@ -131,3 +131,98 @@ func TestScriptDir(t *testing.T) { | |||||
| 131 | } | 131 | } | ||
| 132 | } | 132 | } | ||
| 133 | } | 133 | } | ||
| 134 | + | ||||
| 135 | + | func TestSendReachesTheScript(t *testing.T) { | |||
| 136 | + | wake := make(chan struct{}, 1) | |||
| 137 | + | m := New("t", Left, `while read l; do printf '%s\n' "$l"; done`, wake, slog.New(slog.DiscardHandler)) | |||
| 138 | + | m.Interactive = true | |||
| 139 | + | ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second) | |||
| 140 | + | defer cancel() | |||
| 141 | + | m.Start(ctx) | |||
| 142 | + | defer m.Wait(5 * time.Second) | |||
| 143 | + | defer cancel() | |||
| 144 | + | deadline := time.After(5 * time.Second) | |||
| 145 | + | for { | |||
| 146 | + | // The script may start after the first Send: resend until it answers. | |||
| 147 | + | m.Send([]byte("hover 3")) | |||
| 148 | + | select { | |||
| 149 | + | case <-wake: | |||
| 150 | + | if b, ch, failed := m.Take(nil); ch && !failed && string(b) == "hover 3" { | |||
| 151 | + | return | |||
| 152 | + | } | |||
| 153 | + | case <-time.After(50 * time.Millisecond): | |||
| 154 | + | case <-deadline: | |||
| 155 | + | t.Fatal("the script never echoed the line") | |||
| 156 | + | } | |||
| 157 | + | } | |||
| 158 | + | } | |||
| 159 | + | ||||
| 160 | + | func TestSendRefusals(t *testing.T) { | |||
| 161 | + | m, _ := newTest() | |||
| 162 | + | if m.Send([]byte("hover 1")) { | |||
| 163 | + | t.Fatal("a non-interactive module must refuse Send") | |||
| 164 | + | } | |||
| 165 | + | m.Interactive = true | |||
| 166 | + | if m.Send(make([]byte, 128)) { | |||
| 167 | + | t.Fatal("a line over 127 bytes must be refused") | |||
| 168 | + | } | |||
| 169 | + | if !m.Send(make([]byte, 127)) { | |||
| 170 | + | t.Fatal("a 127-byte line must be accepted") | |||
| 171 | + | } | |||
| 172 | + | } | |||
| 173 | + | ||||
| 174 | + | func TestSendNeverBlocks(t *testing.T) { | |||
| 175 | + | m, _ := newTest() | |||
| 176 | + | m.Interactive = true | |||
| 177 | + | done := make(chan int) | |||
| 178 | + | go func() { | |||
| 179 | + | ok := 0 | |||
| 180 | + | for i := 0; i < 33; i++ { | |||
| 181 | + | if m.Send([]byte("hover 1")) { | |||
| 182 | + | ok++ | |||
| 183 | + | } | |||
| 184 | + | } | |||
| 185 | + | done <- ok | |||
| 186 | + | }() | |||
| 187 | + | select { | |||
| 188 | + | case ok := <-done: | |||
| 189 | + | if ok != 32 { | |||
| 190 | + | t.Fatalf("%d sends accepted, want 32 (the queue size)", ok) | |||
| 191 | + | } | |||
| 192 | + | case <-time.After(2 * time.Second): | |||
| 193 | + | t.Fatal("Send blocked on a full queue") | |||
| 194 | + | } | |||
| 195 | + | } | |||
| 196 | + | ||||
| 197 | + | // A script that never reads its stdin must not hold the module back. | |||
| 198 | + | func TestUnreadStdinDoesNotBlockStop(t *testing.T) { | |||
| 199 | + | m := New("t", Left, "echo up; sleep 30", make(chan struct{}, 1), slog.New(slog.DiscardHandler)) | |||
| 200 | + | m.Interactive = true | |||
| 201 | + | ctx, cancel := context.WithCancel(context.Background()) | |||
| 202 | + | m.Start(ctx) | |||
| 203 | + | for i := 0; i < 40; i++ { | |||
| 204 | + | m.Send([]byte(strings.Repeat("x", 100))) | |||
| 205 | + | time.Sleep(5 * time.Millisecond) | |||
| 206 | + | } | |||
| 207 | + | cancel() | |||
| 208 | + | start := time.Now() | |||
| 209 | + | m.Wait(5 * time.Second) | |||
| 210 | + | if time.Since(start) > 4*time.Second { | |||
| 211 | + | t.Fatal("stopping took too long") | |||
| 212 | + | } | |||
| 213 | + | } | |||
| 214 | + | ||||
| 215 | + | func TestSendAllocs(t *testing.T) { | |||
| 216 | + | if racecheck.Enabled { | |||
| 217 | + | t.Skip("the race detector allocates") | |||
| 218 | + | } | |||
| 219 | + | m, _ := newTest() | |||
| 220 | + | m.Interactive = true | |||
| 221 | + | line := []byte("hover 12") | |||
| 222 | + | if got := testing.AllocsPerRun(100, func() { | |||
| 223 | + | m.Send(line) | |||
| 224 | + | <-m.input | |||
| 225 | + | }); got > 0 { | |||
| 226 | + | t.Errorf("Send allocates %.1f objects; want 0", got) | |||
| 227 | + | } | |||
| 228 | + | } | |||