buffer statistics

src/tcp/flow.ml

@@ -58,6 +58,9 @@ struct
     keepalive: KEEPALIVE.t option; (* Optional TCP keepalive state *)
   }
 
+  let buffer_sizes { rxq ; txq ; urx ; utx ; _ } =
+    RXS.size rxq, TXS.size txq, User_buffer.Rx.size urx, UTX.size utx
+
   type flow = pcb
   type connection = flow * unit Lwt.t
 
@@ -86,13 +89,30 @@ struct
   let _pp_pcb fmt pcb =
     Format.fprintf fmt "id=[%a] state=[%a]" WIRE.pp pcb.id State.pp pcb.state
 
+  let pp_sizes fmt (rx, tx, urx, utx) =
+    Format.fprintf fmt "RX %d TX %d URX %d UTX %d TOTAL %d" rx tx urx utx
+      (rx + tx + urx + utx)
+
   let pp_stats fmt t =
-    Format.fprintf fmt "[channels=%d listens=%d connects=%d]"
+    let add_sizes (a, b, c, d) (a', b', c', d') =
+      a + a', b + b', c + c', d + d'
+    in
+    let ch_size =
+      Hashtbl.fold (fun _ (c, _) acc -> add_sizes (buffer_sizes c) acc)
+        t.channels (0, 0, 0, 0)
+    and l_size =
+      Hashtbl.fold (fun _ (_, (_, (c, _))) acc -> add_sizes (buffer_sizes c) acc)
+        t.listens (0, 0, 0, 0)
+    in
+    Format.fprintf fmt "[channels=%d (%a) listens=%d (%a) connects=%d]"
       (Hashtbl.length t.channels)
+      pp_sizes ch_size
       (Hashtbl.length t.listens)
+      pp_sizes l_size
       (Hashtbl.length t.connects)
 
-  let log_with_stats name t = Log.debug (fun fmt -> fmt "%s: %a" name pp_stats t)
+  let log_with_stats name t =
+    Log.debug (fun fmt -> fmt "%s: %a" name pp_stats t)
 
   let wscale_default = 2
 

src/tcp/segment.ml

@@ -87,6 +87,9 @@ module Rx(Time:Mirage_time.S) = struct
     state: State.t;
   }
 
+  let size { segs ; _ } =
+    S.fold (fun seg acc -> Cstruct.length seg.payload + acc) segs 0
+
   let create ~rx_data ~wnd ~state ~tx_ack =
     let segs = S.empty in
     { segs; rx_data; tx_ack; wnd; state }
@@ -276,6 +279,11 @@ module Tx (Time:Mirage_time.S) (Clock:Mirage_clock.MCLOCK) = struct
 
   type t = T: ('a, 'b) q -> t
 
+  let size (T q) =
+    Lwt_dllist.fold_l (fun seg acc ->
+        Cstruct.length seg.data + acc)
+      q.segs 0
+
   let ack_segment _ _ = ()
   (* Take any action to the user transmit queue due to this being
      successfully ACKed *)

src/tcp/segment.mli

@@ -36,6 +36,8 @@ module Rx (T:Mirage_time.S) : sig
 
   val pp: Format.formatter -> t -> unit
 
+  val size : t -> int
+
   val create:
     rx_data:(Cstruct.t list option * Sequence.t option) Lwt_mvar.t ->
     wnd:Window.t ->
@@ -64,6 +66,8 @@ module Tx (Time:Mirage_time.S)(Clock:Mirage_clock.MCLOCK) : sig
   type t
   (** Queue of pre-transmission segments *)
 
+  val size : t -> int
+
   val create:
     xmit:('a, 'b) xmit -> wnd:Window.t -> state:State.t ->
     rx_ack:Sequence.t Lwt_mvar.t ->

src/tcp/user_buffer.ml

@@ -35,10 +35,15 @@ module Rx = struct
     writers: unit Lwt.u Lwt_dllist.t;
     readers: Cstruct.t option Lwt.u Lwt_dllist.t;
     mutable watcher: int32 Lwt_mvar.t option;
-    mutable max_size: int32;
+    max_size: int32;
     mutable cur_size: int32;
   }
 
+  let size { q ; _ } =
+    Lwt_dllist.fold_l (fun cs_opt acc ->
+        (match cs_opt with None -> 0 | Some x -> Cstruct.length x) + acc)
+      q 0
+
   let create ~max_size ~wnd =
     let q = Lwt_dllist.create () in
     let writers = Lwt_dllist.create () in
@@ -122,6 +127,9 @@ module Tx(Time:Mirage_time.S)(Clock:Mirage_clock.MCLOCK) = struct
     mutable bufbytes: int32;
   }
 
+  let size { buffer ; _ } =
+    Lwt_dllist.fold_l (fun buf acc -> Cstruct.length buf + acc) buffer 0
+
   let create ~max_size ~wnd ~txq =
     let buffer = Lwt_dllist.create () in
     let writers = Lwt_dllist.create () in

src/tcp/user_buffer.mli

@@ -24,6 +24,8 @@ module Rx : sig
   val cur_size : t -> int32
   val max_size : t -> int32
   val monitor: t -> int32 Lwt_mvar.t -> unit
+
+  val size : t -> int
 end
 
 module Tx(Time:Mirage_time.S)(Clock:Mirage_clock.MCLOCK) : sig
@@ -36,6 +38,8 @@ module Tx(Time:Mirage_time.S)(Clock:Mirage_clock.MCLOCK) : sig
       Cstruct.t -> unit Lwt.t
   end
 
+  val size : t -> int
+
   val create: max_size:int32 -> wnd:Window.t -> txq:TXS.t -> t
   val available: t -> int32
   val wait_for: t -> int32 -> unit Lwt.t