Ordering is preserved

Parallel/Farm.ml

@@ -13,52 +13,90 @@ let run_sequential f stream =
 
 (* Multi-process functions *)
 
+type task_id = int
+
+
 (* Server side *)
-let run_parallel_server stream =
+let run_parallel_server ~ordered stream =
 
 
-  let fetch_result () : 'a option * int =
-    let  result, rank, _tag  =
+  let fetch_result () : (task_id * 'a) option * int =
+    let (message, rank, _tag) : (task_id * 'a) option * int * int =
       Mpi.receive_status  Mpi.any_source  Mpi.any_tag  Mpi.comm_world
     in
-    result, rank 
+    message, rank 
   in
 
 
   let send_task (client_rank : int) : unit =
     let task =
-      try Some (Stream.next stream)
+      try
+        let task_id = Stream.count stream in
+        let element = Stream.next stream  in
+        Some (task_id, element)
       with Stream.Failure -> None
     in
     Mpi.send  task  client_rank  0  Mpi.comm_world
   in
 
 
+  (* n_todo is required for clean termination. It is the
+     number of tasks to wait for when the input stream is
+     empty.  *)
   let n_todo = ref (Mpi.comm_size Mpi.comm_world ) in
 
-  let f i = 
-    let rec get_result () =
+  (* buffer of finished tasks with a task_id greater than the 
+     current result_id. It allows to put back the results in
+     the correct order.
+  *)
+
+  let rec get_result () : (task_id * 'a ) option =
+    begin
+      match Stream.peek stream with
+      | None -> decr n_todo
+      | _    -> ()
+    end;
+    match !n_todo with
+    | 0 -> 
       begin
-        match Stream.peek stream with
-        | None -> decr n_todo
-        | _    -> ()
-      end;
-      match !n_todo with
-      | 0 -> 
-        begin
-          Mpi.barrier  Mpi.comm_world;
-          None
-        end
-      | _ ->
-        begin
-          let result, client  = fetch_result () in
-          send_task  client;
-          match result with
-          | None -> get_result ()
-          | _    -> result
-        end
-    in
-    get_result ()
+        Mpi.barrier  Mpi.comm_world;
+        None
+      end
+    | _ ->
+      begin
+        let message, rank = fetch_result () in
+        send_task  rank;
+        match message with
+        | None -> get_result ()
+        | Some (task_id, result) -> Some (task_id, result)
+      end
+  in
+  
+  let f = 
+   if ordered then
+    let buffer = Hashtbl.create 67 in
+    fun i ->
+      begin
+          match Hashtbl.find_opt buffer i with
+          | Some x ->
+              begin
+                Hashtbl.remove buffer i;
+                Some x 
+              end
+          | None ->
+              let rec loop () =
+                match get_result () with
+                | None -> None
+                | Some (task_id, result) -> 
+                    if task_id = i then Some result
+                    else (Hashtbl.add buffer task_id result; loop () )
+              in loop ()
+      end
+   else
+     fun _ -> 
+       match get_result () with
+       | Some (_, result) -> Some result
+       | None             -> None
   in
   Stream.from f
 
@@ -68,16 +106,16 @@ let run_parallel_client f =
   Mpi.send  None  0  0  Mpi.comm_world;
 
   let rec run () =
-    let task = 
+    let message = 
       Mpi.receive  0  Mpi.any_tag  Mpi.comm_world
     in
 
-    match task with
+    match message with
     | None -> Mpi.barrier  Mpi.comm_world
-    | Some task ->
+    | Some (task_id, task) ->
       let result = f task in
       begin
-        Mpi.send  (Some result)  0  0  Mpi.comm_world;
+        Mpi.send  (Some (task_id, result))  0  0  Mpi.comm_world;
         run ()
       end
   in
@@ -86,14 +124,14 @@ let run_parallel_client f =
 
 
 
-let run_parallel f stream =
+let run_parallel ~ordered f stream =
   match Mpi.comm_rank  Mpi.comm_world with
-  | 0 -> run_parallel_server stream
+  | 0 -> run_parallel_server ~ordered stream
   | _ -> run_parallel_client f
 
 
-let run f stream =
+let run ?(ordered=true) ~f stream =
   match Mpi.comm_size Mpi.comm_world with
   | 1 -> run_sequential f stream
-  | _ -> run_parallel   f stream
+  | _ -> run_parallel ~ordered  f stream
 

run_parallel.ml

@@ -24,10 +24,13 @@ let v = Parallel.Vec.init 47 (fun i -> float_of_int i) in
 let () =
   let f (a,b) = (Parallel.rank, a+b) in
   let input  = Stream.of_list
-    [ (1,2) ; (3,4) ; (5,6) ; (7,8) ; (9,10) ]
+    [ (1,2) ; (3,4) ; (5,6) ; (7,8) ; (9,10)  
+    ; (1,2) ; (3,4) ; (5,6) ; (7,8) ; (9,10)  
+    ; (1,2) ; (3,4) ; (5,6) ; (7,8) ; (9,10)  
+    ; (1,2) ; (3,4) ; (5,6) ; (7,8) ; (9,10) ]
   in
   let stream =
-    Farm.run f input
+    Farm.run ~f input
   in
   Stream.iter (fun (x,y) -> Printf.printf "%d  %d\n%!" x y) stream