Exercices

Exercices

Les exercices proposés permettent de mettre en oeuvre différents types d'applications réparties. Le premier offre un nouveau service réseau permettant une mise à l'heure des machines clientes. Le deuxième exercice montre comment on peut utiliser les ressources de différentes machines pour distribuer un calcul.

Service : horloge

Cet exercice consiste à implanter un service << horloge >> qui donne l'heure à tout client qui le désire. L'idée est d'avoir ainsi une machine de référence pour mettre à l'heure les différentes machines d'un réseau.

Définir un protocole pour la transmission d'une date contenant la description du jour, mois, heure, minute, seconde.


# type horloge = { jour:int; mois:int; heure:int; minute:int; seconde:int } ;;
type horloge =
  { jour: int;
    mois: int;
    heure: int;
    minute: int;
    seconde: int }
# let encode date = 
   let str = String.create 5 in 
   str.[0] <- char_of_int date.jour ;
   str.[1] <- char_of_int date.mois ;
   str.[2] <- char_of_int date.heure ;
   str.[3] <- char_of_int date.minute ;
   str.[4] <- char_of_int date.seconde ;
   str ;;
val encode : horloge -> string = <fun>
# let decode str = 
   { jour = int_of_char str.[0] ;
     mois = int_of_char str.[1] ;
     heure = int_of_char str.[2] ;
     minute = int_of_char str.[3] ; 
     seconde = int_of_char str.[4] } ;;
val decode : string -> horloge = <fun>

Écrire la fonction ou la classe de service en réutilisant un des serveurs génériques présentés. Ce service envoie les informations de la date à chaque connexion acceptée, puis referme la socket. Nous utilisons la fonction main_serveur(20) :


