Use async in the master and add parallelism

master.ml

@@ -69,33 +69,34 @@ let server_handler pass port sockaddr (ic, _oc) =
   | Result.Ok query -> (
     match query with
     | `RESULT result ->
-        if pass = result.pass then (
-          (*Lwt_io.write oc "true"
-              <&>*)
-            Logs_lwt.debug (fun m ->
-                m "Receive result: %s\n"
-                  (Serialization_j.string_of_result
-                     {result with stdout= "<stdout>"; stderr= "<stderr>"}) )
-          >>= fun () ->
-          Lwt_io.with_file ~mode:Lwt_io.output
-            (Printf.sprintf "ocluster_%i_%i.out" (fst result.id)
-               (snd result.id))
-            (fun oc -> Lwt_io.write oc result.stdout)
-          >>= fun () ->
-          Lwt_io.with_file ~mode:Lwt_io.output
-            (Printf.sprintf "ocluster_%i_%i.err" (fst result.id)
-               (snd result.id))
-            (fun oc -> Lwt_io.write oc result.stderr)
+        if pass = result.pass then
+          Logs_lwt.debug (fun m ->
+              m "Receive result: %s\n"
+                (Serialization_j.string_of_result
+                   {result with stdout= "<stdout>"; stderr= "<stderr>"}) )
           >>= fun () ->
-          Lwt_io.with_file ~mode:Lwt_io.output
-            (Printf.sprintf "ocluster_%i_%i.log" (fst result.id)
-               (snd result.id))
-            (fun oc ->
-              Lwt_io.write oc
-                (Printf.sprintf "Job completed at %f\nReturn code: %s\n"
-                   (Unix.time ())
-                   (string_of_ret_code result.ret_code)) )
-          >>= fun () -> end_job result.id sockaddr ; launch_job () )
+          Lwt.return
+            (Lwt.async (fun () ->
+                 Lwt.join
+                   [ Lwt_io.with_file ~mode:Lwt_io.output
+                       (Printf.sprintf "ocluster_%i_%i.out" (fst result.id)
+                          (snd result.id))
+                       (fun oc -> Lwt_io.write oc result.stdout)
+                   ; Lwt_io.with_file ~mode:Lwt_io.output
+                       (Printf.sprintf "ocluster_%i_%i.err" (fst result.id)
+                          (snd result.id))
+                       (fun oc -> Lwt_io.write oc result.stderr)
+                   ; Lwt_io.with_file ~mode:Lwt_io.output
+                       (Printf.sprintf "ocluster_%i_%i.log" (fst result.id)
+                          (snd result.id))
+                       (fun oc ->
+                         Lwt_io.write oc
+                           (Printf.sprintf
+                              "Job completed at %f\nReturn code: %s\n"
+                              (Unix.time ())
+                              (string_of_ret_code result.ret_code)) )
+                   ; end_job result.id sockaddr >>= Lwt.pause >>= launch_job ]
+             ))
         else
           (*Lwt_io.write oc "Wrong password"
               <&>*)