helm/graphs/tools/uriSetQueue.ml

   1 (* Copyright (C) 2002, HELM Team.
   2  *
   3  * This file is part of HELM, an Hypertextual, Electronic
   4  * Library of Mathematics, developed at the Computer Science
   5  * Department, University of Bologna, Italy.
   6  *
   7  * HELM is free software; you can redistribute it and/or
   8  * modify it under the terms of the GNU General Public License
   9  * as published by the Free Software Foundation; either version 2
  10  * of the License, or (at your option) any later version.
  11  *
  12  * HELM is distributed in the hope that it will be useful,
  13  * but WITHOUT ANY WARRANTY; without even the implied warranty of
  14  * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
  15  * GNU General Public License for more details.
  16  *
  17  * You should have received a copy of the GNU General Public License
  18  * along with HELM; if not, write to the Free Software
  19  * Foundation, Inc., 59 Temple Place - Suite 330, Boston,
  20  * MA  02111-1307, USA.
  21  *
  22  * For details, see the HELM World-Wide-Web page,
  23  * http://cs.unibo.it/helm/.
  24  *)
  25
  26 let debug = true;;
  27 let debug_print s = if debug then prerr_endline s;;
  28 Http_common.debug := debug;;
  29
  30 open Printf;;
  31
  32 let daemon_name = "Uri Set Queue";;
  33 let default_port = 48082;;
  34 let port_env_var = "URI_SET_QUEUE_PORT";;
  35
  36 module OrderedUri: Set.OrderedType with type t = string =
  37   struct
  38     type t = string
  39     let compare = compare
  40   end
  41 module UriSet = Set.Make (OrderedUri)
  42 type uri_queue = {
  43   mutable size: int;
  44   mutable overflowed: bool;
  45   uris: string Queue.t;
  46   mutable olduris: UriSet.t;
  47 }
  48   (** raised when a queue is accessed before being defined *)
  49 exception Queue_not_found of int;;
  50   (** global uri_queue, used by all children *)
  51 let uri_queue = {
  52   size = 0; overflowed = false; uris = Queue.create (); olduris = UriSet.empty
  53 };;
  54 let (get_queue, add_queue, remove_queue) =
  55   let uri_queues = Hashtbl.create 17 in
  56   ((fun pid -> (* get_queue *)
  57     try
  58       Hashtbl.find uri_queues pid
  59     with Not_found -> raise (Queue_not_found pid)),
  60   (fun pid size ->  (* add_queue *)
  61     Hashtbl.replace
  62       uri_queues
  63       pid
  64       { size = size; overflowed = false;
  65         uris = Queue.create (); olduris = UriSet.empty }),
  66   (fun pid -> (* remove_queue *)
  67     try
  68       Hashtbl.remove uri_queues pid
  69     with Not_found -> raise (Queue_not_found pid)))
  70 ;;
  71
  72 exception Found;;
  73 let queue_mem item queue =  (* mem function over queues *)
  74   try
  75     Queue.iter (fun e -> if item = e then raise Found) queue;
  76     false
  77   with Found -> true
  78 ;;
  79
  80 let port =
  81   try
  82     int_of_string (Sys.getenv port_env_var)
  83   with
  84   | Not_found -> default_port
  85   | Failure "int_of_string" ->
  86       prerr_endline "Warning: invalid port, reverting to default";
  87       default_port
  88 in
  89 let callback (req: Http_types.request) outchan =
  90   try
  91     let res = new Http_response.response () in
  92     res#addBasicHeaders;
  93     res#setContentType "text/xml";
  94     (match req#path with
  95
  96     | "/add_if_not_in" ->
  97         let (uri, pid) = (req#param "uri", int_of_string (req#param "PID")) in
  98         debug_print (sprintf "Adding uri '%s' to queue '%d'" uri pid);
  99         let queue = get_queue pid in
 100         let result =
 101           if (Queue.length queue.uris) + (UriSet.cardinal queue.olduris) >=
 102              queue.size
 103           then
 104             begin (* overflow! *)
 105               queue.overflowed <- true;
 106               debug_print "Answer: not_added_because_already_too_many";
 107               "not_added_because_already_too_many"
 108             end else begin  (* there's room for another uri *)
 109               if (queue_mem uri queue.uris) || (UriSet.mem uri queue.olduris)
 110               then
 111                 begin (* url already in *)
 112                   debug_print "Answer: already_in";
 113                   "already_in"
 114                 end else begin (* uri not in *)
 115                   Queue.add uri queue.uris;
 116                   debug_print "Answer: added";
 117                   "added"
 118                 end
 119             end
 120         in
 121         res#setBody (sprintf "<?xml version=\"1.0\"?>\n<%s/>\n" result);
 122         if debug then res#serialize stderr;
 123         Http_daemon.respond_with res outchan
 124
 125     | "/is_overflowed" ->
 126         let pid = int_of_string (req#param "PID") in
 127         let queue = get_queue pid in
 128         let result = string_of_bool (queue.overflowed) in
 129         debug_print (sprintf "%d queue is_overflowed = %s" pid result);
 130         res#setBody (sprintf "<?xml version=\"1.0\"?>\n<%s/>\n" result);
 131         if debug then res#serialize stderr;
 132         Http_daemon.respond_with res outchan
 133
 134     | "/set_uri_set_size" ->
 135         let (pid, size) =
 136           (int_of_string (req#param "PID"), int_of_string (req#param "size"))
 137         in
 138         debug_print (sprintf "Setting size '%d' for queue '%d'" size pid);
 139         (try
 140           let queue = get_queue pid in
 141           queue.size <- size;
 142         with Queue_not_found p ->
 143           assert (p = pid);
 144           add_queue pid size);
 145         res#setBody "<?xml version=\"1.0\"?>\n<done/>\n";
 146         if debug then res#serialize stderr;
 147         Http_daemon.respond_with res outchan
 148
 149     | "/get_next" ->
 150         let pid = int_of_string (req#param "PID") in
 151         debug_print (sprintf "Getting next uri from queue '%d'" pid);
 152         let queue = get_queue pid in
 153         let element = (* xml response's root element *)
 154           try
 155             let uri = Queue.take queue.uris in
 156             queue.olduris <- UriSet.add uri queue.olduris;
 157             sprintf
 158               "<%suri value=\"%s\"/>"
 159               (if queue.overflowed then "marked_" else "")
 160               uri
 161           with Queue.Empty -> "<empty/>"
 162         in
 163         res#setBody ("<?xml version=\"1.0\"?>\n" ^ element ^ "\n");
 164         if debug then res#serialize stderr;
 165         Http_daemon.respond_with res outchan
 166
 167     | "/reset_to_empty" ->
 168         let pid = int_of_string (req#param "PID") in
 169         remove_queue pid;
 170         debug_print (sprintf "Resetting queue '%d'" pid);
 171         res#setBody "<?xml version=\"1.0\"?>\n<done/>\n";
 172         if debug then res#serialize stderr;
 173         Http_daemon.respond_with res outchan
 174
 175     | invalid_request ->
 176         debug_print ("Invalid request received");
 177         Http_daemon.respond_error ~status:(`Client_error `Bad_request) outchan);
 178         prerr_endline "Request done!\n"
 179   with
 180   | Http_types.Param_not_found attr_name ->
 181       Http_daemon.respond_error
 182         ~status:(`Client_error `Bad_request)
 183         ~body:(sprintf "Parameter '%s' is missing" attr_name)
 184         outchan
 185   | Failure "int_of_string" ->  (* error in converting some paramters *)
 186       Http_daemon.respond_error ~status:(`Client_error `Bad_request) outchan
 187   | Queue_not_found queue_name ->
 188       Http_daemon.respond_error
 189         ~status:(`Client_error `Bad_request)
 190         ~body:(sprintf "Queue '%d' is not defined" queue_name)
 191         outchan
 192 in
 193
 194 printf "%s started and listening on port %d\n" daemon_name port;
 195 flush stdout;
 196 Http_daemon.start' ~port ~mode:`Thread callback;
 197 printf "%s is terminating, bye!\n" daemon_name
 198