PiCalculus module, to emulate the Pi-calculus concurrency model;

it uses only one thread on one machine though

containers.mllib

@@ -17,3 +17,4 @@ SplayMap
 Univ
 Vector
 Bij
+PiCalculus

containers.odocl

@@ -8,3 +8,4 @@ PersistentHashtbl
 Sequence
 Univ
 Bij
+PiCalculus

piCalculus.ml

@@ -0,0 +1,287 @@
+(*
+copyright (c) 2013, simon cruanes
+all rights reserved.
+
+redistribution and use in source and binary forms, with or without
+modification, are permitted provided that the following conditions are met:
+
+redistributions of source code must retain the above copyright notice, this
+list of conditions and the following disclaimer.  redistributions in binary
+form must reproduce the above copyright notice, this list of conditions and the
+following disclaimer in the documentation and/or other materials provided with
+the distribution.
+
+this software is provided by the copyright holders and contributors "as is" and
+any express or implied warranties, including, but not limited to, the implied
+warranties of merchantability and fitness for a particular purpose are
+disclaimed. in no event shall the copyright holder or contributors be liable
+for any direct, indirect, incidental, special, exemplary, or consequential
+damages (including, but not limited to, procurement of substitute goods or
+services; loss of use, data, or profits; or business interruption) however
+caused and on any theory of liability, whether in contract, strict liability,
+or tort (including negligence or otherwise) arising in any way out of the use
+of this software, even if advised of the possibility of such damage.
+*)
+
+(** {1 Pi-calculus model of concurrency} *)
+
+module DList = struct
+  type 'a t = {
+    value : 'a wrapper;
+    mutable prev : 'a t;
+    mutable next : 'a t;
+  }
+  and 'a wrapper =
+    | First  (* first element of the list *)
+    | Element of 'a
+
+  (** New empty list *)
+  let create () =
+    let rec node = {
+      value = First;
+      prev = node;
+      next = node;
+    } in
+    node
+
+  let is_empty l =
+    let ans = l.prev == l in
+    (if ans then (assert (l.next == l && l.value == First)));
+    ans
+
+  (** Add element at the end *)
+  let append l x =
+    let node = {
+      value = Element x;
+      prev = l.prev;
+      next = l;
+    } in
+    l.prev.next <- node;
+    l.prev <- node;
+    node
+
+  (** Add element at the beginning *)
+  let prepend l x =
+    let node = {
+      value = Element x;
+      prev = l;
+      next = l.next;
+    } in
+    l.next.prev <- node;
+    l.next <- node;
+    node
+
+  (* remove the given element *)
+  let remove x =
+    assert (not (x.prev == x || x.next == x));
+    x.prev.next <- x.next;
+    x.next.prev <- x.prev;
+    ()
+
+  (** Pop the first element *)
+  let pop l =
+    match l.next.value with
+    | First -> failwith "DList.pop: empty list"
+    | Element x ->
+      remove l.next;
+      x
+
+  let rec remove_list l = match l with
+    | [] -> ()
+    | x::l' -> remove x; remove_list l'
+
+  (** Iterate on all elements *)
+  let iter l f =
+    let rec iter l = match l.value with
+    | First -> ()
+    | Element x ->
+      f x;
+      iter l.next
+    in
+    iter l.next
+end
+
+type 'a chan = {
+  receivers : 'a transition_node DList.t;
+  senders : 'a transition_node DList.t;
+} (** Channel conveying values of type 'a. Invariant: receivers = None || senders = None *)
+and 'a transition_node = {
+  tn_transition : 'a __transition;
+  mutable tn_hook : unit -> unit;       (* hook to call after transition *)
+  tn_to_replicate : to_replicate ref;   (* do we have to replicate a process *)
+} (** List of transitions for a channel *)
+and to_replicate =
+  | ReplicateNothing
+  | ReplicateThis of process
+  (** Do we have to replicate a process? *)
+and process =
+  | Parallel : process list -> process      (** Spawn several processes *)
+  | Sum : transition list -> process        (** Choice point *)
+  | Replicate : process -> process          (** Replication of a process *)
+  | New : ('a chan -> process) -> process   (** New local name *)
+  | Escape : (unit -> process) -> process   (** Run a user function *)
+  | Stop : process                          (** Stop this process *)
+  (** A process of the Pi-calculus *)
+and _ __transition =
+  | Receive : 'a chan * ('a -> process) -> 'a __transition
+  | Send : 'a chan * 'a * process -> 'a __transition
+  (** Transition: send or receive a message *)
+and transition =
+  | Transition : 'a __transition -> transition
+
+let parallel l = (assert (l <> []); Parallel l)
+let sum l = (assert (l <> []); Sum l)
+let replicate p = Replicate p
+let new_ f = New f
+let escape f = Escape f
+let stop = Stop
+
+let send ch x p = Transition (Send (ch, x, p))
+let receive ch f = Transition (Receive (ch, f))
+
+let send_one ch x p = sum [send ch x p]
+let receive_one ch f = sum [receive ch f]
+
+let (>>) f p =
+  escape (fun () -> f (); p)
+
+let (|||) a b = parallel [a; b]
+
+let (++) a b = sum [a; b]
+
+(** New channel (name) *)
+let mk_chan () =
+  let ch = {
+    receivers = DList.create ();
+    senders = DList.create ();
+  } in
+  ch
+
+type run_env = {
+  tasks : (process * to_replicate ref) Queue.t;
+} (** Environment for running processes *)
+
+let mk_env () =
+  { tasks = Queue.create (); }
+
+(** Push the process in the queue of processes to eval *)
+let push_process ~env p to_restart =
+  Queue.push (p, to_restart) env.tasks
+
+(** Check whether there is a process to replicate now *)
+let check_replicate ~env to_replicate =
+  match !to_replicate with
+  | ReplicateNothing -> ()
+  | ReplicateThis p' ->
+    (* replicate p' now; it will be useless from now on to replicate it again *)
+    push_process ~env p' (ref ReplicateNothing);
+    to_replicate := ReplicateNothing
+
+(** Make a new transition node (linked to nothing) *)
+let mk_transition_node transition to_replicate = 
+  let node = {
+    tn_transition = transition;
+    tn_hook = (fun () -> ());
+    tn_to_replicate = to_replicate;
+  } in
+  node
+  
+(** Perform the given transition (one send, one receive). *)
+let perform_transition
+: type a. env:run_env -> a transition_node -> a transition_node -> unit =
+fun ~env sender receiver ->
+  (* cleanup alternatives, replicate some processes if needed *)
+  sender.tn_hook ();
+  receiver.tn_hook ();
+  check_replicate ~env sender.tn_to_replicate;
+  check_replicate ~env receiver.tn_to_replicate;
+  match sender.tn_transition, receiver.tn_transition with
+  | Send (ch, x, send_p), Receive (ch', receive_p) ->
+    assert (ch == ch');
+    (* receiving channel gets the sent value *)
+    let receive_p = receive_p x in
+    (* push the two new processes (with no process to replicate) *)
+    push_process ~env send_p (ref ReplicateNothing);
+    push_process ~env receive_p (ref ReplicateNothing);
+    ()
+  | _ -> assert false
+
+(** Check whether any transition in the list can be performed; otherwise,
+    register all of them to their respective channels; Returns the
+    list of corresponding  [transition_node] (empty if some
+    transition fired immediately). *)
+let try_transitions ~env transitions to_replicate =
+  try
+    let set_hooks, hook = List.fold_left
+      (fun (set_hooks, hook) transition -> match transition with
+        | Transition (Receive (ch, _) as transition) ->
+          let receiver = mk_transition_node transition to_replicate in
+          if DList.is_empty ch.senders
+            then (* wait *)
+              let dlist = DList.append ch.receivers receiver in
+              (fun hook -> receiver.tn_hook <- hook) :: set_hooks,
+              (fun () -> DList.remove dlist; hook ())
+            else begin (* fire *)
+              let sender = DList.pop ch.senders in
+              perform_transition ~env sender receiver;
+              hook ();  (* cancel previous sum cases *)
+              raise Exit
+            end
+        | Transition (Send (ch, _, _) as transition) ->
+          let sender = mk_transition_node transition to_replicate in
+          if DList.is_empty ch.receivers
+            then  (* wait *)
+              let dlist = DList.append ch.senders sender in
+              (fun hook -> sender.tn_hook <- hook) :: set_hooks,
+              (fun () -> DList.remove dlist; hook ())
+            else begin (* fire *)
+              let receiver = DList.pop ch.receivers in
+              perform_transition ~env sender receiver;
+              hook ();  (* cancel previous sum cases *)
+              raise Exit
+            end)
+      ([], fun () -> ()) transitions
+    in
+    (* we have a list of transition nodes; save it for when a transition fires *)
+    List.iter (fun set_hook -> set_hook hook) set_hooks
+  with Exit -> (* some transition fired immediately *)
+    ()
+
+(** Run the simulation until all processes are stuck, or stopped. *)
+let run p =
+  (* run tasks one by one until none remains *)
+  let rec run  : env:run_env -> unit = fun ~env ->
+    if not (Queue.is_empty env.tasks) then begin
+      (* eval next process *)
+      let p, to_replicate = Queue.pop env.tasks in
+      eval_process ~env p to_replicate;
+      run ~env
+    end
+  (* evaluate this process *)
+  and eval_process : env:run_env -> process -> to_replicate ref -> unit
+  = fun ~env p to_replicate ->
+    match p with
+    | Stop -> (* stop, but maybe there is a process to replicate *)
+      check_replicate ~env to_replicate
+    | New f ->
+      (* apply [f] to a new chan *)
+      let c = mk_chan () in
+      let p' = f c in
+      eval_process ~env p' to_replicate
+    | Parallel l ->
+      (* evaluate each process *)
+      List.iter (fun p -> push_process ~env p to_replicate) l
+    | Replicate p' ->
+      (* run [p'] within an env where [p] is to be replicated *)
+      let to_replicate' = ref (ReplicateThis p) in
+      eval_process ~env p' to_replicate'
+    | Escape f ->
+      let p' = f () in
+      push_process ~env p' to_replicate  (* yield before processing the result *)
+    | Sum transitions ->
+      try_transitions ~env transitions to_replicate
+  in
+  (* initial env *)
+  let env = mk_env () in
+  push_process ~env p (ref ReplicateNothing);
+  run ~env

piCalculus.mli

@@ -0,0 +1,74 @@
+(*
+copyright (c) 2013, simon cruanes
+all rights reserved.
+
+redistribution and use in source and binary forms, with or without
+modification, are permitted provided that the following conditions are met:
+
+redistributions of source code must retain the above copyright notice, this
+list of conditions and the following disclaimer.  redistributions in binary
+form must reproduce the above copyright notice, this list of conditions and the
+following disclaimer in the documentation and/or other materials provided with
+the distribution.
+
+this software is provided by the copyright holders and contributors "as is" and
+any express or implied warranties, including, but not limited to, the implied
+warranties of merchantability and fitness for a particular purpose are
+disclaimed. in no event shall the copyright holder or contributors be liable
+for any direct, indirect, incidental, special, exemplary, or consequential
+damages (including, but not limited to, procurement of substitute goods or
+services; loss of use, data, or profits; or business interruption) however
+caused and on any theory of liability, whether in contract, strict liability,
+or tort (including negligence or otherwise) arising in any way out of the use
+of this software, even if advised of the possibility of such damage.
+*)
+
+(** {1 Pi-calculus model of concurrency} *)
+
+type 'a chan
+  (** Channel conveying values of type 'a *)
+
+type process = private
+  | Parallel : process list -> process      (** Spawn several processes *)
+  | Sum : transition list -> process        (** Choice point *)
+  | Replicate : process -> process          (** Replication of a process *)
+  | New : ('a chan -> process) -> process   (** New local name *)
+  | Escape : (unit -> process) -> process   (** Run a user function *)
+  | Stop : process                          (** Stop this process *)
+and 'a __transition =
+  | Receive : 'a chan * ('a -> process) -> 'a __transition
+  | Send : 'a chan * 'a * process -> 'a __transition
+and transition =
+  | Transition : 'a __transition -> transition
+
+val parallel : process list -> process
+val sum : transition list -> process
+val replicate : process -> process
+val new_ : ('a chan -> process) -> process
+val escape : (unit -> process) -> process
+val stop : process
+
+val send : 'a chan -> 'a -> process -> transition
+val receive : 'a chan -> ('a -> process) -> transition
+
+(** Be careful: there must be at least one send/receive between a replicate
+   and a stop, otherwise {! run} will get stuck in a loop, replicating the
+   process forever. *)
+
+val send_one : 'a chan -> 'a -> process -> process
+  (** Send a value, with no alternative *)
+
+val receive_one : 'a chan -> ('a -> process) -> process
+  (** Receive a value, with no alternative *)  
+
+val (>>) : (unit -> unit) -> process -> process
+  (** Perform the action, then proceed to the following process *)
+
+val (|||) : process -> process -> process
+  (** Infix version of {! parallel} for two processes *)
+
+val (++) : transition -> transition -> process
+  (** Infix version of {! sum} for two processes *)
+
+val run : process -> unit
+  (** Run the simulation until all processes are stuck, or stopped. *)