Keep track of transmit queue lengths

Log if we have to wait to send a frame.

client_net.ml

@@ -11,13 +11,16 @@ let src = Logs.Src.create "net" ~doc:"Client networking"
 module Log = (val Logs.src_log src : Logs.LOG)
 
 class client_iface eth ~gateway_ip ~client_ip client_mac : client_link = object
+  val queue = FrameQ.create (Ipaddr.V4.to_string client_ip)
   method my_mac = ClientEth.mac eth
   method other_mac = client_mac
   method my_ip = gateway_ip
   method other_ip = client_ip
   method writev ip =
-    let eth_hdr = eth_header_ipv4 ~src:(ClientEth.mac eth) ~dst:client_mac in
-    ClientEth.writev eth (fixup_checksums (Cstruct.concat (eth_hdr :: ip)))
+    FrameQ.send queue (fun () ->
+      let eth_hdr = eth_header_ipv4 ~src:(ClientEth.mac eth) ~dst:client_mac in
+      ClientEth.writev eth (fixup_checksums (Cstruct.concat (eth_hdr :: ip)))
+    )
 end
 
 let clients : Cleanup.t IntMap.t ref = ref IntMap.empty

frameQ.ml

@@ -0,0 +1,25 @@
+(* Copyright (C) 2016, Thomas Leonard <thomas.leonard@unikernel.com>
+   See the README file for details. *)
+
+let src = Logs.Src.create "frameQ" ~doc:"Interface output queue"
+module Log = (val Logs.src_log src : Logs.LOG)
+
+type t = {
+  name : string;
+  mutable items : int;
+}
+
+let create name = { name; items = 0 }
+  
+let send q fn =
+  (* TODO: drop if queue too long *)
+  let sent = fn () in
+  if Lwt.state sent = Lwt.Sleep then (
+    q.items <- q.items + 1;
+    Log.info (fun f -> f "Queue length for %s: incr to %d" q.name q.items);
+    Lwt.on_termination sent (fun () ->
+      q.items <- q.items - 1;
+      Log.info (fun f -> f "Queue length for %s: decr to %d" q.name q.items);
+    )
+  );
+  sent

frameQ.mli

@@ -0,0 +1,15 @@
+(* Copyright (C) 2016, Thomas Leonard <thomas.leonard@unikernel.com>
+   See the README file for details. *)
+
+(** Keep track of the queue length for output buffers. *)
+
+type t
+
+val create : string -> t
+(** [create name] is a new empty queue. [name] is used in log messages. *)
+
+val send : t -> (unit -> 'a Lwt.t) -> 'a Lwt.t
+(** [send t fn] checks that the queue isn't overloaded and calls [fn ()] if it's OK.
+    The item is considered to be queued until the result of [fn] has resolved.
+    In the case of mirage-net-xen's [writev], this happens when the frame has been
+    added to the ring (not when it is consumed), which is fine for us. *)

uplink.ml

@@ -21,13 +21,16 @@ module Make(Clock : V1.CLOCK) = struct
   }
 
   class netvm_iface eth mac ~my_ip ~other_ip : interface = object
+    val queue = FrameQ.create (Ipaddr.V4.to_string other_ip)
     method my_mac = Eth.mac eth
     method my_ip = my_ip
     method other_ip = other_ip
     method writev ip =
-      mac >>= fun dst ->
-      let eth_hdr = eth_header_ipv4 ~src:(Eth.mac eth) ~dst in
-      Eth.writev eth (eth_hdr :: ip)
+      FrameQ.send queue (fun () ->
+        mac >>= fun dst ->
+        let eth_hdr = eth_header_ipv4 ~src:(Eth.mac eth) ~dst in
+        Eth.writev eth (eth_hdr :: ip)
+      )
   end
 
   let listen t router =