Complete transition to epoll, all unit tests passing

OMakefile

@@ -1,6 +1,4 @@
 OCAMLPACKS[] = 
-    equeue
-    pcre
     str
 OCAML_CLIBS = ocamlepoll
 OCAMLCFLAGS += -g
@@ -26,7 +24,7 @@ section
     dispatch_tests.cmo:
     dispatch_tests$(EXT_OBJ):
 
-    OCamlProgram(tests, tests dispatch chat irc command iobuf client channel)
+    OCamlProgram(tests, tests dispatch irc command iobuf client channel)
 
 .PHONY: test
 test: tests

client.ml

@@ -40,7 +40,7 @@ let reply cli num ?(args=[]) text =
                ([!(cli.nick)] @ args)
                (Some text))
 
-let handle_close cli () =
+let handle_close cli message =
   Hashtbl.remove by_nick !(cli.nick)
 
 let handle_command cli iobuf cmd =
@@ -161,22 +161,22 @@ let set_nick cli nick =
   Hashtbl.replace by_nick nick cli;
   cli.nick := nick
 
-let rec handle_command_prereg (nick', username', realname', password') iobuf cmd =
+let rec handle_command_prereg (nick, username, realname, password) iobuf cmd =
   (* Handle a command during the login phase *)
   let acc =
     match (Command.as_tuple cmd) with
