#
#
# add_file "automate.ml"
# content [ad1749526d052113da277fbcb68813a836cfed37]
#
# add_file "automate.mli"
# content [af3fd8e56df2a66b0113a47ca374f1d065211b44]
#
# patch "Makefile"
# from [9e00a714809bb52983da5f1d34790d1727a4472f]
# to [32ba8864bebd73979ee240329267e77b001e66d4]
#
# patch "ui.ml"
# from [89f9af8477b455fc355375fdc5edbc0f7d3a9241]
# to [a1771efa0cc6a979a8436e99a43df62b74881081]
#
============================================================
--- automate.ml ad1749526d052113da277fbcb68813a836cfed37
+++ automate.ml ad1749526d052113da277fbcb68813a836cfed37
@@ -0,0 +1,459 @@
+open Viz_misc
+
+let init =
+ Giochannel.init ; Gspawn.init
+
+let debug = Viz_misc.debug "automate"
+
+let log fmt =
+ Printf.kprintf
+ (fun s -> Printf.eprintf "### automate: %s\n%!" s)
+ fmt
+
+
+(** Type definitions *)
+
+type pb = [
+ | `EXIT of int
+ | `FAILURE
+ | `ERROR of exn ]
+type watch = {
+ w_name : string ;
+ w_chan : Giochannel.t ;
+ mutable w_finished : bool ;
+ mutable w : Giochannel.source_id option ;
+ exn_cb : pb -> unit ;
+ }
+type in_watch = {
+ in_w : watch ;
+ mutable in_data : (int * string) list ;
+ mutable in_pos : int ;
+ }
+type out_watch = {
+ out_w : watch ;
+ out_sb : string ;
+ out_buffer : Buffer.t ;
+ out_cb : (Buffer.t -> unit)
+ }
+type process = {
+ p_in : in_watch ;
+ p_out : out_watch ;
+ p_err : out_watch ;
+ pid : Gspawn.pid ;
+ mutable state : [`RUNNING | `EXITING ]
+ }
+
+
+type command_id = int
+type output = [
+ | `OUTPUT of string
+ | `ERROR of string
+ | `SYNTAX_ERROR of string]
+type chunk = command_id * int * bool * string
+type t = {
+ mtn : string ;
+ db_fname : string ;
+ mutable process : process option ;
+ mutable cmd_number : command_id ;
+ mutable callbacks : (command_id * (output -> unit)) list ;
+ mutable chunks : (command_id * chunk list ref) list ;
+ }
+
+let string_of_conditions cond =
+ let s = String.make 6 '.' in
+ Array.iteri
+ (fun i (v, c) -> if List.mem v cond then s.[i] <- c)
+ [| `IN, 'I' ; `OUT, 'O' ;
+ `HUP, 'H' ; `ERR, 'E' ;
+ `PRI, 'P' ; `NVAL, 'N' |] ;
+ s
+
+
+
+let do_write w data =
+ let bytes_written = ref 0 in
+ try
+ match Giochannel.write_chars w.in_w.w_chan ~bytes_written ~off:w.in_pos data with
+ | `NORMAL written ->
+ if debug then log "%s cb: wrote %d" w.in_w.w_name written ;
+ w.in_pos <- w.in_pos + written
+ | `AGAIN ->
+ (* should not happen, our out channels are blocking *)
+ if debug then log "%s cb: EAGAIN ?" w.in_w.w_name
+ with
+ | Giochannel.Error (_, msg) as exn ->
+ (* an error occurred, we continue here but the the exn callback will probably tear down everything *)
+ if debug then log "%s cb: error %s, wrote %d" w.in_w.w_name msg !bytes_written ;
+ w.in_w.exn_cb (`ERROR exn)
+
+let _write_cb w conditions =
+ if debug then log "%s cb = %s" w.in_w.w_name (string_of_conditions conditions) ;
+ match w.in_data with
+ | [] ->
+ (* nothing to write, remove the source from the main loop *)
+ if debug then log "%s cb: empty write queue, removing watch" w.in_w.w_name ;
+ w.in_w.w <- None ;
+ false
+
+ | (_, data) :: tl ->
+ (* some data to write *)
+ let len = String.length data in
+ let to_write = len - w.in_pos in
+ assert (len > 0 && to_write > 0) ;
+ if debug then log "%s cb: %d left in buffer" w.in_w.w_name to_write ;
+ if debug && w.in_pos = 0 then log "%s cb: writing '%s'" w.in_w.w_name (String.escaped data) ;
+
+ if List.mem `OUT conditions
+ then begin
+ do_write w data ;
+ if w.in_pos >= len
+ then begin
+ (* written everything, proceed to the next chunk *)
+ w.in_data <- tl ;
+ w.in_pos <- 0
+ end ;
+ true
+ end
+ else begin
+ (* no `OUT condition, only `HUP or `ERR: not good, the channel is going down ! *)
+ w.in_w.exn_cb `FAILURE ;
+ false
+ end
+
+
+
+let _read_cb w conditions =
+ if debug then log "%s cb = %s" w.out_w.w_name (string_of_conditions conditions) ;
+ if List.mem `IN conditions
+ then begin
+ try
+ match Giochannel.read_chars w.out_w.w_chan w.out_sb with
+ | `NORMAL read ->
+ if debug then log "%s cb: read %d" w.out_w.w_name read ;
+ Buffer.add_substring w.out_buffer w.out_sb 0 read ;
+ w.out_cb w.out_buffer ;
+ true
+ | `EOF ->
+ if debug then log "%s cb: eof" w.out_w.w_name ;
+ w.out_w.exn_cb `FAILURE ;
+ false
+ | `AGAIN ->
+ if debug then log "%s cb: AGAIN" w.out_w.w_name ;
+ true
+ with exn ->
+ if debug then log "%s cb: error %s" w.out_w.w_name (Printexc.to_string exn) ;
+ w.out_w.exn_cb (`ERROR exn) ;
+ false
+ end
+ else begin
+ (* no `IN condition, only `HUP or `ERR: not good, the channel is going down ! *)
+ w.out_w.exn_cb `FAILURE ;
+ false
+ end
+
+let write_cb w c =
+ try _write_cb w c
+ with exn ->
+ if debug
+ then log "write cb %s: uncaught exception '%s'" w.in_w.w_name (Printexc.to_string exn) ;
+ false
+
+let read_cb w c =
+ try _read_cb w c
+ with exn ->
+ if debug
+ then log "read cb %s: uncaught exception '%s'" w.out_w.w_name (Printexc.to_string exn) ;
+ false
+
+
+let setup_watch_write w nb data =
+ match w.in_w.w with
+ | None ->
+ w.in_data <- [ nb, data ] ;
+ w.in_pos <- 0 ;
+ let id = Giochannel.add_watch w.in_w.w_chan [ `OUT ; `HUP ; `ERR ] (write_cb w) in
+ w.in_w.w <- Some id
+ | Some id ->
+ w.in_data <- w.in_data @ [ nb, data ]
+
+
+let setup_watch_read w =
+ assert (w.out_w.w = None) ;
+ let id = Giochannel.add_watch w.out_w.w_chan [ `IN ; `HUP ; `ERR ] (read_cb w) in
+ w.out_w.w <- Some id
+
+
+let setup_channel ~nonblock fd =
+ let chan = Giochannel.unix_new (some fd) in
+ if nonblock then Giochannel.set_flags chan [`NONBLOCK] ;
+ Giochannel.set_encoding chan None ;
+ Giochannel.set_buffered chan false ;
+ chan
+let make_watch name chan exn_cb =
+ { w_name = name ; w_chan = chan ; w_finished = false ; w = None ; exn_cb = exn_cb }
+let make_in_watch name fd exn_cb =
+ let chan = setup_channel ~nonblock:true fd in
+ { in_w = make_watch name chan exn_cb ; in_data = [] ; in_pos = 0 }
+let make_out_watch name fd exn_cb out_cb =
+ let chan = setup_channel ~nonblock:false fd in
+ let w = {
+ out_w = make_watch name chan exn_cb ;
+ out_sb = String.create 4096 ;
+ out_buffer = Buffer.create 1024 ;
+ out_cb = out_cb
+ } in
+ setup_watch_read w ;
+ w
+
+
+
+
+
+let spawn mtn db exn_cb out_cb =
+ let flags =
+ [ `PIPE_STDIN ; `PIPE_STDOUT ; `PIPE_STDERR ;
+ `SEARCH_PATH ; `DO_NOT_REAP_CHILD ] in
+ let child =
+ Gspawn.async_with_pipes
+ ~flags
+ [ mtn ; "-d" ; db ; "automate" ; "stdio" ] in
+ let pid = some child.Gspawn.pid in
+ ignore (Gspawn.add_child_watch pid (fun st -> exn_cb (`EXIT st))) ;
+ { p_in = make_in_watch "stdin" child.Gspawn.standard_input exn_cb ;
+ p_out = make_out_watch "stdout" child.Gspawn.standard_output exn_cb out_cb ;
+ p_err = make_out_watch "stderr" child.Gspawn.standard_error exn_cb ignore ;
+ pid = pid ;
+ state = `RUNNING
+ }
+
+
+
+let send_data p nb data =
+ if String.length data = 0
+ then invalid_arg "Automate.send_data: empty string" ;
+ setup_watch_write p.p_in nb data
+
+
+
+let encode_stdio cmd =
+ let b = Buffer.create 512 in
+ Buffer.add_char b 'l' ;
+ List.iter
+ (fun s -> Printf.bprintf b "%d:%s" (String.length s) s)
+ cmd ;
+ Buffer.add_string b "e\n" ;
+ Buffer.contents b
+
+
+
+let find_four_colon b =
+ let to_find = ref 4 in
+ let i = ref 0 in
+ while !to_find > 0 do
+ let c = Buffer.nth b !i in
+ if c = ':' then decr to_find ;
+ incr i
+ done ;
+ !i
+
+let truncate_buffer b off len =
+ let data = Buffer.sub b off len in
+ let rest = Buffer.sub b (off + len) (Buffer.length b - off - len) in
+ Buffer.clear b ;
+ Buffer.add_string b rest ;
+ data
+
+let decode_stdio_chunk b =
+ try
+ let header_len = find_four_colon b in
+ let h = Buffer.sub b 0 header_len in
+ let c1 = String.index_from h 0 ':' in
+ let number = int_of_string (string_slice ~e:c1 h) in
+ let code = int_of_char h.[c1 + 1] - int_of_char '0' in
+ let c2 = String.index_from h (c1 + 1) ':' in
+ let last = h.[c2 + 1] in
+ let c3 = String.index_from h (c2 + 1) ':' in
+ let c4 = String.index_from h (c3 + 1) ':' in
+ let len = int_of_string (string_slice ~s:(c3 + 1) ~e:c4 h) in
+ if Buffer.length b < header_len + len
+ then
+ `INCOMPLETE
+ else
+ let data = truncate_buffer b header_len len in
+ `CHUNK (number, code, last = 'l', data)
+ with Invalid_argument _ ->
+ `INCOMPLETE
+
+
+let aborted_cmd ctrl nb =
+ not (List.mem_assoc nb ctrl.callbacks)
+
+let rec out_cb ctrl b =
+ match decode_stdio_chunk b with
+ | `INCOMPLETE ->
+ ()
+
+ | `CHUNK (nb, _, _, _) when aborted_cmd ctrl nb ->
+ ctrl.chunks <- List.remove_assoc nb ctrl.chunks ;
+ out_cb ctrl b
+
+ | `CHUNK ((nb, code, false, data) as chunk) ->
+ let previous_chunks =
+ try List.assoc nb ctrl.chunks
+ with Not_found ->
+ let c = ref [] in
+ ctrl.chunks <- (nb, c) :: ctrl.chunks ;
+ c in
+ previous_chunks := chunk :: !previous_chunks ;
+ out_cb ctrl b
+
+ | `CHUNK ((nb, code, true, data) as chunk) ->
+ let chunks =
+ try
+ let c = List.assoc nb ctrl.chunks in
+ ctrl.chunks <- List.remove_assoc nb ctrl.chunks ;
+ List.rev (chunk :: !c)
+ with Not_found ->
+ [ chunk ] in
+ let cb = List.assoc nb ctrl.callbacks in
+ ctrl.callbacks <- List.remove_assoc nb ctrl.callbacks ;
+ let msg =
+ String.concat ""
+ (List.map (fun (_, _, _, d) -> d) chunks) in
+ let data =
+ match code with
+ | 0 -> `OUTPUT msg
+ | 1 -> `SYNTAX_ERROR msg
+ | 2 -> `ERROR msg
+ | _ -> failwith "invalid_code in automate stdio output" in
+ ignore (Glib.Idle.add ~prio:0 (fun () -> cb data ; false)) ;
+ out_cb ctrl b
+
+
+let stop_watch w =
+ if debug then log "stopping watch %s" w.w_name ;
+ begin
+ match w.w with
+ | Some id ->
+ if debug then log "stopping watch %s: remove" w.w_name ;
+ Giochannel.remove_watch id ;
+ w.w <- None
+ | None ->
+ ()
+ end ;
+ if not w.w_finished
+ then begin
+ if debug then log "stopping watch %s: shutdown" w.w_name ;
+ Giochannel.shutdown w.w_chan false ;
+ w.w_finished <- true
+ end ;
+ if debug then log "stopping watch %s: all done" w.w_name
+
+
+let tear_down ctrl arg =
+ let msg =
+ match arg with
+ | `EXIT s ->
+ Printf.sprintf "exiting with status %d" s
+ | `FAILURE ->
+ "some channel is closing"
+ | `ERROR (Giochannel.Error (_, msg)) ->
+ Printf.sprintf "GIOChannel error: %s" msg
+ | `ERROR exn ->
+ Printf.sprintf "uncaught exception: %s" (Printexc.to_string exn) in
+ if debug then log "tear_down cb: '%s'" msg ;
+ match ctrl.process with
+ | None ->
+ if debug then log "tear_down cb: no process ?"
+ | Some p ->
+ if p.state <> `EXITING
+ then begin
+ if debug then log "tear_down cb: removing watches" ;
+ p.state <- `EXITING ;
+ stop_watch p.p_in.in_w ;
+ stop_watch p.p_out.out_w ;
+ stop_watch p.p_err.out_w ;
+ end
+ else
+ if debug then log "tear_down cb: process already exiting" ;
+ match arg with
+ | `EXIT _ ->
+ if debug then log "tear_down cb: clearing process" ;
+ ctrl.callbacks <- [] ;
+ ctrl.chunks <- [] ;
+ Gspawn.close_pid p.pid ;
+ ctrl.process <- None
+ | _ -> ()
+
+let ensure_process ctrl =
+ match ctrl.process with
+ | Some p ->
+ assert (p.state = `RUNNING) ;
+ p
+ | None ->
+ let p =
+ spawn
+ ctrl.mtn ctrl.db_fname
+ (tear_down ctrl)
+ (out_cb ctrl) in
+ ctrl.process <- Some p ;
+ ctrl.cmd_number <- 0 ;
+ p
+
+
+
+
+
+
+
+
+
+let make mtn db = {
+ mtn = mtn ;
+ db_fname = db ;
+ process = None ;
+ cmd_number = 0 ;
+ callbacks = [] ;
+ chunks = [] ;
+}
+
+let exit ctrl =
+ match ctrl.process with
+ | None ->
+ ()
+ | Some p ->
+ stop_watch p.p_in.in_w
+
+
+let submit c cmd cb =
+ let p = ensure_process c in
+ let id = c.cmd_number in
+ send_data p id (encode_stdio cmd) ;
+ c.cmd_number <- id + 1 ;
+ c.callbacks <- (id, cb) :: c.callbacks ;
+ id
+
+let abort ctrl nb =
+ ctrl.callbacks <- List.remove_assoc nb ctrl.callbacks ;
+ match ctrl.process with
+ | None ->
+ ()
+ | Some p ->
+ match p.p_in.in_data with
+ | (id, _) :: tl when id = nb && p.p_in.in_pos = 0 ->
+ p.p_in.in_data <- tl ;
+ p.p_in.in_pos <- 0
+ | h :: tl ->
+ p.p_in.in_data <- h :: (List.remove_assoc nb tl)
+ | [] ->
+ ()
+
+
+(* TODO:
+ - add a timeout to exit the subprocess in case of inactivity
+ - add a submit_delayed to submit a cancellable command
+ with a small timeout (for keyboard nav)
+ - check exceptions and callbacks
+ - add asserts and sanity checks
+ *)
+
============================================================
--- automate.mli af3fd8e56df2a66b0113a47ca374f1d065211b44
+++ automate.mli af3fd8e56df2a66b0113a47ca374f1d065211b44
@@ -0,0 +1,15 @@
+
+type t
+type command_id = int
+type output = [
+ | `OUTPUT of string
+ | `ERROR of string
+ | `SYNTAX_ERROR of string]
+
+val make : string -> string -> t
+val exit : t -> unit
+
+val submit : t -> string list -> (output -> unit) -> command_id
+
+val abort : t -> command_id -> unit
+
============================================================
--- Makefile 9e00a714809bb52983da5f1d34790d1727a4472f
+++ Makefile 32ba8864bebd73979ee240329267e77b001e66d4
@@ -22,6 +22,7 @@ SRC = $(OCAMLNET)/base64.ml $(OCAMLNET)/
revision_types.mli revision_lexer.ml revision_parser.ml revision_parser.mli \
subprocess.ml subprocess.mli \
components.ml schema_lexer.ml \
+ automate.mli automate.ml \
database.ml database.mli agraph.ml agraph.mli \
autocolor.ml autocolor.mli viz_style.ml viz_style.mli \
icon.ml ui.ml ui.mli unidiff.ml unidiff.mli \
============================================================
--- ui.ml 89f9af8477b455fc355375fdc5edbc0f7d3a9241
+++ ui.ml a1771efa0cc6a979a8436e99a43df62b74881081
@@ -367,6 +367,7 @@ let ui_info = "\
\
\
\
+ \
\
\