Shells in OCaml
Process Groups
A major overhaul of the way that we execute pipelines. Here, we make
sure to add all of the commands to a relevant process group which will
be important for implementing asynchronous jobs later.
+206
-79
-9
src/lib/eunix.ml
-9
src/lib/eunix.ml
···
1
1
let cwd () = Eio_unix.run_in_systhread ~label:"cwd" @@ fun () -> Unix.getcwd ()
2
2
3
-
let chdir p =
4
-
try
5
-
let dir =
6
-
Eio_unix.run_in_systhread ~label:"chdir" @@ fun () -> Unix.chdir p
7
-
in
8
-
Exit.zero dir
9
-
with Unix.Unix_error (Unix.ENOENT, _, _) ->
10
-
Exit.nonzero_msg () "no such file or directory: %s" p
11
-
12
3
let env () =
13
4
Eio_unix.run_in_systhread ~label:"env" @@ fun () ->
14
5
Unix.environment ()
+107
-53
src/lib/eval.ml
+107
-53
src/lib/eval.ml
···
6
6
open Exit.Syntax
7
7
8
8
module Options = struct
9
-
type t = { noclobber : bool }
9
+
type t = { noclobber : bool; pipefail : bool }
10
10
11
-
let default = { noclobber = false }
11
+
let default = { noclobber = false; pipefail = false }
12
12
13
-
let with_options ?noclobber t =
14
-
{ noclobber = Option.value ~default:t.noclobber noclobber }
13
+
let with_options ?noclobber ?pipefail t =
14
+
{
15
+
noclobber = Option.value ~default:t.noclobber noclobber;
16
+
pipefail = Option.value ~default:t.pipefail pipefail;
17
+
}
15
18
end
16
19
17
20
(** An evaluator over the AST *)
···
21
24
It starts from point (4), completing a series of expansions on the AST,
22
25
then redirection is setup, and finally functions/built-ins/commands are
23
26
executed. *)
27
+
28
+
module J = Job.Make (E)
24
29
25
30
class default_map =
26
31
object (_)
···
172
177
match path with
173
178
| Some p ->
174
179
let fp = Fpath.append cwd (Fpath.v p) in
175
-
Exit.map' (Eunix.chdir p)
176
-
~zero:(fun () -> S.set_cwd ctx.state fp)
177
-
~nonzero:(fun () -> ctx.state)
180
+
Exit.zero @@ S.set_cwd ctx.state fp
178
181
| None -> (
179
182
match Eunix.find_env "HOME" with
180
183
| None -> Exit.nonzero_msg ctx.state "HOME not set"
···
182
185
in
183
186
{ ctx with state }
184
187
| Pwd ->
185
-
Fmt.pr "%s\n%!" (Eunix.cwd ());
188
+
Fmt.pr "%a\n%!" Fpath.pp (S.cwd ctx.state);
186
189
Exit.zero ctx
187
190
| Exit n ->
188
191
let should_exit =
···
204
207
List.fold_left (fun acc (k, _) -> List.remove_assoc k acc) env extra
205
208
|> List.append extra
206
209
207
-
let rec execute_commands initial_ctx local_switch p =
208
-
let rec loop (exit_ctx : ctx Exit.t)
209
-
(stdout_of_previous : Eio_unix.source_ty Eio_unix.source option) :
210
-
Ast.command list -> ctx Exit.t =
210
+
let rec execute_commands initial_ctx pipeline_switch p : ctx Exit.t =
211
+
let handle_job ~pgid j p =
212
+
match (j, p) with
213
+
| None, _ ->
214
+
Option.some
215
+
@@ J.make ~state:`Running ~bang:false pgid (Nlist.Singleton p)
216
+
| Some j, `Process p -> Option.some @@ J.add_process p j
217
+
| Some j, `Built_in p -> Option.some @@ J.add_built_in p j
218
+
in
219
+
let rec loop (ctx : ctx) (job : J.t option)
220
+
((pgid, stdout_of_previous) :
221
+
int * Eio_unix.source_ty Eio_unix.source option) :
222
+
Ast.command list -> ctx * J.t option =
211
223
fun c ->
212
-
let ctx = Exit.value exit_ctx in
213
224
match c with
214
225
| Ast.SimpleCommand (Prefixed (prefix, None, _suffix)) :: rest ->
215
226
let ctx = collect_assignments ctx prefix in
216
-
loop (Exit.zero ctx) stdout_of_previous rest
227
+
loop ctx job (pgid, stdout_of_previous) rest
217
228
| Ast.SimpleCommand (Prefixed (prefix, Some executable, suffix)) :: rest
218
229
->
219
230
let ctx = collect_assignments ~update:false ctx prefix in
220
-
loop (Exit.zero ctx) stdout_of_previous
231
+
loop ctx job (pgid, stdout_of_previous)
221
232
(Ast.SimpleCommand (Named (executable, suffix)) :: rest)
222
233
| Ast.SimpleCommand (Named (executable, None)) :: rest -> (
223
234
let ctx, executable = expand_cst ctx executable in
···
225
236
Built_ins.of_args
226
237
[ handle_word_components_to_string ctx executable ]
227
238
with
228
-
| Some bi -> handle_built_in ctx bi
239
+
| Some bi ->
240
+
let ctx = handle_built_in ctx bi in
241
+
let built_in = ctx >|= fun _ -> () in
242
+
(Exit.value ctx, handle_job ~pgid job (`Built_in built_in))
229
243
| None -> (
230
244
let some_read, some_write =
231
-
stdout_for_pipeline ctx ~sw:local_switch rest
245
+
stdout_for_pipeline ctx ~sw:pipeline_switch rest
232
246
in
233
247
match stdout_of_previous with
234
248
| None ->
235
249
let executable =
236
250
handle_word_components_to_string ctx executable
237
251
in
238
-
let res =
239
-
E.exec ctx.executor ?stdout:some_write ~cwd:(cwd_of_ctx ctx)
240
-
~env:(get_env ~extra:ctx.local_state ())
241
-
[ executable ]
242
-
>|= fun () -> clear_local_state ctx
252
+
let ctx, job =
253
+
let process =
254
+
E.exec ctx.executor ?stdout:some_write ~pgid
255
+
~sw:pipeline_switch ~cwd:(cwd_of_ctx ctx)
256
+
~env:(get_env ~extra:ctx.local_state ())
257
+
[ executable ]
258
+
in
259
+
let job = handle_job ~pgid job (`Process process) in
260
+
(clear_local_state ctx, job)
243
261
in
244
262
Option.iter Eio.Flow.close some_write;
245
-
loop res some_read rest
263
+
loop ctx job (pgid, some_read) rest
246
264
| Some stdout ->
247
265
let executable =
248
266
handle_word_components_to_string ctx executable
249
267
in
250
-
let res =
251
-
E.exec ctx.executor ~stdin:stdout ?stdout:some_write
252
-
~env:(get_env ~extra:ctx.local_state ())
253
-
~cwd:(cwd_of_ctx ctx) [ executable ]
254
-
>|= fun () -> clear_local_state ctx
268
+
let ctx, job =
269
+
let process =
270
+
E.exec ctx.executor ~stdin:stdout ~pgid
271
+
~sw:pipeline_switch ?stdout:some_write
272
+
~env:(get_env ~extra:ctx.local_state ())
273
+
~cwd:(cwd_of_ctx ctx) [ executable ]
274
+
in
275
+
let job = handle_job ~pgid job (`Process process) in
276
+
(clear_local_state ctx, job)
255
277
in
256
278
Option.iter Eio.Flow.close some_write;
257
-
loop res some_read rest))
279
+
loop ctx job (pgid, some_read) rest))
258
280
| Ast.SimpleCommand (Named (executable, Some suffix)) :: rest -> (
259
281
let ctx, executable = expand_cst ctx executable in
260
282
let ctx, suffix = expand_redirects (ctx, []) suffix in
···
263
285
Built_ins.of_args
264
286
(handle_word_components_to_string ctx executable :: args)
265
287
with
266
-
| Some bi -> handle_built_in ctx bi
288
+
| Some bi ->
289
+
let ctx = handle_built_in ctx bi in
290
+
let built_in = ctx >|= fun _ -> () in
291
+
(Exit.value ctx, handle_job ~pgid job (`Built_in built_in))
267
292
| None -> (
268
293
let redirect =
269
294
List.fold_left
270
295
(fun acc -> function
271
296
| Ast.Suffix_word _ -> acc
272
297
| Ast.Suffix_redirect rdr ->
273
-
handle_one_redirection ~sw:local_switch ctx rdr :: acc)
298
+
handle_one_redirection ~sw:pipeline_switch ctx rdr
299
+
:: acc)
274
300
[] suffix
275
301
|> List.rev |> List.filter_map Fun.id
276
302
in
277
303
let some_read, some_write =
278
-
stdout_for_pipeline ~sw:local_switch ctx rest
304
+
stdout_for_pipeline ~sw:pipeline_switch ctx rest
279
305
in
280
306
match stdout_of_previous with
281
307
| None ->
282
-
let res =
283
-
E.exec ~fds:redirect ctx.executor ?stdout:some_write
284
-
~cwd:(cwd_of_ctx ctx)
285
-
~env:(get_env ~extra:ctx.local_state ())
286
-
(handle_word_components_to_string ctx executable :: args)
287
-
>|= fun () -> clear_local_state ctx
308
+
let executable =
309
+
handle_word_components_to_string ctx executable
310
+
in
311
+
let ctx, job =
312
+
let process =
313
+
E.exec ctx.executor ?stdout:some_write ~pgid
314
+
~sw:pipeline_switch ~fds:redirect ~cwd:(cwd_of_ctx ctx)
315
+
~env:(get_env ~extra:ctx.local_state ())
316
+
(executable :: args)
317
+
in
318
+
let job = handle_job ~pgid job (`Process process) in
319
+
(clear_local_state ctx, job)
288
320
in
289
321
Option.iter Eio.Flow.close some_write;
290
-
loop res some_read rest
322
+
loop ctx job (pgid, some_read) rest
291
323
| Some stdout ->
292
-
let res =
293
-
E.exec ~fds:redirect ctx.executor ~stdin:stdout
294
-
~cwd:(cwd_of_ctx ctx) ?stdout:some_write
295
-
~env:(get_env ~extra:ctx.local_state ())
296
-
(handle_word_components_to_string ctx executable :: args)
297
-
>|= fun () -> clear_local_state ctx
324
+
let executable =
325
+
handle_word_components_to_string ctx executable
326
+
in
327
+
let ctx, job =
328
+
let process =
329
+
E.exec ctx.executor ~stdin:stdout ~pgid
330
+
~sw:pipeline_switch ?stdout:some_write ~fds:redirect
331
+
~env:(get_env ~extra:ctx.local_state ())
332
+
~cwd:(cwd_of_ctx ctx) (executable :: args)
333
+
in
334
+
let job = handle_job ~pgid job (`Process process) in
335
+
(clear_local_state ctx, job)
298
336
in
299
337
Option.iter Eio.Flow.close some_write;
300
-
loop res some_read rest))
338
+
loop ctx job (pgid, some_read) rest))
301
339
| CompoundCommand (c, rdrs) :: _rest ->
302
340
let _rdrs =
303
-
List.map (handle_one_redirection ~sw:local_switch ctx) rdrs
341
+
List.map (handle_one_redirection ~sw:pipeline_switch ctx) rdrs
304
342
in
305
-
let ctx = handle_compound_command ctx c in
306
-
ctx
343
+
(* TODO: No way this is right *)
344
+
(Exit.value @@ handle_compound_command ctx c, job)
307
345
| v :: _ ->
308
346
Fmt.epr "TODO: %a" Yojson.Safe.pp (Ast.command_to_yojson v);
309
347
failwith "Err"
310
-
| [] -> exit_ctx
348
+
| [] -> (ctx, job)
311
349
in
312
-
loop (Exit.zero initial_ctx) None p
350
+
(* HACK: when running the pipeline, we need a process group to
351
+
put everything in. Eio's model of execution is nice, but we cannot
352
+
safely delay execution of a process. So instead we create a ghost
353
+
process that last just until all of the processes are setup. *)
354
+
let ctx, job =
355
+
Eio.Switch.run @@ fun ghost_switch ->
356
+
let ghost_process =
357
+
E.exec ~sw:ghost_switch ~pgid:0 ~cwd:(cwd_of_ctx initial_ctx)
358
+
initial_ctx.executor [ "sleep"; "99999999" ]
359
+
in
360
+
loop initial_ctx None (E.pid ghost_process, None) p
361
+
in
362
+
match job with
363
+
| None -> Exit.zero ctx
364
+
| Some job -> J.await_exit ~pipefail:false job >|= fun () -> ctx
313
365
314
366
and expand_cst (ctx : ctx) cst =
315
367
let cst = tilde_expansion ctx cst in
···
372
424
in
373
425
fold (Noand_or, Exit.zero ctx) c
374
426
375
-
and handle_for_clause ctx = function
427
+
and handle_for_clause ctx v : ctx Exit.t =
428
+
match v with
376
429
| Ast.For_Name_DoGroup (_, (term, sep)) -> exec ctx (term, Some sep)
377
430
| Ast.For_Name_In_WordList_DoGroup (Name name, wdlist, (term, sep)) ->
378
431
let wdlist = Nlist.flatten @@ Nlist.map (word_glob_expand ctx) wdlist in
···
408
461
| Exit.Zero ctx -> exec ctx (e2, Some sep2)
409
462
| Exit.Nonzero { value = ctx; _ } -> handle_else_part ctx else_part)
410
463
411
-
and handle_compound_command ctx = function
464
+
and handle_compound_command ctx v : ctx Exit.t =
465
+
match v with
412
466
| Ast.ForClause fc -> handle_for_clause ctx fc
413
467
| Ast.IfClause if_ -> handle_if_clause ctx if_
414
468
| _ as c ->
+6
src/lib/import.ml
+6
src/lib/import.ml
···
3
3
module Nlist = struct
4
4
type 'a t = Singleton of 'a | Cons of 'a * 'a t
5
5
6
+
let hd = function Singleton s -> s | Cons (s, _) -> s
7
+
8
+
let rec length = function
9
+
| Singleton _ -> 1
10
+
| Cons (_, rest) -> 1 + length rest
11
+
6
12
let rec of_list = function
7
13
| [] -> invalid_arg "Empty list"
8
14
| [ x ] -> Singleton x
+28
src/lib/job.ml
+28
src/lib/job.ml
···
1
+
open Import
2
+
3
+
module Make (E : Types.Exec) = struct
4
+
type t = {
5
+
state : [ `Running ];
6
+
id : int;
7
+
bang : bool;
8
+
(* Process list is in reverse order *)
9
+
processes : [ `Process of E.process | `Built_in of unit Exit.t ] Nlist.t;
10
+
}
11
+
12
+
let make ?(state = `Running) ~bang id processes =
13
+
{ state; id; processes; bang }
14
+
15
+
let add_process proc t =
16
+
{ t with processes = Nlist.cons (`Process proc) t.processes }
17
+
18
+
let add_built_in b t =
19
+
{ t with processes = Nlist.cons (`Built_in b) t.processes }
20
+
21
+
(* Section 2.9.2 https://pubs.opengroup.org/onlinepubs/9799919799/ *)
22
+
let await_exit ~pipefail t =
23
+
let await = function `Process p -> E.await p | `Built_in b -> b in
24
+
match (pipefail, t.bang) with
25
+
| false, false -> await (Nlist.hd t.processes)
26
+
| false, true -> await (Nlist.hd t.processes) |> Exit.not
27
+
| _ -> Fmt.failwith "TODO: pipefail"
28
+
end
+6
-3
src/lib/posix/exec.ml
+6
-3
src/lib/posix/exec.ml
···
91
91
Eio_unix.Private.Fork_action.
92
92
{ run = (fun k -> k (Obj.repr (action_dups, plan, blocking))) }
93
93
94
-
let spawn_unix () ~sw ?pgid ?uid ?gid ~env ~fds ~executable ~cwd args =
94
+
let spawn_unix () ~sw ~fork_actions ?pgid ?uid ?gid ~env ~fds ~executable ~cwd
95
+
args =
95
96
let open Eio_posix in
96
97
let actions =
97
98
[
···
115
116
| None -> actions
116
117
| Some gid -> Eio_unix.Private.Fork_action.setgid gid :: actions
117
118
in
119
+
let actions = actions @ fork_actions in
118
120
let with_actions cwd fn =
119
121
let ((dir, path) : Eio.Fs.dir_ty Eio.Path.t) = cwd in
120
122
match Eio_posix__.Fs.as_posix_dir dir with
···
140
142
141
143
let pp_redirections ppf (i, fd, _) = Fmt.pf ppf "(%i,%a)" i Eio_unix.Fd.pp fd
142
144
143
-
let run ~sw _ ?stdin ?stdout ?stderr ?(fds = []) ~cwd ?env ?executable args =
145
+
let run ~sw _ ?stdin ?stdout ?stderr ?(fds = []) ?(fork_actions = []) ~pgid ~cwd
146
+
?env ?executable args =
144
147
with_close_list @@ fun to_close ->
145
148
let check_fd n = function
146
149
| Merry.Types.Redirect (m, _, _) -> Int.equal n m
···
183
186
let fds = std_fds @ fds in
184
187
let executable = get_executable executable ~args in
185
188
let env = get_env env in
186
-
spawn_unix ~sw ~cwd ~fds ~env ~executable () args
189
+
spawn_unix ~sw ~fork_actions ~cwd ~pgid ~fds ~env ~executable () args
+14
-9
src/lib/posix/merry_posix.ml
+14
-9
src/lib/posix/merry_posix.ml
···
4
4
5
5
module Exec = struct
6
6
type t = { mgr : Eio_unix.Process.mgr_ty Eio_unix.Process.mgr }
7
-
type fork_action = unit
7
+
type process = Eio_unix.Process.ty Eio_unix.Process.t
8
8
9
-
let exec ?fork_actions:_ ?(fds = []) ?stdin ?stdout ?stderr ?env ~cwd t args =
10
-
Eio.Switch.run @@ fun sw ->
9
+
let pid = Eio.Process.pid
10
+
let signal v i = Eio.Process.signal v i
11
+
12
+
let await v =
13
+
Eio.Process.await v |> function
14
+
| `Exited 0 -> Merry.Exit.zero ()
15
+
| `Exited n -> Merry.Exit.nonzero () n
16
+
| `Signaled n -> Merry.Exit.nonzero () n
17
+
18
+
let exec ?(fork_actions = []) ?(fds = []) ?stdin ?stdout ?stderr ?env ~sw
19
+
~pgid ~cwd t args : process =
11
20
let env =
12
21
Option.map
13
22
(fun lst -> List.map (fun (a, b) -> a ^ "=" ^ b) lst |> Array.of_list)
14
23
env
15
24
in
16
-
Exec.run ~sw ~fds ~cwd ?stdin ?stdout ?stderr ?env t args
17
-
|> Eio.Process.await
18
-
|> function
19
-
| `Exited 0 -> Merry.Exit.zero ()
20
-
| `Exited n -> Merry.Exit.nonzero () n
21
-
| `Signaled n -> Merry.Exit.nonzero () n
25
+
Exec.run ~fork_actions ~sw ~fds ~pgid ~cwd ?stdin ?stdout ?stderr ?env t
26
+
args
22
27
end
+10
-5
src/lib/types.ml
+10
-5
src/lib/types.ml
···
37
37
type t
38
38
(** An executor for commands *)
39
39
40
-
type fork_action
41
-
(** A fork action is a piece of C-code to run inbetween the fork and the exec
42
-
*)
40
+
type process
41
+
42
+
val signal : process -> int -> unit
43
+
val pid : process -> int
43
44
44
45
val exec :
45
-
?fork_actions:fork_action list ->
46
+
?fork_actions:Eio_unix__.Fork_action.t list ->
46
47
?fds:redirect list ->
47
48
?stdin:_ Eio.Flow.source ->
48
49
?stdout:_ Eio.Flow.sink ->
49
50
?stderr:_ Eio.Flow.sink ->
50
51
?env:(string * string) list ->
52
+
sw:Eio.Switch.t ->
53
+
pgid:int ->
51
54
cwd:Eio.Fs.dir_ty Eio.Path.t ->
52
55
t ->
53
56
string list ->
54
-
unit Exit.t
57
+
process
55
58
(** Run a command in a child process *)
59
+
60
+
val await : process -> unit Exit.t
56
61
end
+34
test/pipelines.t
+34
test/pipelines.t
···
1
+
Some more tricky parts of pipelines.
2
+
3
+
Under normal execution, only the very last command matters in terms of exit code!
4
+
5
+
$ mkdir hello
6
+
$ sh -c "ls -j"
7
+
ls: invalid option -- 'j'
8
+
Try 'ls --help' for more information.
9
+
[2]
10
+
$ sh -c "ls -j | ls"
11
+
ls: invalid option -- 'j'
12
+
Try 'ls --help' for more information.
13
+
hello
14
+
$ osh -c "ls -j"
15
+
ls: invalid option -- 'j'
16
+
Try 'ls --help' for more information.
17
+
[2]
18
+
$ osh -c "ls -j | ls"
19
+
ls: invalid option -- 'j'
20
+
Try 'ls --help' for more information.
21
+
hello
22
+
23
+
And an exclaimation point should invert that.
24
+
25
+
$ sh -c "! ls -j | ls"
26
+
ls: invalid option -- 'j'
27
+
Try 'ls --help' for more information.
28
+
hello
29
+
[1]
30
+
$ osh -c "! ls -j | ls"
31
+
ls: invalid option -- 'j'
32
+
Try 'ls --help' for more information.
33
+
hello
34
+
[1]