-      | (None, "PASS", [password], None) ->
+      | (None, "PASS", [password'], None) ->
-          (nick', username', realname', Some password)
+          (nick, username, realname, Some password')
-      | (None, "USER", [username; _; _], Some realname) ->
+      | (None, "USER", [username'; _; _], Some realname') ->
-          (nick', Some username, Some (Irc.truncate realname 40), password')
+          (nick, Some username', Some (Irc.truncate realname' 40), password)
-      | (None, "NICK", [nick], None) ->
+      | (None, "NICK", [nick'], None) ->
-          (Some nick, username', realname', password')
+          (Some nick', username, realname, password)
       | _ ->
           Iobuf.write iobuf (Command.create
                                (Some !(Irc.name)) 
                                "451" ["*"] 
                                (Some "Register first."));
-          (nick', username', realname', password')
+          (nick, username, realname, password)
   in
   let welcome cli =
     try
@@ -211,8 +211,8 @@ let rec handle_command_prereg (nick', username', realname', password') iobuf cmd
     | _ ->
         Iobuf.rebind iobuf (handle_command_prereg acc) ignore
 
-let handle_connection ues grp fd =
+let handle_connection d fd addr =
   let command_handler = handle_command_prereg (None, None, None, None) in
   let close_handler = ignore in
-    Iobuf.bind ues grp fd command_handler close_handler
+    Iobuf.bind d fd command_handler close_handler
     

client.mli

@@ -1,5 +1,5 @@
 type t
 
 val write : t -> Command.t -> unit
-val handle_connection : Unixqueue.event_system -> Unixqueue.group -> Unix.file_descr -> unit
+val handle_connection : Dispatch.t -> Unix.file_descr -> Unix.sockaddr -> unit
   

command.ml

@@ -39,7 +39,7 @@ let extract_word s =
 let rec from_string line =
   (* Very simple.  Pull out words until you get one starting with ":".
      The very first word might start with ":", that doesn't count
-     because it's the sender.. *)
+     because it's the sender. *)
   let rec loop sender acc line =
     let c = (if (line = "") then None else (Some line.[0])) in
       match (c, acc) with

iobuf.ml

@@ -1,8 +1,7 @@
-(* ==========================================
+(* **************************************
- * I/O buf stuff
+ * IRC Command I/O buffers
- *)
+ * **************************************)
-type t = {ues: Unixqueue.event_system;
+type t = {d: Dispatch.t;
-          grp: Unixqueue.group;
           fd: Unix.file_descr;
           outq: Command.t Queue.t;
           unsent: string ref;
@@ -10,39 +9,24 @@ type t = {ues: Unixqueue.event_system;
           ibuf_len: int ref;
           addr: string;
           command_handler: (t -> Command.t -> unit) ref;
-	  close_handler: (unit -> unit) ref}
+	  close_handler: (string -> unit) ref}
 
 let ibuf_max = 4096
 let max_outq = 50
 let obuf_max = 4096
 
-let by_file_descr = Hashtbl.create 25
-
 let addr iobuf = iobuf.addr
 
 let write iobuf cmd =
   let was_empty = Queue.is_empty iobuf.outq in
     Queue.add cmd iobuf.outq;
     if (was_empty && (!(iobuf.unsent) = ""))  then
-      Unixqueue.add_resource
+      Dispatch.modify iobuf.d iobuf.fd [Dispatch.Input; Dispatch.Output]
-        iobuf.ues iobuf.grp (Unixqueue.Wait_out iobuf.fd, -.1.0)
 
-let close iobuf =
+let close iobuf message =
-  !(iobuf.close_handler) ();
+  !(iobuf.close_handler) message;
-  Hashtbl.remove by_file_descr iobuf.fd;
+  Dispatch.delete iobuf.d iobuf.fd;
-  Unix.close iobuf.fd;
+  Unix.close iobuf.fd
-  Unixqueue.remove_resource iobuf.ues iobuf.grp (Unixqueue.Wait_in iobuf.fd);
-  try
-    Unixqueue.remove_resource iobuf.ues iobuf.grp (Unixqueue.Wait_out iobuf.fd);
-  with Not_found ->
-    ()
-
-let handle_close fd =
-  try
-    let iobuf = Hashtbl.find by_file_descr fd in
-      close iobuf
-  with Not_found ->
-    ()
 
 let crlf = Str.regexp "\r?\n"
 
@@ -63,27 +47,22 @@ let handle_input iobuf =
   in
     loop lines
 
-let handle_event ues esys e =
+let rec handle_events iobuf fd events =
-  match e with
+  match events with
-    | Unixqueue.Input_arrived (g, fd) ->
+    | [] ->
-        let iobuf = Hashtbl.find by_file_descr fd in
+	()
+    | Dispatch.Input :: tl ->
         let size = ibuf_max - !(iobuf.ibuf_len) in
         let len = Unix.read fd iobuf.ibuf !(iobuf.ibuf_len) size in
-          if (len > 0) then
-            begin
           iobuf.ibuf_len := !(iobuf.ibuf_len) + len;
-              try
+          handle_input iobuf;
-                handle_input iobuf
-              with Not_found ->
           if (!(iobuf.ibuf_len) = ibuf_max) then
             (* No newline found, and the buffer is full *)
-                  raise (Failure "Buffer overrun");
+	    close iobuf "Input buffer overrun"
-            end
 	  else
-            close iobuf
+	    handle_events iobuf fd tl
-    | Unixqueue.Output_readiness (g, fd) ->
+    | Dispatch.Output :: tl ->
         (* XXX: Could be optimized to try and fill the output buffer *)
-        let iobuf = Hashtbl.find by_file_descr fd in
         let buf =
           if (!(iobuf.unsent) = "") then
             let cmd = Queue.pop iobuf.outq in
@@ -96,17 +75,19 @@ let handle_event ues esys e =
 	  if n < buflen then
             iobuf.unsent := Str.string_after buf n
 	  else if Queue.is_empty iobuf.outq then
-	    Unixqueue.remove_resource ues g (Unixqueue.Wait_out fd)
+	    Dispatch.modify iobuf.d fd [Dispatch.Input];
-    | Unixqueue.Out_of_band (g, fd) ->
+	  handle_events iobuf fd tl
-        print_endline "oob"
+    | Dispatch.Priority :: tl ->
-    | Unixqueue.Timeout (g, op) ->
+	let s = String.create 4096 in
-        print_endline "timeout"
+	  ignore (Unix.recv fd s 0 4096 [Unix.MSG_OOB]);
-    | Unixqueue.Signal ->
+	  handle_events iobuf fd tl
-        print_endline "signal"
+    | Dispatch.Error :: tl ->
-    | Unixqueue.Extra exn ->
+	close iobuf "Error"
-        print_endline "extra"
+    | Dispatch.Hangup :: tl ->
+	close iobuf "Hangup"
 
-let bind ues grp fd command_handler close_handler =
+
+let bind d fd command_handler close_handler =
   let (outq, unsent, ibuf, ibuf_len) =
     (Queue.create (), ref "", String.create ibuf_max, ref 0)
   in
@@ -117,8 +98,7 @@ let bind ues grp fd command_handler close_handler =
       | Unix.ADDR_INET (addr, port) ->
           Unix.string_of_inet_addr addr
   in
-  let iobuf = {ues = ues;
+  let iobuf = {d = d;
-               grp = grp;
                fd = fd;
                outq = outq;
                unsent = unsent;
@@ -128,9 +108,7 @@ let bind ues grp fd command_handler close_handler =
                command_handler = ref command_handler;
 	       close_handler = ref close_handler}
   in
-    Hashtbl.replace by_file_descr fd iobuf;
+    Dispatch.add d fd (handle_events iobuf) [Dispatch.Input]
-    Unixqueue.add_resource ues grp (Unixqueue.Wait_in fd, -.1.0);
-    Unixqueue.add_close_action ues grp (fd, handle_close)
 
 let rebind t command_handler close_handler =
   t.command_handler := command_handler;

iobuf.mli

@@ -3,7 +3,6 @@ type t
 val addr : t -> string
 
 val write : t -> Command.t -> unit
-val bind : Unixqueue.event_system -> Unixqueue.group -> Unix.file_descr -> (t -> Command.t -> unit) -> (unit -> unit) -> unit
+val bind : Dispatch.t -> Unix.file_descr -> (t -> Command.t -> unit) -> (string -> unit) -> unit
-val rebind: t -> (t -> Command.t -> unit) -> (unit -> unit) -> unit
+val rebind: t -> (t -> Command.t -> unit) -> (string -> unit) -> unit
-val close: t -> unit
+val close: t -> string -> unit
-val handle_event : Unixqueue.event_system -> Unixqueue.event Equeue.t -> Unixqueue.event -> unit

irc.ml

@@ -1,10 +1,6 @@
 let name = ref "irc.test"
 let version = "0.1"
 
-let newline_re = Pcre.regexp "\n\r?"
-let argsep_re = Pcre.regexp " :"
-let space_re = Pcre.regexp " "
-
 let dbg msg a =
   prerr_endline ("[" ^ msg ^ "]");
   a

ircd.ml

@@ -7,32 +7,34 @@ let dbg msg a =
     [connection_handler] will be called with the file descriptor of
     any new connections.
 *)
-let establish_server ues connection_handler addr =
+let establish_server d connection_handler addr =
-  let g = Unixqueue.new_group ues in
+  let rec handle_event fd events =
-  let handle_event ues esys e =
+    match events with
-    match e with
+      | [] ->
-      | Unixqueue.Input_arrived (g, fd) ->
+	  ()
+      | Dispatch.Input :: tl ->
 	  let cli_fd, cli_addr = Unix.accept fd in
-	    connection_handler cli_fd
+	    connection_handler cli_fd cli_addr;
-      | _ ->
+	    handle_event fd tl
-	  raise Equeue.Reject
+      | Dispatch.Hangup :: tl ->
+	  Dispatch.delete d fd;
+	  handle_event fd tl
+      | _ :: tl ->
+	  handle_event fd tl
   in
   let srv = Unix.socket Unix.PF_INET Unix.SOCK_STREAM 0 in
     Unix.bind srv addr;
     Unix.listen srv 50;
     Unix.setsockopt srv Unix.SO_REUSEADDR true;
-    Unixqueue.add_handler ues g handle_event;
+    Dispatch.add d fd handle_event [Dispatch.Input];
-    Unixqueue.add_resource ues g (Unixqueue.Wait_in srv, -.1.0)
 
 let main () =
-  let ues = Unixqueue.create_unix_event_system () in
+  let d = Dispatch.create 50 in
-  let g = Unixqueue.new_group ues in
-    Unixqueue.add_handler ues g Iobuf.handle_event;
     establish_server
       ues
-      Client.handle_connection
+      (Client.handle_connection d)
-      (Unix.ADDR_INET (Unix.inet_addr_any, 7777));
+      (Unix.ADDR_INET (Unix.inet_addr_any, 6667));
-    ues#run ()
+    Dispatch.run d
 
 let _ =
   main ()

tests.ml

@@ -1,6 +1,4 @@
-open Unixqueue
 open OUnit
-open Irc
 
 let dump x =
   Printf.ksprintf (fun str -> prerr_string str; flush stderr) x
@@ -335,7 +333,6 @@ let unit_tests =
         );
     ]
 
-(*
 let do_login nick =
   [
     Send ("USER " ^ nick ^ " +iw " ^ nick ^ " :gecos\r\n");
@@ -378,12 +375,11 @@ let regression_tests =
                  Recv ":testserver.test 401 nick otherguy :No such nick/channel\r\n";
                ]
            in
-           let g = Unixqueue.new_group ues in
+           let d = Dispatch.create 2 in
            let a,b = Unix.socketpair Unix.PF_UNIX Unix.SOCK_STREAM 0 in
-             Unixqueue.add_handler ues g Iobuf.handle_event;
+             Client.handle_connection d a (Unix.getpeername a);
-             Client.handle_connection ues g a;
+             chat d b script;
-             ignore (new chat_handler script ues b);
+             Dispatch.run d);
-             chat_run ues);
 
       "Second connection" >::
         (fun () ->
@@ -394,12 +390,11 @@ let regression_tests =
                  Recv ":testserver.test 303 otherguy :otherguy\r\n";
                ]
            in
-           let g = Unixqueue.new_group ues in
+           let d = Dispatch.create 2 in
            let a,b = Unix.socketpair Unix.PF_UNIX Unix.SOCK_STREAM 0 in
-             Unixqueue.add_handler ues g Iobuf.handle_event;
+             Client.handle_connection d a (Unix.getpeername a);
-             Client.handle_connection ues g a;
+             chat d b script;
-             ignore (new chat_handler script ues b);
+             Dispatch.run d);
-             chat_run ues);
 
       "Simultaneous connections" >::
         (fun () ->
@@ -421,20 +416,16 @@ let regression_tests =
                  Recv ":alice!alice@UDS PRIVMSG bob :Hi Bob!\r\n";
                ]
            in
-           let g = Unixqueue.new_group ues in
+           let d = Dispatch.create 4 in
            let aa,ab = Unix.socketpair Unix.PF_UNIX Unix.SOCK_STREAM 0 in
            let ba,bb = Unix.socketpair Unix.PF_UNIX Unix.SOCK_STREAM 0 in
-             Unixqueue.add_handler ues g Iobuf.handle_event;
+             Client.handle_connection d aa (Unix.getpeername aa);
-             Client.handle_connection ues g aa;
+             Client.handle_connection d ba (Unix.getpeername ba);
-             Client.handle_connection ues g ba;
+             chat d ab script1;
-             ignore (new chat_handler script1 ues ab);
+             chat d bb script2;
-             ignore (new chat_handler script2 ues bb);
+             Dispatch.run d);
-             chat_run ues);
     ]
-*)
 
 let _ =
   Irc.name := "testserver.test";
-  run_test_tt_main (TestList [unit_tests (*; regression_tests *)])
+  run_test_tt_main (TestList [unit_tests; regression_tests])
-	
-