Merge pull request #200 from talex5/benchmark

Improve benchmark performance

capnp-rpc-lwt/capnp_rpc_lwt.mli

@@ -71,9 +71,12 @@ module Capability : sig
     type 'a t
     (** An ['a t] is a builder for the out-going request's payload. *)
 
-    val create : (Capnp.Message.rw Slice.t -> 'a) -> 'a t * 'a
+    val create : ?message_size:int -> (Capnp.Message.rw Slice.t -> 'a) -> 'a t * 'a
     (** [create init] is a fresh request payload and contents builder.
-        Use one of the generated [init_pointer] functions for [init]. *)
+        Use one of the generated [init_pointer] functions for [init].
+        @param message_size An estimate of the size of the payload. If this is too small,
+                            additional segments will be allocated automatically, but this
+                            is less efficient than getting the size right to start with. *)
 
     val create_no_args : unit -> 'a t
     (** [create_no_args ()] is a payload with no content. *)
@@ -210,7 +213,7 @@ module Service : sig
     type 'b t
     (** An ['a t] is a builder for the out-going response. *)
 
-    val create : (Capnp.Message.rw Slice.t -> 'a) -> 'a t * 'a
+    val create : ?message_size:int -> (Capnp.Message.rw Slice.t -> 'a) -> 'a t * 'a
     (** [create init] is a fresh response and contents builder.
         Use one of the generated [init_pointer] functions for [init]. *)
 

capnp-rpc-lwt/msg.ml

@@ -106,7 +106,7 @@ module Response = struct
     | _ -> assert false
 
   let bootstrap () =
-    let msg = B.Message.init_root () in
+    let msg = B.Message.init_root ~message_size:150 () in
     let ret = B.Message.return_init msg in
     let p = B.Return.results_init ret in
     B.Payload.content_set_interface p (Some Stdint.Uint32.zero);   (* Cap index 0 *)

capnp-rpc-lwt/request.ml

@@ -5,17 +5,17 @@ module StructStorage = Capnp.Message.BytesMessage.StructStorage
 
 type 'a t = Message.t
 
-let create init =
+let create ?message_size init =
   let msg =
-    Message.init_root ()
+    Message.init_root ?message_size ()
     |> StructStorage.with_attachments (Msg.wrap_attachments (Core_types.Attachments.builder ())) in
   let call = Message.call_init msg in
   let p = Call.params_get call in
   let content = init (Payload.content_get p) in
   msg, content
 
 let create_no_args () =
-  let msg = Message.init_root () in
+  let msg = Message.init_root ~message_size:200 () in
   ignore (Message.call_init msg);
   msg
 

capnp-rpc-lwt/request.mli

@@ -2,7 +2,7 @@ type 'a t
 
 module RO_array = Capnp_rpc.RO_array
 
-val create : (Capnp.Message.rw Capnp.BytesMessage.Slice.t -> 'a) -> 'a t * 'a
+val create : ?message_size:int -> (Capnp.Message.rw Capnp.BytesMessage.Slice.t -> 'a) -> 'a t * 'a
 val create_no_args : unit -> 'a t
 val finish : (_, 'a, _) Capnp.RPC.MethodID.t -> 'a t -> Msg.Request.t
 val release : 'a t -> unit

capnp-rpc-lwt/response.ml

@@ -7,17 +7,17 @@ type 'a cap = Core_types.cap
 
 type 'a t = Message.t
 
-let create init =
+let create ?message_size init =
   let msg =
-    Message.init_root ()
+    Message.init_root ?message_size ()
     |> StructStorage.with_attachments (Msg.wrap_attachments (Core_types.Attachments.builder ())) in
   let ret = Message.return_init msg in
   let p = Return.results_init ret in
   let content = init (Payload.content_get p) in
   msg, content
 
 let create_empty () =
-  let msg = Message.init_root () in
+  let msg = Message.init_root ~message_size:100 () in
   let _ = Message.return_init msg in
   msg
 

capnp-rpc-net/capTP_capnp.ml

@@ -100,6 +100,12 @@ module Make (Network : S.NETWORK) = struct
 
   (* Enqueue [message] in [xmit_queue] and ensure the flush thread is running. *)
   let queue_send ~xmit_queue endpoint message =
+    Log.debug (fun f ->
+        let module M = Capnp_rpc_lwt.Private.Schema.MessageWrapper.Message in
+        f "queue_send: %d/%d allocated bytes in %d segs"
+                  (M.total_size message)
+                  (M.total_alloc_size message)
+                  (M.num_segments message));
     let was_idle = Queue.is_empty xmit_queue in
     Queue.add message xmit_queue;
     Prometheus.Counter.inc_one Metrics.messages_outbound_enqueued_total;

capnp-rpc-net/serialise.ml

@@ -67,7 +67,7 @@ module Make (EP : Capnp_core.ENDPOINT) = struct
       write_exn (Message.abort_init b) ex;
       Message.to_message b
     | `Bootstrap (qid, object_id) ->
-      let b = Message.init_root () in
+      let b = Message.init_root ~message_size:100 () in
       let boot = Message.bootstrap_init b in
       Bootstrap.question_id_set boot (QuestionId.uint32 qid);
       Schema.BuilderOps.write_string (Bootstrap.deprecated_object_id_get boot) object_id;
@@ -86,19 +86,19 @@ module Make (EP : Capnp_core.ENDPOINT) = struct
       end;
       Call.to_message c
     | `Finish (qid, release_result_caps) ->
-      let b = Message.init_root () in
+      let b = Message.init_root ~message_size:42 () in
       let fin = Message.finish_init b in
       Finish.question_id_set fin (QuestionId.uint32 qid);
       Finish.release_result_caps_set fin release_result_caps;
       Message.to_message b
     | `Release (id, count) ->
-      let m = Message.init_root () in
+      let m = Message.init_root ~message_size:48 () in
       let rel = Message.release_init m in
       Release.id_set rel (ImportId.uint32 id);
       Release.reference_count_set_int_exn rel count;
       Message.to_message m
     | `Disembargo_request disembargo_request ->
-      let m = Message.init_root () in
+      let m = Message.init_root ~message_size:200 () in
       let dis = Message.disembargo_init m in
       let ctx = Disembargo.context_init dis in
       begin match disembargo_request with
@@ -108,7 +108,7 @@ module Make (EP : Capnp_core.ENDPOINT) = struct
       end;
       Message.to_message m
     | `Disembargo_reply (target, embargo_id) ->
-      let m = Message.init_root () in
+      let m = Message.init_root ~message_size:200 () in
       let dis = Message.disembargo_init m in
       let ctx = Disembargo.context_init dis in
       set_target (Disembargo.target_init dis) target;
@@ -128,17 +128,17 @@ module Make (EP : Capnp_core.ENDPOINT) = struct
           write_exn (Return.exception_init ret) ex;
           ret
         | `Cancelled ->
-          let m = Message.init_root () in
+          let m = Message.init_root ~message_size:200 () in
           let ret = Message.return_init m in
           Return.canceled_set ret;
           ret
         | `ResultsSentElsewhere ->
-          let m = Message.init_root () in
+          let m = Message.init_root ~message_size:200 () in
           let ret = Message.return_init m in
           Return.results_sent_elsewhere_set ret;
           ret
         | `TakeFromOtherQuestion qid ->
-          let m = Message.init_root () in
+          let m = Message.init_root ~message_size:200 () in
           let ret = Message.return_init m in
           Return.take_from_other_question_set ret (QuestionId.uint32 qid);
           ret

examples/calc.ml

@@ -70,7 +70,7 @@ let evaluate t expr =
 let getOperator t op =
   let open C.GetOperator in
   let module O = Api.Builder.Calculator.Operator in
-  let req, p = Capability.Request.create Params.init_pointer in
+  let req, p = Capability.Request.create ~message_size:200 Params.init_pointer in
   Params.op_set p (match op with
       | `Add -> O.Add
       | `Subtract -> O.Subtract
@@ -104,7 +104,7 @@ module Value = struct
       method read_impl _ release_params =
         let open Value.Read in
         release_params ();
-        let resp, c = Service.Response.create Results.init_pointer in
+        let resp, c = Service.Response.create ~message_size:200 Results.init_pointer in
         Results.value_set c f;
         Service.return resp
     end
@@ -173,7 +173,7 @@ module Fn = struct
         (* Functions return floats, not Value objects, so we have to wait here. *)
         Service.return_lwt (fun () ->
             Value.final_read value >|= fun value ->
-            let resp, r = Service.Response.create Results.init_pointer in
+            let resp, r = Service.Response.create ~message_size:200 Results.init_pointer in
             Results.value_set r value;
             Ok resp
           )
@@ -216,7 +216,7 @@ let local =
       let fn_obj = Fn.local n_args body in
       Expr.release body;
       release_params ();
-      let resp, results = Service.Response.create Results.init_pointer in
+      let resp, results = Service.Response.create ~message_size:200 Results.init_pointer in
       Results.func_set results (Some fn_obj);
       Service.return resp
 
@@ -226,7 +226,7 @@ let local =
       release_params ();
       let value_obj = eval expr in
       Expr.release expr;
-      let resp, results = Service.Response.create Results.init_pointer in
+      let resp, results = Service.Response.create ~message_size:200 Results.init_pointer in
       Results.value_set results (Some value_obj);
       Capability.dec_ref value_obj;
       Service.return resp
@@ -243,7 +243,7 @@ let local =
         | O.Divide      -> Fn.div
         | O.Undefined _ -> failwith "Unknown operator"
       in
-      let resp, results = Service.Response.create Results.init_pointer in
+      let resp, results = Service.Response.create ~message_size:200 Results.init_pointer in
       Results.func_set results (Some op_obj);
       Service.return resp
   end

test-bin/calc.ml

@@ -82,7 +82,11 @@ let cmds = [serve_cmd; connect_cmd]
 let () =
   Fmt_tty.setup_std_outputs ();
   Logs.set_reporter reporter;
-  Logs.set_level ~all:true (Some Logs.Debug);
+  Logs.set_level ~all:true (Some Logs.Info);
+  Logs.Src.list () |> List.iter (fun src ->
+      if Astring.String.is_prefix ~affix:"capnp" (Logs.Src.name src) then
+        Logs.Src.set_level src (Some Logs.Debug);
+    );
   match Term.eval_choice ~catch:false default_cmd cmds with
   | exception Failure msg -> Fmt.epr "%s@." msg; exit 1
   | status -> Term.exit status

test-bin/echo/echo.ml

@@ -14,8 +14,10 @@ let local =
       let open Echo.Ping in
       let msg = Params.msg_get params in
       release_param_caps ();
-      let response, results = Service.Response.create Results.init_pointer in
-      Results.reply_set results ("echo:" ^ msg);
+      let resp = "echo:" ^ msg in
+      let message_size = 100 + String.length resp in
+      let response, results = Service.Response.create ~message_size Results.init_pointer in
+      Results.reply_set results resp;
       Service.return response
   end
 
@@ -24,6 +26,7 @@ module Echo = Api.Client.Echo
 
 let ping t msg =
   let open Echo.Ping in
-  let request, params = Capability.Request.create Params.init_pointer in
+  let message_size = 200 + String.length msg in  (* (rough estimate) *)
+  let request, params = Capability.Request.create ~message_size Params.init_pointer in
   Params.msg_set params msg;
   Capability.call_for_value_exn t method_id request >|= Results.reply_get

test-bin/echo/echo_bench.ml

@@ -17,7 +17,7 @@ let run_client service =
         assert (res = desired_result)
     ) in
   let st = Unix.gettimeofday () in
-  Lwt_list.iter_p (fun v -> v ()) ops >>= fun () ->
+  Lwt_stream.of_list ops |> Lwt_stream.iter_n ~max_concurrency:12 (fun v -> v ()) >>= fun () ->
   let ed = Unix.gettimeofday () in 
   let rate = (Int.to_float n) /. (ed -. st) in
   Logs.info (fun m -> m "rate = %f" rate );