# main_serveur ;;
- : (in_channel -> out_channel -> 'a) -> unit = <fun>

# let horloge_service ic oc =
   try 
     let date = Unix.localtime (Unix.time ()) in 
     let date_horloge = 
         { jour = date.Unix.tm_mday ;
           mois = date.Unix.tm_mon + 1 ; 
           heure = date.Unix.tm_hour ; 
           minute = date.Unix.tm_min ; 
           seconde = date.Unix.tm_sec  } in
     output_string oc (encode date_horloge) ; 
     flush oc  
   with exn -> print_endline "Fin du traitement"; flush stdout 
 
 let main_horloge () = main_serveur horloge_service ;;
val horloge_service : 'a -> out_channel -> unit = <fun>
val main_horloge : unit -> unit = <fun>

Écrire un client qui effectue une mise à jour de sa date toutes les heures. Nous utilisons la fonction main_client20 :


# main_client ;;
- : (in_channel -> out_channel -> 'a) -> unit = <fun>

# let client_horloge ic oc = 
   let date = ref { jour=0; mois=0; heure=0; minute=0; seconde=0 } in
   try
     while true do
       let buffer = "xxxxx" in   
       ignore (input ic buffer 0 5) ;
       date := decode buffer ;
       print_endline "BIP";
       flush stdout ;
       Unix.sleep 3600 
     done
   with 
     exn -> shutdown_connection ic ; raise exn  ;;
val client_horloge : in_channel -> 'a -> unit = <fun>

# let main_horloge () = main_client client_horloge ;;
val main_horloge : unit -> unit = <fun>

Comment tenir compte du décalage de temps dû à la requête? On peut mesurer le temps écoulé entre la demande de connexion et la réception de la réponse. On suppose que ce délai est le double de celui mis par la réponse et on corrige le résultat en conséquence.

Une machine à café en réseau

On veut réaliser un petit service simulant un distributeur de boissons. La description sommaire du protocole d'échange entre un client et le service est la suivante :

à sa connexion, le client reçoit la liste des boissons disponibles ;
il envoie alors au serveur le nom de la boisson de son choix ;
le serveur lui retourne le prix à payer ;
le client envoie ce prix, ou toute autre somme ;
le serveur répond par le nom de la boisson choisie et le montant de la monnaie rendue.

Le serveur peut aussi répondre par un message d'erreur s'il n'a pas compris une requête, n'a plus assez de monnaie, etc. Une requête d'un client ne contient toujours qu'un seul composant.

Les échanges se font sous forme de chaînes de caractères. Les différents composants des messages sont séparés par deux points et toute chaîne se termine par :$\n.

La fonction de service communique avec la machine à café en utilisant une file d'attente pour passer les commandes et une table de hachage pour récupérer les boissons préparées et la monnaie.

Cet exercice mettra en oeuvre la communication via des sockets, des processus légers avec un peu de concurrence et des objets.

Réécrire la fonction establish_server en utilisant les primitives de ThreadUnix. On reprend les fonctions hostaddr et my_inet_addr de ce chapitre.


# val hostaddr : string -> Unix.inet_addr = <fun>
val my_inet_addr : unit -> Unix.inet_addr = <fun>

 let establish_server f saddr = 
   let sock = ThreadUnix.socket Unix.PF_INET Unix.SOCK_STREAM 0 in
     Unix.bind sock saddr;
     Unix.listen sock 5;
     while true do
       let (s,_) = ThreadUnix.accept sock in
       let ic = Unix.in_channel_of_descr s
       and oc = Unix.out_channel_of_descr s in
         ignore (Thread.create (f ic) oc)
     done;;
val establish_server :
  (in_channel -> out_channel -> 'a) -> Unix.sockaddr -> unit = <fun>

Écrire deux fonctions, get_request et send_answer. La première lit et déchiffre une requête et la seconde formate et envoie une réponse à partir d'une liste de chaînes de caractères.


# let read fd =
   let buf = String.create 1024 in
   let n = ThreadUnix.read fd buf 0 1024 in
   let s = String.sub buf 0 n in
   s ;;
val read : Unix.file_descr -> string = <fun>

# let get_request fd =
   let s = read fd in
   match Str.split (Str.regexp "[:]") (String.sub s 0 (String.index s '$')) with
      [s1] -> s1
     | _ -> failwith "BadRequestFormat" ;;
val get_request : Unix.file_descr -> string = <fun>

On redéfinit des fonctions d'entrées-sorties utilisant celles de Threadunix puis on les utilise pour get_request et send_answer. Comme elle revient souvent, on définit également une fonction send_cancel qui envoie un message d'erreur.


# let write fd s =
   let leng = (String.length s) in
   let n = ThreadUnix.write fd s 0 leng in
   if n<leng then failwith "I/O error" ;;
val write : Unix.file_descr -> string -> unit = <fun>

# let send_answer fd ss =
   let rec mk_answer = function 
       [] -> ":$\n"
     | [s] -> s ^ ":$\n"
     | s::ss -> s ^ ":" ^ (mk_answer ss) 
   in
   write fd (mk_answer ss) ;;
val send_answer : Unix.file_descr -> string list -> unit = <fun>

# let send_cancel = let s = "cancel:$\n" in function fd -> write fd s ;;
val send_cancel : Unix.file_descr -> unit = <fun>

Écrire une classe cmd_fifo pour gérer les commandes en attente. On attribuera un numéro unique à chaque nouvelle commande. Implanter pour cela une classe num_cmd_gen.


# class cmd_fifo =
   object(self)
     val n = new num_cmd_gen
     val f = (Queue.create (): (int*int*int) Queue.t)
     val m = Mutex.create ()
     val c = Condition.create ()
 
    method add num_drink paid =
      let num_cmd = n#get() in
      Mutex.lock m ;
      Queue.add (num_cmd, num_drink, paid) f ;
      Mutex.unlock m ;
      Condition.signal c ;
      num_cmd
 
    method wait () =
      Mutex.lock m ;
      Condition.wait c m ;
      let cmd = Queue.take f in
      Mutex.unlock m ;
      cmd
   end ;;
class cmd_fifo :
  object
    val c : Condition.t
    val f : (int * int * int) Queue.t
    val m : Mutex.t
    val n : num_cmd_gen
    method add : int -> int -> int
    method wait : unit -> int * int * int
  end


# class num_cmd_gen =
   object
     val mutable x = 0
     val m = Mutex.create ()
 
     method get() =
       Mutex.lock m ;
       x <- x+1 ;
       let r = x in
       Mutex.unlock m ;
       r
   end ;;
class num_cmd_gen :
  object val m : Mutex.t val mutable x : int method get : unit -> int end

Écrire une classe ready_table pour stocker les boissons préparées par la machine.


# class ready_table size =
   object
     val t = (Hashtbl.create size : (int, (string * int)) Hashtbl.t)
     val m = Mutex.create ()
     val c = Condition.create ()
 
     method add num_cmd num_drink change =
       Mutex.lock m ;
       Hashtbl.add t num_cmd (num_drink, change) ;
       Mutex.unlock m ;
       Condition.broadcast c
 
     method wait num_cmd =
       Mutex.lock m;
       while not(Hashtbl.mem t num_cmd) do Condition.wait c m done ;
       let cmd = Hashtbl.find t num_cmd in
       Hashtbl.remove t num_cmd ;
       Mutex.unlock m ;
       cmd
   end ;;
class ready_table :
  int ->
  object
    val c : Condition.t
    val m : Mutex.t
    val t : (int, string * int) Hashtbl.t
    method add : int -> string -> int -> unit
    method wait : int -> string * int
  end

Écrire la classe machine qui modélise la machine à café. Elle possédera une méthode run bouclant sur la séquence : attendre une commande puis la préparer, tant qu'il reste des boissons disponibles. On définira un type drink_descr indiquant, pour chaque boisson : son nom, la quantité en stock, la quantité restant après satisfaction des commandes et le prix. On utilisera également une fonction auxiliaire array_index qui donne l'indice du premier élément d'un tableau satisfaisant un critère passé en paramètre.


# class machine (f_cmd0:cmd_fifo) (t_ready0:ready_table) =
   object(self)
     val f_cmd = f_cmd0
     val t_ready = t_ready0
     val mutable nb_available_drinks = 0
     val drinks_table = 
       [| { name="cafe"; real_stock=10; virtual_stock=10; price=300 };
          { name="the"; real_stock=5; virtual_stock=5; price=250 };
          { name="chocolat"; real_stock=10; virtual_stock=10; price=250 } |]
     val mutable cash = 0
     val m = Mutex.create()
 
     initializer  nb_available_drinks <- Array.length drinks_table
  
     method get_drink_price i = drinks_table.(i).price
     method get_drink_index s = array_index drinks_table (fun d -> d.name=s)
 
     method get_menu () =
       let f d ns = if d.real_stock > 0 then d.name::ns else ns in
       Array.fold_right f drinks_table []
 
     method cancel_cmd num_drink =
       let drink = drinks_table.(num_drink) in
       drink.virtual_stock <- drink.virtual_stock+1
 
     method set_cmd num_drink paid = f_cmd#add num_drink paid
 
     method wait_cmd num_cmd = t_ready#wait num_cmd
 
     method deliver_drink num_drink =
       let drink = drinks_table.(num_drink) in
       drink.real_stock <- drink.real_stock-1 ;
       if drink.real_stock = 0 then nb_available_drinks <- nb_available_drinks-1
 
     method run() =
       while nb_available_drinks>0 do
         let (num_cmd, num_drink, amount) = f_cmd#wait () in
         let drink = drinks_table.(num_drink) in
         let change = amount - drink.price in
         Mutex.lock m ;
        if (drink.virtual_stock > 0) & (cash >= change) 
        then 
          begin
            drink.virtual_stock <- drink.virtual_stock-1 ;
            cash <- cash + drink.price ;
            t_ready#add num_cmd drink.name change
          end
        else t_ready#add num_cmd "cancel" 0 ;
        Mutex.unlock m
      done
   end ;;
class machine :
  cmd_fifo ->
  ready_table ->
  object
    val mutable cash : int
    val drinks_table : drink_descr array
    val f_cmd : cmd_fifo
    val m : Mutex.t
    val mutable nb_available_drinks : int
    val t_ready : ready_table
    method cancel_cmd : int -> unit
    method deliver_drink : int -> unit
    method get_drink_index : string -> int
    method get_drink_price : int -> int
    method get_menu : unit -> string list
    method run : unit -> unit
    method set_cmd : int -> int -> int
    method wait_cmd : int -> string * int
  end


# type drink_descr =
   { name : string; 
     mutable real_stock : int; 
     mutable virtual_stock : int;
     price : int } ;;


# let array_index t f =
   let i = ref 0 in
   let n = Array.length t in
   while (!i < n) & (not (f t.(!i))) do incr i done ;
   if !i=n then raise Not_found else !i ;;
val array_index : 'a array -> ('a -> bool) -> int = <fun>

Écrire une fonction de service waiter.


# let waiter mach ic oc =
   let f_in = Unix.descr_of_in_channel ic in
   let f_out = Unix.descr_of_out_channel oc in
   (try  
      send_answer f_out (mach#get_menu()) ;
      let drink_name = get_request f_in in 
      let num_drink = mach#get_drink_index drink_name in
      let drink_price = mach#get_drink_price num_drink in
      send_answer f_out [string_of_int drink_price] ;
      let paid = int_of_string (get_request f_in) in
      if paid < drink_price then failwith"NotEnough" ;
      let num_cmd = mach#set_cmd num_drink paid in
      let drink_name, change = mach#wait_cmd num_cmd in
      mach#deliver_drink num_drink;
      send_answer f_out [drink_name; (string_of_int change)]
    with 
       Not_found -> send_cancel f_out
     | Failure("int_of_string") -> send_cancel f_out
     | Failure("I/O error") -> send_cancel f_out
     | Failure("NotEnough") -> send_cancel f_out
     | Failure("BadRequestFormat") -> send_cancel f_out
   );
   close_in ic ;
   flush oc ;
   close_out oc ;
   Thread.exit () ;;
val waiter :
  < deliver_drink : 'a -> 'b; get_drink_index : string -> 'a;
    get_drink_price : 'a -> int; get_menu : unit -> string list;
    set_cmd : 'a -> int -> 'c; wait_cmd : 'c -> string * int; .. > ->
  in_channel -> out_channel -> unit = <fun>

Écrire une fonction principale main qui récupère le numéro de port du service sur la ligne de commande et procède aux diverses initialisations. En particulier, la machine à café est exécutée par un processus.


# let main () =
   if Array.length Sys.argv < 2 
   then 
     begin
       Printf.eprintf "usage : %s port\n" Sys.argv.(0) ;
       exit 1
     end
   else 
     begin
       let port = int_of_string Sys.argv.(1) in
       let f_cmd = new cmd_fifo in
       let t_ready = new ready_table in
       let mach = new machine f_cmd (t_ready 13) in
       ignore (Thread.create mach#run ()) ;
       establish_server (waiter mach) (Unix.ADDR_INET (my_inet_addr (), port))
     end ;;
val main : unit -> unit = <fun>