fifowatcher.ml

   1 (** fifowatcher.ml: Routines for creating and managing fifos *)
   2
   3 open Inotify
   4 open Unix
   5 open Globals
   6 open Dirwatcher
   7 open Printf
   8
   9 (** A connected process, FIFO *)
  10 type channel_pipe = Process of out_channel | Fifo of out_channel
  11
  12 type signed_fd = Infd of Unix.file_descr | Outfd of Unix.file_descr
  13
  14 let fdmap: (Unix.file_descr,string*string) Hashtbl.t = Hashtbl.create 1024
  15 let pidmap: (int,signed_fd*signed_fd*Unix.file_descr) Hashtbl.t = Hashtbl.create 1024
  16 let backend_prefix = ref ""
  17 let open_fds: (Unix.file_descr,channel_pipe) Hashtbl.t = Hashtbl.create 1024
  18
  19 (** Receive an event from a running script. This event must be relayed to the
  20   slice that invoked it
  21   @param idesc fd/fname identifier for process
  22   *)
  23 let receive_process_event (idesc: fname_and_fd) (_: fname_and_fd) =
  24   let (_,ifd) = idesc in
  25   let cp = try Hashtbl.find open_fds ifd with
  26       Not_found->
  27         printf "Fifo fd disappeared\n";raise Bug
  28   in
  29     match (cp) with
  30       | Fifo(fifo_outchan) ->
  31           let process_inchan = in_channel_of_descr ifd in
  32           let cont = ref true in
  33           let count = ref 0 in
  34             count:=!count + 1;
  35             while (!cont) do
  36               try
  37                 let curline = input_line process_inchan in
  38                   fprintf fifo_outchan "%s\n" curline;flush fifo_outchan
  39               with
  40                 | End_of_file|Sys_blocked_io|Unix_error(EPIPE,_,_)|Unix_error(EBADF,_,_) ->
  41                     begin
  42                       cont:=false
  43                     end
  44                 | Unix_error(_,s1,s2) -> printf "Unix error %s - %s\n" s1 s2;flush Pervasives.stdout;cont:=false
  45                 | e -> printf "Error - received unexpected event from file system !!!\n";raise e
  46             done
  47       | _ -> printf "Bug! Process fd received in the channel handler\n";raise Bug
  48
  49
  50 let rec openentry_int fifoin fifoout (abspath:string*string) =
  51   let fdin =
  52     try openfile fifoin [O_RDONLY;O_NONBLOCK] 0o777 with
  53         e->printf "Error opening and connecting FIFO: %s,%o\n" fifoin 0o777;flush Pervasives.stdout;raise e
  54   in
  55     Hashtbl.replace fdmap fdin abspath;
  56     Fdwatcher.add_fd (Some(fifoin),fdin) (Some(fifoout),stdout) receive_fifo_event
  57 and reopenentry_int fdin fifoin fifoout =
  58   close fdin;
  59     Fdwatcher.del_fd fdin;
  60     let abspath = try
  61       Hashtbl.find fdmap fdin with _ -> printf "Bug: Phantom pipe\n";flush Pervasives.stdout;raise Bug
  62     in
  63       openentry_int fifoin fifoout abspath
  64 and receive_fifo_event eventdescriptor outdescriptor =
  65   let evfname,evfd = eventdescriptor in
  66   let (fname_other,fd_other) = outdescriptor in
  67   let outfd =
  68     match (fname_other) with
  69       | Some(str)->
  70           (
  71             try openfile str [O_WRONLY;O_NONBLOCK] 0o777 with
  72                 _->printf "Output pipe not open, using stdout in place of %s\n" str;flush Pervasives.stdout;stdout
  73           )
  74       | None-> printf "Bug, nameless pipe\n";raise Bug
  75   in
  76   let pipe = try Hashtbl.find open_fds evfd with
  77     | Not_found ->
  78         (* This is a fifo fd for sure *)
  79         let execpath,slice_name = Hashtbl.find fdmap evfd in
  80         (* Spawn server. We assume that the fd is one fifo opened RW *)
  81         let (myinfd,pout) = Unix.pipe () in
  82         let (pin,myoutfd) = Unix.pipe () in
  83           set_nonblock myinfd;
  84           let pid = try create_process execpath [|execpath;slice_name|] pin pout pout with e -> printf "Error executing service: %s\n" execpath;flush Pervasives.stdout;raise e
  85           in
  86             Hashtbl.add pidmap pid (Infd(myinfd),Outfd(myoutfd),evfd);
  87             Hashtbl.add open_fds evfd (Process(out_channel_of_descr myoutfd));
  88             Hashtbl.add open_fds myinfd (Fifo(out_channel_of_descr outfd));
  89             Fdwatcher.add_fd (None,myinfd) (None,myinfd) receive_process_event;
  90             (Process(out_channel_of_descr myoutfd))
  91   in
  92   let inchan_fd = in_channel_of_descr evfd in
  93     match (pipe) with
  94       | Process(out_channel) ->
  95           let cont = ref true in
  96             while (!cont) do
  97               try
  98                 printf "Reading...\n";flush Pervasives.stdout;
  99                 let curline = input_line inchan_fd in
 100                   fprintf out_channel "%s\n" curline;flush out_channel
 101               with
 102                 |End_of_file->
 103                     (
 104                       match (evfname,fname_other) with
 105                         | Some(str1),Some(str2)->
 106                             reopenentry_int evfd str1 str2
 107                         | Some(str1),None ->
 108                             printf "Bug, nameless pipe\n";flush Pervasives.stdout;raise Bug
 109                         | None,_ ->
 110                             printf "Race condition -> user deleted file before closing it. Clever ploy, but won't work.\n";
 111                             flush Pervasives.stdout
 112                     );
 113                     cont:=false
 114                 |Sys_blocked_io ->printf "Sysblockedio\n";flush Pervasives.stdout;
 115                                   cont:=false
 116                 | _ ->printf "Bug: unhandled exception\n";flush Pervasives.stdout;raise Bug
 117             done
 118       | _ -> printf "BUG! received process event from fifo\n";raise Bug
 119
 120
 121 let mkentry fqp abspath perm =
 122   printf "Making entry %s->%s\n" fqp abspath;flush Pervasives.stdout;
 123   let fifoin=sprintf "%s.in" fqp in
 124   let fifoout=sprintf "%s.out" fqp in
 125     (try Unix.unlink fifoin with _ -> ());
 126     (try Unix.unlink fifoout with _ -> ());
 127     (try
 128        Unix.mkfifo (sprintf "%s.in" fqp) 0o666
 129      with
 130          e->printf "Error creating FIFO: %s->%s,%o\n" fqp fifoin perm;flush Pervasives.stdout;raise e);
 131     (try
 132        Unix.mkfifo (sprintf "%s.out" fqp) 0o666
 133      with
 134          e->printf "Error creating FIFO: %s->%s,%o\n" fqp fifoout perm;flush Pervasives.stdout;raise e)
 135
 136 (** Open fifos for a session *)
 137 let openentry fqp abspath perm =
 138   let fifoin = String.concat "." [fqp;"in"] in
 139   let fifoout = String.concat "." [fqp;"out"] in
 140     openentry_int fifoin fifoout abspath
 141
 142 let sigchld_handle s =
 143   let pid,_=Unix.waitpid [Unix.WNOHANG] 0 in
 144     try
 145       let value = Hashtbl.find pidmap pid in
 146         match value with
 147           | (Infd(ifd),Outfd(ofd),fd) ->
 148               close(ifd);close(ofd);
 149               Hashtbl.remove open_fds fd;
 150               Fdwatcher.del_fd ifd;
 151               Hashtbl.remove pidmap pid
 152           | _ -> printf "BUG! Got fds in the wrong order\n";
 153                  flush Pervasives.stdout;
 154                  raise Bug
 155     with
 156         Not_found-> (* Do nothing, probably a grandchild *)
 157           ()
 158
 159 let initialize () =
 160   Sys.set_signal Sys.sigchld (Sys.Signal_handle sigchld_handle)