open Lwt open Lwt_stream (* ***** *) (* Types *) (* ***** *) type resource = Fish | Cider | Hemp | Garment | Void;; type r_stream = resource Lwt_stream.t type r_loader = resource option -> unit type store = Store of (resource * (r_stream * r_loader)) list let all_resources = [ Fish; Cider; Hemp; Garment ];; (* ************** *) (* Helper methods *) (* ************** *) let resource_to_s r = match r with | Fish -> "Fish" | Cider -> "Cider" | Hemp -> "Hemp" | Garment -> "Garment" | Void -> "Not a real resource";; let out_s = Lwt_io.write Lwt_io.stdout;; let stream_is_not_empty str = let l = Lwt_stream.get_available (Lwt_stream.clone str) in (List.length l > 0);; let stream_length str = Lwt_stream.get_available (Lwt_stream.clone str);; (* ***************************** *) (* Synchronization / Termination *) (* ***************************** *) let must_terminate = ref false;; let stopwatch delay = (Lwt_unix.sleep delay) >> (must_terminate := true; return ()) >> (out_s "\x1B[31;1mStop\x1B[0m\n");; (* ************** *) (* Global streams *) (* ************** *) let store = Store (List.map (fun r -> let (str,push) = (Lwt_stream.create()) in (r, (str, push))) all_resources);; (* Stream storing the missing resources *) let (miss_str,miss_push):(r_stream*r_loader) = Lwt_stream.create();; (* Stream storing the list of factories *) let (fact_str,fact_push):(r_stream*r_loader) = Lwt_stream.create();; (* ************************ *) (* Producer / Consumer loop *) (* ************************ *) let load_in_store (Store s) res = let (_,push) = List.assoc res s in push (Some res); return ();; let rec fetch_from_store (Store s) res = if (!must_terminate) then return None else let (str,_) = List.assoc res s in if (stream_is_not_empty str) then Lwt_stream.get str else (out_s (Printf.sprintf "\x1B[31;1mWait %s\x1B[0m\n" (resource_to_s res))) >> (miss_push (Some res); return ()) >> (Lwt_unix.sleep 0.1) >> (fetch_from_store (Store s) res);; let rec display_store (Store s) delay = if (!must_terminate) then return () else (out_s "\x1B[1mStore\x1B[0m : |") >> Lwt_list.iter_s (fun (res,(str,_)) -> let l = stream_length str in out_s (Printf.sprintf "%s=%d|" (resource_to_s res) (List.length l)) ) s >> (out_s "\n") >> (Lwt_unix.sleep delay) >> (display_store (Store s) delay);; let rec create_producer store res delay = if (!must_terminate) then return () else (Lwt_unix.sleep delay) >> (out_s (Printf.sprintf "Produced %s\n" (resource_to_s res))) >> (load_in_store store res) >> (create_producer store res delay);; let rec create_consumer store res delay = if (!must_terminate) then return () else (Lwt_unix.sleep delay) >> (fetch_from_store store res) >>= (fun s -> match s with | Some r -> out_s (Printf.sprintf "Consumed %s\n" (resource_to_s r)) | None -> out_s "Still waiting\n") >> (create_consumer store res delay);; let min_snd l = List.fold_left (fun (rm,vm) (r,v) -> if (v < vm) then (r,v) else (rm,vm)) (Void, 10.) l;; let sub_add l d res nd = List.map (fun (u,v) -> if (u = res) then (u,nd) else (u,v-.d)) l;; (* Creates a consumer of a list of resources *) let rec create_consumer_l store l_res = let rec create_rec l_cur = if (!must_terminate) then return () else let (res,delay) = min_snd l_cur in let m_res = sub_add l_cur delay res (List.assoc res l_res) in (Lwt_unix.sleep delay) >> (fetch_from_store store res) >>= (fun s -> match s with | Some r -> out_s (Printf.sprintf "Consumed %s\n" (resource_to_s r)) | None -> out_s "Still waiting\n") >> (create_rec m_res) in (create_rec l_res);; let rec create_transformer store source dest delay = if (!must_terminate) then return () else (Lwt_unix.sleep delay) >> (fetch_from_store store source) >>= (fun res -> load_in_store store dest) >> (create_transformer store source dest delay);; (* ************************************ *) (* List of lazy producers and consumers *) (* ************************************ *) let interv = 0.05;; let maxtime = 40.*.interv;; let create_factory res = match res with | Cider -> (fun () -> create_producer store Cider (0.66*.interv)) | Fish -> (fun () -> create_producer store Fish (0.5*.interv)) | Hemp -> (fun () -> create_producer store Hemp interv) | Garment -> (fun () -> create_transformer store Hemp Garment (0.5*.interv)) | Void -> failwith "Impossible to create a void factory";; let peasant_consumer n = let nf = float n in (fun () -> create_consumer_l store [ (Fish, interv*.100./.nf); (Cider, interv*.227.3/.nf) ]);; let citizen_consumer n = let nf = float n in (fun () -> create_consumer_l store [ (Fish, interv*.250./.nf); (Cider, interv*.227.3/.nf); (Garment, interv*.238.1/.nf) ]);; (* ********** *) (* Supervisor *) (* ********** *) (* Returns the most appearing element inside a list *) let most_appearing l = if (l = []) then Void else let rec count_elems l x = List.fold_left (fun c u -> if (u = x) then c+1 else c) 0 l in let counts_s = List.map (fun r -> (count_elems l r,r)) all_resources in let (max,res) = List.fold_left (fun (r,m) (s,n) -> if (r > s) then (r,m) else (s,n)) (0,Void) counts_s in res;; let supervisor delay = let rec super_rec () = if (!must_terminate) then return () else (Lwt_unix.sleep delay) >> (let l = Lwt_stream.get_available miss_str in let r = most_appearing l in match r with | Void -> return () | _ -> (out_s (Printf.sprintf "Must add %s\n" (resource_to_s r))) (let thr = create_factory r in fact_push (Some r); thr()) ) >> (super_rec ()) in (fun () -> super_rec());; (* ****************************************** *) (* Display the final results of the algorithm *) (* ****************************************** *) let display_results () = let l = Lwt_stream.get_available fact_str in let l_s = String.concat "," (List.map resource_to_s l) in out_s ("Constructed factories : "^l_s^"\n");; (* ************************ *) (* Starting list of threads *) (* ************************ *) let agents_list = [ peasant_consumer 250; citizen_consumer 250; supervisor (4.*.interv); ] let launch_agents () = Lwt_list.iter_p (fun f -> f()) agents_list;; (* ********* *) (* Main loop *) (* ********* *) lwt () = ((launch_agents()) <&> (display_store store 0.1) <&> (stopwatch maxtime)) >> (display_results());;