Merge pull request #15 from talex5/wait_for_free

Add Front.wait_for_free

lwt/lwt_ring.ml

@@ -44,6 +44,18 @@ module Front = struct
       get_free_slot t
     end 
 
+  let rec wait_for_free t n =
+    if Ring.Rpc.Front.get_free_requests t.ring >= n then
+      return ()
+    else begin
+      assert (n <= Ring.Rpc.Front.nr_ents t.ring);
+      let th, u = MProf.Trace.named_task "ring.wait_for_free" in
+      let node = Lwt_sequence.add_r u t.waiters in
+      Lwt.on_cancel th (fun _ -> Lwt_sequence.remove node);
+      lwt () = th in
+      wait_for_free t n
+    end
+
   let poll t respfn =
     Ring.Rpc.Front.ack_responses t.ring (fun slot ->
         MProf.Trace.label "ring.poll ack_response";

lwt/lwt_ring.mli

@@ -61,6 +61,12 @@ module Front : sig
 
   val shutdown : ('a, 'b) t -> unit
 
+  val wait_for_free : ('a, 'b) t -> int -> unit Lwt.t
+  (** [wait_for_free frontend n] waits until at least [n] slots are free.
+   * It doesn't reserve them, so you'll probably need to use your own mutex
+   * around this call. Requests are handled in order, so even large requests
+   * will eventually be served. *)
+
   val to_string : ('a,'b) t -> string
   (** [to_string t] returns debug-printable description of the ring
       metadata *)