Got dispatch unit tests working, not happy about how epoll timeouts work though

dispatch.ml

@@ -5,7 +5,7 @@ type fd_handler = Unix.file_descr -> event list -> unit
 module Timer =
   Set.Make (struct
               type t = (float * timer_handler)
-              let compare = compare
+              let compare (time, handler) (time', handler') = compare time time'
             end)
 
 module Fd_map =
@@ -21,7 +21,6 @@ type t = {
   timers : Timer.t ref;
 }
 
-
 let to_epoll = function
   | Input -> Epoll.In
   | Priority -> Epoll.Priority
@@ -83,7 +82,7 @@ let delete_timer d time =
 let rec dispatch_timers d now =
   if (!(d.timers) != Timer.empty) then
     let (time, handler) = Timer.min_elt !(d.timers) in
-      if now > time then
+      if now < time then
         ()
       else begin
         handler time;
@@ -102,19 +101,25 @@ let rec dispatch_results d events_list =
           dispatch_results d tl
 
 let once d =
-  let now = Unix.time () in
+  let now = Unix.gettimeofday () in
   let timeout =
     try
       let (time, _) = Timer.min_elt !(d.timers) in
-      let timeout_s = max (time -. now) 0.0 in
+      let delta = (time -. now) in
-        int_of_float (timeout_s *. 1000.0)
+	max delta 0.0
     with Not_found ->
-      -1
+      (-1.0)
   in
