CP-47033: Add test for concurrent message switch server

Signed-off-by: Vincent Liu <[email protected]>

ocaml/message-switch/core_test/async/server_async_main.ml

@@ -23,6 +23,8 @@ let path = ref "/var/run/message-switch/sock"
 
 let name = ref "server"
 
+let concurrent = ref false
+
 let shutdown = Ivar.create ()
 
 let process = function
@@ -33,7 +35,10 @@ let process = function
 
 let main () =
   let (_ : 'a Deferred.t) =
-    Server.listen ~process ~switch:!path ~queue:!name ()
+    if !concurrent then
+      Server.listen_p ~process ~switch:!path ~queue:!name ()
+    else
+      Server.listen ~process ~switch:!path ~queue:!name ()
   in
   Ivar.read shutdown >>= fun () ->
   Clock.after (Time.Span.of_sec 1.) >>= fun () -> exit 0
@@ -49,6 +54,11 @@ let _ =
       , Arg.Set_string name
       , Printf.sprintf "name to send message to (default %s)" !name
       )
+    ; ( "-concurrent"
+      , Arg.Set concurrent
+      , Printf.sprintf "set concurrent processing of messages (default %b)"
+          !concurrent
+      )
     ]
     (fun x -> P.fprintf stderr "Ignoring unexpected argument: %s" x)
     "Respond to RPCs on a name" ;

ocaml/message-switch/core_test/basic-rpc-test.sh

@@ -1,12 +1,14 @@
 #!/bin/bash
 set -e
 
-SPATH=${TMPDIR:-/tmp}/sock
-SWITCHPATH=${TMPDIR:-/tmp}/switch
+SPATH=${TMPDIR:-/tmp}/sock_s
+SWITCHPATH=${TMPDIR:-/tmp}/switch_s
 
 
 rm -rf ${SWITCHPATH} && mkdir -p ${SWITCHPATH}
 
+echo Test message switch serial processing
+
 echo Checking the switch can start late
 ./server_unix_main.exe -path $SPATH &
 sleep 1

ocaml/message-switch/core_test/concur-rpc-test.sh

@@ -0,0 +1,45 @@
+#!/bin/bash
+set -e
+
+SPATH="${TMPDIR:-/tmp}/sock_p-$$"
+SWITCHPATH="${TMPDIR:-/tmp}/switch_p-$$"
+
+trap "cleanup" TERM INT
+
+function cleanup {
+  rm -rf "${SWITCHPATH}"
+}
+
+rm -rf "${SWITCHPATH}" && mkdir -p "${SWITCHPATH}"
+
+echo Test message switch concurrent processing
+
+echo Checking the switch can start late
+test -x ./server_unix_main.exe || exit 1
+./server_unix_main.exe -path "$SPATH" &
+sleep 1
+test -x ../switch/switch_main.exe && test -x ./client_unix_main.exe || exit 1
+../switch/switch_main.exe --path "$SPATH" --statedir "${SWITCHPATH}" &
+./client_unix_main.exe -path "$SPATH" -secs 5
+sleep 2
+
+echo Performance test of Lwt to Lwt
+test -x lwt/server_main.exe && test -x lwt/client_main.exe || exit 1
+lwt/server_main.exe -path "$SPATH" -concurrent &
+lwt/client_main.exe -path "$SPATH" -secs 5
+sleep 2
+
+echo Performance test of Async to Lwt
+test -x lwt/server_main.exe && test -x async/client_async_main.exe || exit 1
+lwt/server_main.exe -path "$SPATH" -concurrent &
+async/client_async_main.exe -path "$SPATH" -secs 5
+sleep 2
+
+echo Performance test of Async to Async
+test -x async/server_async_main.exe && test -x async/client_async_main.exe || exit 1
+async/server_async_main.exe -path "$SPATH" -concurrent &
+async/client_async_main.exe -path "$SPATH" -secs 5
+sleep 2
+
+../cli/main.exe shutdown --path "$SPATH"
+sleep 2

ocaml/message-switch/core_test/dune

@@ -27,3 +27,20 @@
   (package message-switch)
 )
 
+(rule
+  (alias runtest)
+  (deps
+    client_unix_main.exe
+    server_unix_main.exe
+    async/client_async_main.exe
+    async/server_async_main.exe
+    lwt/client_main.exe
+    lwt/server_main.exe
+    lwt/link_test_main.exe
+    ../switch/switch_main.exe
+    ../cli/main.exe
+  )
+  (action (run ./concur-rpc-test.sh))
+  (package message-switch)
+)
+

ocaml/message-switch/core_test/lwt/server_main.ml

@@ -20,6 +20,8 @@ let path = ref "/var/run/message-switch/sock"
 
 let name = ref "server"
 
+let concurrent = ref false
+
 let t, u = Lwt.task ()
 
 let process = function
@@ -29,8 +31,13 @@ let process = function
       return x
 
 let main () =
-  Message_switch_lwt.Protocol_lwt.Server.listen ~process ~switch:!path
-    ~queue:!name ()
+  ( if !concurrent then
+      Message_switch_lwt.Protocol_lwt.Server.listen_p ~process ~switch:!path
+        ~queue:!name ()
+    else
+      Message_switch_lwt.Protocol_lwt.Server.listen ~process ~switch:!path
+        ~queue:!name ()
+  )
   >>= fun _ ->
   t >>= fun () -> Lwt_unix.sleep 1.
 
@@ -45,6 +52,11 @@ let _ =
       , Arg.Set_string name
       , Printf.sprintf "name to send message to (default %s)" !name
       )
+    ; ( "-concurrent"
+      , Arg.Set concurrent
+      , Printf.sprintf "set concurrent processing of messages (default %b)"
+          !concurrent
+      )
     ]
     (fun x -> Printf.fprintf stderr "Ignoring unexpected argument: %s" x)
     "Respond to RPCs on a name" ;