Projet SysRes

kahn.ml

module type S = sig
    type 'a process
    type 'a in_port
    type 'a out_port

    val new_channel: unit -> 'a in_port * 'a out_port
    val put: 'a -> 'a out_port -> unit process
    val get: 'a in_port -> 'a process
    val doco: unit process list -> unit process
    val return: 'a -> 'a process
    val bind: 'a process -> ('a -> 'b process) -> 'b process
    val run: 'a process -> 'a
end


module Lib (K : S) = struct
    let ( >>= ) x f = K.bind x f

    let delay f x =
        (K.return ()) >>= (fun () -> K.return (f x))
end


module Th: S = struct
    type 'a process = (unit -> 'a)
    type 'a channel = { q: 'a Queue.t ; m: Mutex.t; }
    type 'a in_port = 'a channel
    type 'a out_port = 'a channel

    let new_channel () =
        let q = { q = Queue.create (); m = Mutex.create (); } in
        q, q

    let put v c () =
        Mutex.lock c.m;
        Queue.push v c.q;
        Mutex.unlock c.m;
        Thread.yield ()

    let rec get c () =
        try
            Mutex.lock c.m;
            let v = Queue.pop c.q in
            Mutex.unlock c.m;
            v
        with Queue.Empty ->
            Mutex.unlock c.m;
            Thread.yield ();
            get c ()

    let doco l () =
        let ths = List.map (fun f -> Thread.create f ()) l in
        List.iter (fun th -> Thread.join th) ths

    let return v = (fun () -> v)

    let bind e e' () =
        let v = e () in
        Thread.yield ();
        e' v ()

    let run e = e ()
end


module Pr: S = struct
    type 'a process = (unit -> 'a)
    type 'a in_port = in_channel
    type 'a out_port = out_channel

    let new_channel () =
        let in_fd, out_fd = Unix.pipe ()
        in
            (Unix.in_channel_of_descr in_fd, Unix.out_channel_of_descr out_fd)

    (* FIXME Check for errors *)
    let put v c () = Marshal.to_channel c v [Marshal.Closures]

    let get c () = Marshal.from_channel c

    let rec doco l () =
        match l with
        | [] -> ()
        | h :: t ->
            let cid = Unix.fork ()
            in
                if cid = 0
                then
                    (doco t ();
                     exit 0)
                else
                    (h ();
                     ignore (Unix.waitpid [] cid))

    let return v = (fun () -> v)

    let bind e e' () =
        let v = e ()
        in
            e' v ()

    let run e = e ()
end


module Cn : S = struct
    (* XXX If the type were ('a -> 'b) -> 'b return would probably be a lot simpler *)
    type 'a process = ('a -> unit) -> unit

    type 'a channel = { q: 'a Queue.t ; m: Mutex.t; }
    type 'a in_port = 'a channel
    type 'a out_port = 'a channel


    let new_channel () =
        let q = { q = Queue.create (); m = Mutex.create (); } in
            q, q

    let put v c cont =
        Mutex.lock c.m;
        Queue.push v c.q;
        Mutex.unlock c.m;
        cont ()

    let rec get c cont =
        try
            Mutex.lock c.m;
            let v = Queue.pop c.q in
            Mutex.unlock c.m;
            cont v
        with Queue.Empty ->
            Mutex.unlock c.m;
            get c cont

    let rec doco l cont =
        match l with
        | [] -> cont ()
        | h :: t -> h (fun () -> doco t cont)

    let return v cont =
        cont v

    let bind e e' cont =
        e (fun x -> e' x cont)

    let run e =
        let res = ref None
        in
            e (fun x -> res := Some x; ());
            match !res with
            | None -> failwith "Continuation not called!"
            | Some res -> res

    exception ReturnException of int

    let run2 e =
        try
            e (fun x -> raise (ReturnException x));
            failwith "Continuation not called!"
        with ReturnException x ->
            x
end

main.ml

module Example (K : Kahn.S) = struct
    module Lib = Kahn.Lib(K)
    open Lib

    let integers (qo : int K.out_port) : unit K.process =
        let rec loop n =
            (K.put n qo) >>= (fun () -> loop (n + 1))
        in
        loop 2

    let output (qi : int K.in_port) : unit K.process =
        let rec loop () =
            (K.get qi) >>= (fun v -> Format.printf "%d@." v; loop ())
        in
        loop ()

    let main : unit K.process =
        (delay K.new_channel ()) >>=
        (fun (q_in, q_out) -> K.doco [ integers q_out ; output q_in ; ])
end

module ExampleTh = Example(Kahn.Th)
module ExamplePr = Example(Kahn.Pr)
module ExampleCn = Example(Kahn.Cn)

let () =
    Kahn.Cn.run ExampleCn.main