Make request I/O pure and expose the fd

lib/fuse.ml

@@ -24,6 +24,7 @@ module Struct = Profuse.Struct
 
 type 'a structure = 'a Ctypes_static.structure
 type request = In.Message.t Profuse.request
+type response = In.Message.t Profuse.response
 
 module type IO = sig
   type 'a t
@@ -51,7 +52,7 @@ end
 
 module type RO_SIMPLE = sig
   module IO : IO
-  type 'a req_handler = request -> 'a -> 'a IO.t
+  type 'a req_handler = request -> 'a -> ('a * response list) IO.t
   type t
 
   val getattr    : t req_handler
@@ -131,7 +132,7 @@ module type FS_IO = sig
   val negotiate_mount :
     In.Init.T.t structure -> request -> t -> (request * t) IO.t
 
-  val dispatch : request -> t -> t IO.t
+  val dispatch : request -> t -> (t * response list) IO.t
 end
 
 module type STATE = sig
@@ -152,15 +153,16 @@ end
 module Zero(State : STATE)(IO : IO)
   : RW_FULL with module IO = IO and type t = State.t = struct
   module IO = IO
-  type 'a req_handler = In.Message.t Profuse.request -> 'a -> 'a IO.t
+  type 'a req_handler = request -> 'a -> ('a * response list) IO.t
   type t = State.t
 
   let enosys req st =
     (* Throw away error messages during the write. *)
     let log_error _msg = () in
-    IO.(Out.write_error log_error req Errno.ENOSYS >>= fun () -> return st)
+    IO.return (st, [`Error (Errno.ENOSYS, log_error)])
 
-  let drop _req st = IO.return st
+  let drop _req st =
+    IO.return (st, [`Drop])
 
   let getattr      = enosys
   let opendir _    = enosys
@@ -269,10 +271,7 @@ module Support(IO : IO) = struct
   let enosys req st =
     (* Throw away error messages during the write. *)
     let log_error _msg = () in
-    IO.(Out.write_error log_error req Errno.ENOSYS
-        >>= fun () ->
-        return st
-       )
+    IO.return (st, [`Error (Errno.ENOSYS, log_error)])
 
   let store_entry
     ?(entry_valid=Unsigned.UInt64.zero)
@@ -300,10 +299,10 @@ module Support(IO : IO) = struct
     >>= fun store_attr ->
     let nodeid = Unsigned.UInt64.of_int64 node.Nodes.id in
     let generation = Unsigned.UInt64.of_int64 node.Nodes.gen in
-    IO.Out.write_reply req
+    IO.return [`Reply
       (Profuse.Out.Entry.create ~nodeid ~generation
          ~entry_valid ~entry_valid_nsec
          ~attr_valid ~attr_valid_nsec
-         ~store_attr)
+         ~store_attr)]
   )
 end

lib/fuse.mli

@@ -5,6 +5,7 @@ module Nodes = Nodes
 
 type 'a structure = 'a Ctypes_static.structure
 type request = Profuse.In.Message.t Profuse.request
+type response = Profuse.In.Message.t Profuse.response
 
 module type IO = sig
   type 'a t
@@ -35,7 +36,7 @@ end
 
 module type RO_SIMPLE = sig
   module IO : IO
-  type 'a req_handler = request -> 'a -> 'a IO.t
+  type 'a req_handler = request -> 'a -> ('a * response list) IO.t
 
   type t
 
@@ -119,7 +120,7 @@ module type FS_IO = sig
   val negotiate_mount :
     Profuse.In.Init.T.t structure -> request -> t -> (request * t) IO.t
 
-  val dispatch : request -> t -> t IO.t
+  val dispatch : request -> t -> (t * response list) IO.t
 end
 
 module type STATE = sig
@@ -170,7 +171,7 @@ module Support : functor(IO : IO) -> sig
 
   val nodeid : request -> Unsigned.uint64
 
-  val enosys : request -> 'a -> 'a IO.t
+  val enosys : request -> 'a -> ('a * response list) IO.t
 
   val store_entry :
     ?entry_valid:Unsigned.uint64 ->
@@ -187,5 +188,5 @@ module Support : functor(IO : IO) -> sig
     ?attr_valid:Unsigned.uint64 ->
     ?attr_valid_nsec:Unsigned.uint32 ->
     ('a Nodes.node -> (Profuse.Struct.Attr.T.t structure -> unit) IO.t)
-    -> 'a Nodes.node -> request -> unit IO.t
+    -> 'a Nodes.node -> request -> response list IO.t
 end

lib/profuse_7_23.ml

@@ -1581,3 +1581,11 @@ module Out = struct
 end
 
 type 'a request = (In.Hdr.T.t, 'a) packet
+
+type 'a response = [
+  | `Reply of ('a request -> char Ctypes.CArray.t)
+  | `Ack
+  | `Drop
+  | `Error of Errno.t * (string -> unit)
+  | `Raise of exn
+]

lib/profuse_7_23.mli

@@ -639,3 +639,11 @@ module Out : sig
     val describe : ('a,t) packet -> string
   end
 end
+
+type 'a response = [
+  | `Reply of ('a request -> char Ctypes.CArray.t)
+  | `Ack
+  | `Drop
+  | `Error of Errno.t * (string -> unit)
+  | `Raise of exn
+]

lwt/fuse_lwt.ml

@@ -20,6 +20,7 @@ end
 
 type socket = {
   id    : int;
+  fd    : Unix.file_descr;
   read  : int -> uint8 Ctypes.CArray.t Lwt.t;
   write : uint8 Ctypes.ptr -> int -> int Lwt.t;
   nwrite: uint8 Ctypes.ptr -> int -> int Lwt.t;
@@ -28,6 +29,7 @@ type socket = {
 
 let null_socket = {
   id = -1;
+  fd    = Unix.stdout;
   read  = (fun _ -> Lwt.return (Ctypes.CArray.make uint8_t 0));
   write = (fun _ len -> Lwt.return len);
   nwrite= (fun _ len -> Lwt.return len);
@@ -37,19 +39,21 @@ let null_socket = {
 let socket_table = ref (Array.make 0 null_socket)
 
 (* TODO: release socket *)
-let new_socket ~read ~write ~nwrite ~nread =
+let new_socket ~fd ~read ~write ~nwrite ~nread =
   let table = !socket_table in
   let next_id = Array.length table in
   let table = Array.init (next_id + 1) (fun i ->
     if i <> next_id
     then table.(i)
-    else { id = next_id; read; write; nwrite; nread }
+    else { id = next_id; fd; read; write; nwrite; nread }
   )  in
   socket_table := table;
   table.(next_id)
 
 let socket_id { id } = id
 
+let socket_fd { fd } = fd
+
 let get_socket k =
   (!socket_table).(k)
 
@@ -295,9 +299,9 @@ module Trace(F : FS_LWT) : FS_LWT with type t = F.t = struct
     let dispatch req t =
       Printf.eprintf "    %s\n%!" (Profuse.In.Message.describe req);
       Fs.dispatch req t
-      >>= fun t ->
+      >>= fun (t, response) ->
       Printf.eprintf "    %s\n%!" (F.string_of_state req t);
-      return t
+      return (t, response)
   end
 end
 
@@ -354,8 +358,7 @@ module Dispatch(F : FS_LWT) : FS_LWT with type t = F.t = struct
         | Destroy -> destroy req t
         | Setattr s -> setattr s req t
         | Other _ | Unknown _ ->
-          IO.(Out.write_error log_error req Errno.ENOSYS
-              >>= fun () -> return t)
+          IO.return (t, [`Error (Errno.ENOSYS, log_error)])
       ) (function
         | Unix.Unix_error(e, _, _) as exn ->
           let host = req.chan.host.Host.errno in
@@ -366,20 +369,13 @@ module Dispatch(F : FS_LWT) : FS_LWT with type t = F.t = struct
               Errno.EIO
             | errno::_ -> errno
           in
-          IO.(Out.write_error log_error req errno
-              >>= fun () ->
-              return t
-             )
+          IO.return (t, [`Error (errno, log_error)])
         | Errno.Error { Errno.errno = errno :: _ } ->
-          IO.(Out.write_error log_error req errno
-              >>= fun () ->
-              return t
-             )
+          IO.return (t, [`Error (errno, log_error)])
         | (Destroy k) as exn -> IO.fail exn
         | exn ->
           log_error ("Unknown exception caught: "^(Printexc.to_string exn));
-          IO.(Out.write_error log_error req Errno.EIO
-              >>= fun () -> fail exn)
+          IO.return (t, [`Error (Errno.EIO, log_error); `Raise exn])
       )
   end
 end

lwt/fuse_lwt.mli

@@ -31,6 +31,7 @@ module Server(M : MOUNT_LWT)(F : FS_LWT)(IO : IO_LWT)
 type socket
 
 val new_socket :
+  fd:Unix.file_descr ->
   read:(int -> Unsigned.uint8 Ctypes.CArray.t Lwt.t) ->
   write:(Unsigned.uint8 Ctypes.ptr -> int -> int Lwt.t) ->
   nwrite:(Unsigned.uint8 Ctypes.ptr -> int -> int Lwt.t) ->
@@ -39,6 +40,8 @@ val new_socket :
 
 val socket_id : socket -> int
 
+val socket_fd : socket -> Unix.file_descr
+
 val get_socket : int -> socket
 
 val read_socket : socket -> (int -> Unsigned.uint8 Ctypes.CArray.t Lwt.t)