-  let result = Epoll.wait d.e !(d.numfds) timeout in
+    (if !(d.numfds) = 0 then
-    dispatch_timers d (Unix.time ());
+       (* epoll()--and probably poll()--barfs if it has no file descriptors *)
-    dispatch_results d result
+       ignore (Unix.select [] [] [] timeout)
-
+     else
+       (* poll() and epoll() wait *at most* timeout ms.  If you have fds but they're not
+	  doing anything, multiple calls to once may be required.  This is lame. *)
+       let timeout_ms = int_of_float (timeout *. 1000.0) in
+       let result = Epoll.wait d.e !(d.numfds) timeout_ms in
+	 dispatch_results d result);
+    dispatch_timers d (Unix.gettimeofday ())
 
 let rec run d =    
   if ((!(d.fds) == Fd_map.empty) &&

tests.ml

@@ -2,6 +2,9 @@ open Unixqueue
 open OUnit
 open Irc
 
+let dump x =
+  Printf.ksprintf (fun str -> prerr_string str; flush stderr) x
+
 (* Return true iff str starts with substr *)
 let startswith str substr =
   let l = String.length substr in
@@ -67,31 +70,54 @@ type chat_event =
 exception Chat_match of (string * chat_event)
 exception Chat_timeout of chat_event
 
+let string_of_chat_event e =
+  match e with
+    | Send str ->
+        ("Send (\"" ^ (String.escaped str) ^ "\")")
+    | Recv str ->
+        ("Recv (\"" ^ (String.escaped str) ^ "\")")
+    | Regex str ->
+        ("Regex (\"" ^ (String.escaped str) ^ "\")")
+
 (* Return a [Dispatch.fd_handler] function to run script [s] *)
 let chat d fd s =
   let script = ref s in
-  let timer = (Unix.time ()) +. 1.0 in
+    (* Add some amount, dependent on fd, to the timeout value, so peers won't obliterate it *)
+  let timer = (Unix.time ()) +. 1.0 +. (0.01 *. (float_of_int (int_of_file_descr fd))) in
   let obuf = Buffer.create 4096 in
   let ibuf = Buffer.create 4096 in
   let handle_timer _ =
-    raise (Chat_timeout (List.hd !script))
+    failwith (Printf.sprintf "fd=%d timeout waiting for %s" 
+		(int_of_file_descr fd)
+		(string_of_chat_event (List.hd !script)))
+  in
+  let nomatch got =
+    failwith (Printf.sprintf "fd=%d expecting %s\n got %s"
+		(int_of_file_descr fd)
+		(string_of_chat_event (List.hd !script))
+		(String.escaped got))
   in
   let rec run_script fd =
     match !script with
       | [] ->
+	  if ((Buffer.length obuf) = 0) then begin
             Dispatch.delete_timer d timer;
-          Dispatch.delete d fd;
+	    (try
+	       Dispatch.delete d fd
+	     with (Failure _) ->
+	       ());
             Unix.close fd
+	  end
       | Send buf :: tl ->
           Buffer.add_string obuf buf;
+	  Dispatch.modify d fd [Dispatch.Input; Dispatch.Output];
           script := tl;
           run_script fd
       | Recv buf :: tl ->
           let buf_len = String.length buf in
           let ibuf_str = Buffer.contents ibuf in
-            if (Buffer.length ibuf >= buf_len) then
+            if ((Buffer.length ibuf) >= buf_len) then begin
-              if startswith ibuf_str buf then
+              if startswith ibuf_str buf then begin
-                begin
                 script := tl;
                 Buffer.clear ibuf;
                 Buffer.add_substring
@@ -100,10 +126,9 @@ let chat d fd s =
                   buf_len
                   ((String.length ibuf_str) - buf_len);
                 run_script fd
-                end
+              end else
-              else
+		nomatch ibuf_str
-                raise (Chat_match (ibuf_str, Recv buf))
+	    end else
-            else
               ()
       | Regex buf :: tl ->
           let ibuf_str = Buffer.contents ibuf in
@@ -120,9 +145,10 @@ let chat d fd s =
                     ((String.length ibuf_str) - match_len);
                   run_script fd
               else
-                raise (Chat_match (ibuf_str, Regex buf))
+		nomatch ibuf_str
             else
               ()
+
   in
   let rec handler fd events =
     match events with
@@ -132,8 +158,10 @@ let chat d fd s =
           let s = String.create 4096 in
           let n = Unix.read fd s 0 4096 in
             Buffer.add_substring ibuf s 0 n;
-            run_script fd
+            run_script fd;
+	    handler fd tl
       | Dispatch.Output :: tl ->
+	  begin
             if ((Buffer.length obuf) = 0) then
               Dispatch.modify d fd [Dispatch.Input]
             else
@@ -142,6 +170,14 @@ let chat d fd s =
               let n = Unix.write fd ostr 0 olen in
 		Buffer.clear obuf;
 		Buffer.add_substring obuf ostr n (olen - n)
+	  end;
+	  handler fd tl
+      | Dispatch.Hangup :: tl ->
+	  (* Stop listening to this fd, it will always return Hangup *)
+	  (try
+	     Dispatch.delete d fd
+	   with (Failure _) ->
+	     ())
       | _ ->
           failwith "Unexpected event"
   in
@@ -222,19 +258,36 @@ let unit_tests =
            last_timer := time
          in
 
+	 let s = String.create 4096 in
+
            assert_equal 8 (Unix.write a "dispatch" 0 8);
            Dispatch.add d b handle [Dispatch.Input; Dispatch.Output];
            Dispatch.once d;
            assert_equal (b, [Dispatch.Input; Dispatch.Output]) !last_event;
-           (let s = String.create 4096 in
            assert_equal 8 (Unix.read b s 0 4096);
-              assert_equal "dispatch" (Str.string_before s 8));
+           assert_equal "dispatch" (Str.string_before s 8);
 
-           (let time = ((Unix.time ()) +. 0.1) in
+           (let time = ((Unix.gettimeofday ()) +. 0.01) in
               Dispatch.add_timer d handle_timer time;
-              assert_equal 0.0 !last_timer;
+              Dispatch.add_timer d handle_timer ((Unix.gettimeofday ()) +. 10.0);
+
+              assert_equal ~printer:string_of_float 0.0 !last_timer;
               Dispatch.once d;
-              assert_equal time !last_timer);
+              assert_equal ~printer:string_of_float 0.0 !last_timer;
+
+	      Dispatch.modify d b [Dispatch.Input];
+	      while !last_timer = 0.0 do
+		Dispatch.once d
+	      done;
+              assert_equal ~printer:string_of_float time !last_timer;
+
+	      Dispatch.modify d b [Dispatch.Input; Dispatch.Output];
+	      assert_equal 6 (Unix.write a "gnarly" 0 6);
+              Dispatch.once d;
+              assert_equal (b, [Dispatch.Input; Dispatch.Output]) !last_event;
+              assert_equal 6 (Unix.read b s 0 4096);
+
+              assert_equal ~printer:string_of_float time !last_timer);
 
            Dispatch.once d;
            assert_equal (b, [Dispatch.Output]) !last_event;