10
1
mirror of https://gitlab.com/scemama/qmcchem.git synced 2024-06-22 21:22:06 +02:00

Working on binary input

This commit is contained in:
Anthony Scemama 2022-01-11 13:41:13 +01:00
parent a0323922a8
commit dfbbf8b329
7 changed files with 467 additions and 273 deletions

View File

@ -57,24 +57,65 @@ let zero =
let to_bytes b =
(* [ Length of b
[ Length of value ;
Value ;
Value ;
Length of weight ;
Weight ;
... ] ] *)
let l =
[ Property.to_bytes b.property ;
Sample.to_bytes b.value ;
Weight.to_bytes b.weight ;
bytes_of_int b.pid ;
Block_id.to_bytes b.block_id ;
Compute_node.to_bytes b.compute_node ]
[ Property.to_bytes b.property ;
Sample.to_bytes b.value ;
Weight.to_bytes b.weight ;
bytes_of_int b.pid ;
Block_id.to_bytes b.block_id ;
Compute_node.to_bytes b.compute_node ]
|> List.map (fun x -> [ bytes_of_int (Bytes.length x) ; x ] )
|> List.concat
in
let result =
Bytes.concat Bytes.empty (zero :: l)
in
Bytes.set_int64_le result 8 (Int64.of_int (Bytes.length result));
Bytes.set_int64_le result 0 (Int64.of_int ((Bytes.length result) - 8));
result
let read_bytes b =
(* Reads m, the first 8 bytes as an int64 containing the number of bytes to read.
Then, read the next m bytes and return a tuple containing the decoded data and the rest.
*)
let l = Bytes.length b in
let m =
Bytes.get_int64_le b 0
|> Int64.to_int
in
let nl = l-m-8 in
if nl > 0 then
(Bytes.sub b 8 m, Some (Bytes.sub b (8+m) nl))
else
(Bytes.sub b 8 m, None)
let of_bytes b =
let b, _rest =
read_bytes b
in
let rec loop accu s =
match read_bytes s with
| data, None -> List.rev (data :: accu)
| data, (Some rest) -> loop (data :: accu) rest
in
let result =
match loop [] b with
| value :: weight :: property :: compute_node :: pid :: block_id :: [] ->
{ property = Property.of_bytes property;
value = Sample.of_bytes value;
weight = Weight.of_bytes weight;
pid = int_of_bytes pid;
block_id = Block_id.of_bytes block_id;
compute_node = Compute_node.of_bytes compute_node;
}
| _ -> assert false
in
result
@ -87,11 +128,6 @@ let to_string b =
(string_of_int b.pid)
(Block_id.to_int b.block_id)
(*
let of_string s =
Bytes.of_string s
|> of_bytes
*)
let dir_name = lazy(
let ezfio_filename =

View File

@ -17,9 +17,9 @@ let initialization_timeout = 600.
let bind_socket ~socket_type ~socket ~address =
try
Zmq.Socket.bind socket address
with
| Unix.Unix_error (_, message, f) ->
failwith @@ Printf.sprintf
with
| Unix.Unix_error (_, message, f) ->
failwith @@ Printf.sprintf
"\n%s\nUnable to bind the dataserver's %s socket :\n %s\n%s"
f socket_type address message
| other_exception -> raise other_exception
@ -30,7 +30,7 @@ let run ?(daemon=true) ezfio_filename =
Qputils.set_ezfio_filename ezfio_filename;
(** Check if walkers need to be created. *)
let () =
let () =
if ( not(Ezfio.has_electrons_elec_coord_pool ()) ) then
begin
Printf.printf "Generating initial walkers...\n%!";
@ -39,12 +39,12 @@ let run ?(daemon=true) ezfio_filename =
Unix.execvp
(Lazy.force Qmcchem_config.qmc_create_walkers)
[|"qmc_create_walkers" ; ezfio_filename|]
| pid ->
| pid ->
begin
ignore @@ Unix.waitpid [] pid;
Printf.printf "Initial walkers ready\n%!"
end
end
end
in
@ -74,7 +74,7 @@ let run ?(daemon=true) ezfio_filename =
in
(** Status variable (mutable) *)
let status =
let status =
ref Status.Queued
in
@ -91,7 +91,7 @@ let run ?(daemon=true) ezfio_filename =
let adress_prefix =
"tcp://*:"
in
let result =
let result =
List.fold_left (fun accu i ->
let address =
adress_prefix ^ (string_of_int (n+i))
@ -99,14 +99,14 @@ let run ?(daemon=true) ezfio_filename =
let socket =
Zmq.Socket.create zmq_context Zmq.Socket.rep
in
let result =
let result =
try
Zmq.Socket.bind socket address;
accu
with
| _ -> false
in
Zmq.Socket.close socket;
Zmq.Socket.close socket;
result
) true [0;1;2;3]
in
@ -120,7 +120,7 @@ let run ?(daemon=true) ezfio_filename =
(** Random port number between 49152 and 65535 *)
let port =
let newport =
let newport =
ref ( 1024 + (Random.int (49151-1024)))
in
while ((check_port !newport) = `Unavailable)
@ -131,7 +131,7 @@ let run ?(daemon=true) ezfio_filename =
in
let debug_socket =
let debug_socket =
Zmq.Socket.create zmq_context Zmq.Socket.xpub
and address =
Printf.sprintf "tcp://*:%d" (port+4)
@ -140,17 +140,17 @@ let run ?(daemon=true) ezfio_filename =
Zmq.Socket.set_linger_period debug_socket 100 ;
let close_debug_socket () =
Zmq.Socket.close debug_socket
Zmq.Socket.close debug_socket
in
(** Sends a log text to the debug socket *)
let send_log socket size t0 text =
let dt =
let dt =
delta_t t0
in
let message =
Printf.sprintf "%20s : %8d : %10s : %e"
socket size text dt
let message =
Printf.sprintf "%20s : %8d : %10s : %e"
socket size text dt
in
Zmq.Socket.send debug_socket message
in
@ -159,7 +159,7 @@ let run ?(daemon=true) ezfio_filename =
(** {2 Walkers} *)
(** Number of electrons *)
let elec_num =
let elec_num =
Lazy.force Qputils.elec_num
in
@ -181,14 +181,14 @@ let run ?(daemon=true) ezfio_filename =
in
(** Array of walkers. The size is [walk_num_tot]. *)
let walkers_array =
let t0 =
let walkers_array =
let t0 =
Unix.gettimeofday ()
in
let j =
3*elec_num + 3
in
let result =
let result =
let size =
Ezfio.get_electrons_elec_coord_pool_size ()
and ez =
@ -204,7 +204,7 @@ let run ?(daemon=true) ezfio_filename =
failwith "Walkers file is broken."
in
String.concat " " [ "Read" ; string_of_int (Array.length result) ;
"walkers"]
"walkers"]
|> send_log "status" 0 t0 ;
result
in
@ -217,7 +217,7 @@ let run ?(daemon=true) ezfio_filename =
(** Last time when the walkers were saved to disk. *)
let last_save_walkers =
let last_save_walkers =
ref 0.
in
@ -227,14 +227,12 @@ let run ?(daemon=true) ezfio_filename =
if (delta_t !last_save_walkers > 10. ) then
begin
Ezfio.set_electrons_elec_coord_pool_size walk_num_tot ;
let walkers_list =
Array.map Array.to_list walkers_array
|> Array.to_list
|> List.concat
|> List.rev_map float_of_string
|> List.rev
let walkers_list =
Array.to_list walkers_array
|> Array.concat
|> Array.map float_of_string
in
Ezfio.set_electrons_elec_coord_pool (Ezfio.ezfio_array_of_list
Ezfio.set_electrons_elec_coord_pool (Ezfio.ezfio_array_of_array
~rank:3 ~dim:[| elec_num+1 ; 3 ; walk_num_tot |] ~data:walkers_list);
let t0 =
Unix.gettimeofday ()
@ -263,7 +261,7 @@ let run ?(daemon=true) ezfio_filename =
(** {2 Set of workers} *)
(** A hash table is kept to track the running workers. The keys are the
(** A hash table is kept to track the running workers. The keys are the
built as string containing the couple ([Compute_node], [PID]), and
the values are the last communication time.
*)
@ -271,16 +269,16 @@ let run ?(daemon=true) ezfio_filename =
(** The hash table for workers *)
let workers_hash =
let workers_hash =
Hashtbl.create 63
in
(** Creates a key using the couple ([Compute_node], [PID]) *)
let key compute_node pid =
String.concat " " [
(Compute_node.to_string compute_node);
let key compute_node pid =
String.concat " " [
(Compute_node.to_string compute_node);
(string_of_int pid) ]
in
@ -294,7 +292,7 @@ let run ?(daemon=true) ezfio_filename =
in
match Hashtbl.find_opt workers_hash s with
| Some _ -> failwith (s^" already registered")
| None -> Hashtbl.add workers_hash s (Unix.gettimeofday ())
| None -> Hashtbl.add workers_hash s (Unix.gettimeofday ())
in
@ -307,14 +305,14 @@ let run ?(daemon=true) ezfio_filename =
in
match Hashtbl.find_opt workers_hash s with
| Some x -> Hashtbl.remove workers_hash s
| None -> failwith (s^" not registered")
| None -> failwith (s^" not registered")
in
(** Sets the last access of the worker to [Unix.gettimeofday ()] *)
let touch_worker w pid =
let s =
let touch_worker w pid =
let s =
key w pid
in
Hashtbl.replace workers_hash s (Unix.gettimeofday ())
@ -326,11 +324,11 @@ let run ?(daemon=true) ezfio_filename =
let delta =
initialization_timeout +. block_time *. 2.
in
Hashtbl.fold (fun k v accu ->
Hashtbl.fold (fun k v accu ->
if (now -. v) <= delta then
v :: accu
else
accu ) hash []
accu ) hash []
|> List.length
in
@ -344,14 +342,14 @@ let run ?(daemon=true) ezfio_filename =
(** Name of the blocks file written by the current process. *)
let block_channel_filename =
let dirname =
let block_channel_filename =
let dirname =
Lazy.force Block.dir_name
in
let () =
let () =
if not ( Sys.file_exists dirname ) then
Unix.mkdir dirname 0o755
in
in
Filename.concat dirname (
hostname ^ "." ^ (string_of_int dataserver_pid)
)
@ -387,7 +385,7 @@ let run ?(daemon=true) ezfio_filename =
the compute nodes. Happens when [max_file_size] is reached.
*)
let compress_block_file filename =
let t0 =
let t0 =
Unix.gettimeofday ()
in
close_out !block_channel;
@ -406,16 +404,16 @@ let run ?(daemon=true) ezfio_filename =
(** {3 Status thread} *)
let start_status_thread =
let start_status_thread =
let t0 =
Unix.gettimeofday ()
in
Thread.create (fun () ->
Thread.create (fun () ->
send_log "status" 0 t0 "Starting status thread";
let socket =
Zmq.Socket.create zmq_context Zmq.Socket.pub
and address =
and address =
Printf.sprintf "tcp://*:%d" (port+1)
in
bind_socket "PUB" socket address;
@ -423,15 +421,15 @@ let run ?(daemon=true) ezfio_filename =
and delay_read = 2.
in
let start_time =
let start_time =
Unix.gettimeofday ()
and stop_time =
and stop_time =
ref (Input.Stop_time.(read () |> to_float) )
in
let last_update =
ref start_time
in
in
while (!status <> Status.Stopped)
do
@ -439,46 +437,46 @@ let run ?(daemon=true) ezfio_filename =
let now =
Unix.gettimeofday ()
in
let status_string =
let status_string =
Status.to_string !status
in
Zmq.Socket.send socket status_string;
send_log "status" (String.length status_string) now status_string;
let test =
let test =
if (now -. !last_update > delay_read) then
let n_connect =
n_connected workers_hash now
let n_connect =
n_connected workers_hash now
in
`Update n_connect
else if (now -. start_time > !stop_time) then
`Terminate
`Terminate
else if (now -. start_time > initialization_timeout) then
`Timeout
`Timeout
else
`None
`None
in
match (daemon, !status, test) with
| (_ , _ , `None ) -> ()
| (_ , Status.Running , `Terminate ) -> change_status Status.Stopping
| (false, Status.Running , `Update 0 ) -> change_status Status.Stopped
| (true , Status.Running , `Update 0 ) -> change_status Status.Queued
| (_ , _ , `Update i ) ->
| (true , Status.Running , `Update 0 ) -> change_status Status.Queued
| (_ , _ , `Update i ) ->
begin
status := Status.read ();
last_update := now;
stop_time := Input.Stop_time.(read () |> to_float) ;
let n_tot =
Hashtbl.length workers_hash
Hashtbl.length workers_hash
in
if (i <> n_tot) then
if (i <> n_tot) then
begin
Printf.sprintf "Connected workers : %d / %d" i n_tot
|> send_log "status" 0 now
|> send_log "status" 0 now
end
end
| (false, Status.Queued , `Timeout ) -> change_status Status.Stopped
| (false, Status.Queued , `Timeout ) -> change_status Status.Stopped
| (_, _, _) -> ()
;
done;
@ -487,37 +485,37 @@ let run ?(daemon=true) ezfio_filename =
Zmq.Socket.close socket
)
in
(** {3 Log thread} *)
let start_log_thread =
let start_log_thread =
let t0 =
Unix.gettimeofday ()
in
Thread.create (fun () ->
Thread.create (fun () ->
send_log "status" 0 t0 "Starting log thread";
let socket =
Zmq.Socket.create zmq_context Zmq.Socket.xsub
and address =
and address =
Printf.sprintf "tcp://*:%d" (port+3)
in
bind_socket "XSUB" socket address;
let pollitem =
Zmq.Poll.mask_of
Zmq.Poll.mask_of
[| (socket , Zmq.Poll.In) ;
(debug_socket , Zmq.Poll.In)
(debug_socket , Zmq.Poll.In)
|]
in
while (!status <> Status.Stopped)
do
let polling =
Zmq.Poll.poll ~timeout:1000 pollitem
Zmq.Poll.poll ~timeout:1000 pollitem
in
if (polling.(0) = Some Zmq.Poll.In) then
begin
let message =
let message =
Zmq.Socket.recv_all ~block:false socket
|> String.concat " "
in
@ -530,7 +528,7 @@ let run ?(daemon=true) ezfio_filename =
begin
(* Forward subscription from XPUB to XSUB *)
Zmq.Socket.recv_all ~block:false debug_socket
|> Zmq.Socket.send_all socket
|> Zmq.Socket.send_all socket
end
done;
Zmq.Socket.set_linger_period socket 1000 ;
@ -539,17 +537,17 @@ let run ?(daemon=true) ezfio_filename =
in
(** {3 Main thread} *)
let random_walkers n_walks =
let random_walkers n_walks =
let rec walkers accu = function
| 0 -> accu
| n ->
| 0 -> accu
| n ->
let random_int =
Random.int (Strictly_positive_int.to_int n_walks)
in
let new_accu =
let new_accu =
walkers_array.(random_int) :: accu
in
walkers new_accu (n-1)
walkers new_accu (n-1)
in
walkers [] (Strictly_positive_int.to_int n_walks)
|> Array.concat
@ -560,21 +558,21 @@ let run ?(daemon=true) ezfio_filename =
let wall0 =
Unix.gettimeofday ()
in
let f () =
let f () =
change_status Status.Queued;
send_log "status" 0 wall0 "Starting main thread";
(** Reply socket *)
let rep_socket =
Zmq.Socket.create zmq_context Zmq.Socket.rep
and address =
and address =
Printf.sprintf "tcp://*:%d" port
in
bind_socket "REP" rep_socket address;
Zmq.Socket.set_receive_high_water_mark rep_socket 100_000;
Zmq.Socket.set_send_high_water_mark rep_socket 100_000;
Zmq.Socket.set_immediate rep_socket true;
Zmq.Socket.set_immediate rep_socket true;
Zmq.Socket.set_linger_period rep_socket 600_000 ;
(** EZFIO Cache *)
@ -595,14 +593,14 @@ let run ?(daemon=true) ezfio_filename =
in
List.iter (fun x ->
if handle_ezfio ("has_"^x) = "T" then
try ignore @@ handle_ezfio ("get_"^x)
try ignore @@ handle_ezfio ("get_"^x)
with Failure _ -> ())
Qptypes.all_ezfio_messages;
(** Pull socket for computed data *)
let pull_socket =
Zmq.Socket.create zmq_context Zmq.Socket.pull
and address =
and address =
Printf.sprintf "tcp://*:%d" (port+2)
in
bind_socket "PULL" pull_socket address;
@ -611,7 +609,7 @@ let run ?(daemon=true) ezfio_filename =
(** Address of the dataserver *)
let server_address =
let ip =
Lazy.force Qmcchem_config.ip_address
Lazy.force Qmcchem_config.ip_address
in
Printf.sprintf "tcp://%s:%d" ip port
in
@ -621,7 +619,7 @@ let run ?(daemon=true) ezfio_filename =
(** Polling item to poll REP and PULL sockets. *)
let pollitem =
Zmq.Poll.mask_of
Zmq.Poll.mask_of
[| ( rep_socket, Zmq.Poll.In) ;
( pull_socket, Zmq.Poll.In) ;
|]
@ -629,27 +627,27 @@ let run ?(daemon=true) ezfio_filename =
(** Handles messages coming into the REP socket. *)
let handle_rep () =
let handle_rep () =
let raw_msg =
Zmq.Socket.recv_all ~block:false rep_socket
in
let t0 =
let t0 =
Unix.gettimeofday ()
in
let msg =
List.rev_map String.trim raw_msg
|> List.rev
|> Message.create
|> Message.create
and msg_size =
List.fold_left (fun accu x -> accu + (String.length x)) 0 raw_msg
in
let handle = function
| Message.Error _ -> ()
| Message.Ezfio ezfio_msg ->
let result =
| Message.Ezfio ezfio_msg ->
let result =
handle_ezfio ezfio_msg
in
Zmq.Socket.send_all rep_socket
Zmq.Socket.send_all rep_socket
[ String.length result
|> Printf.sprintf "%d " ;
result ] ;
@ -657,53 +655,53 @@ let run ?(daemon=true) ezfio_filename =
| Message.GetWalkers n_walks ->
begin
send_log "req" msg_size t0 "get_walkers";
let result =
let result =
random_walkers n_walks
in
Zmq.Socket.send_all rep_socket result;
send_log "rep" walkers_size t0 "get_walkers"
end
| Message.Register (w,pid) ->
end
| Message.Register (w,pid) ->
begin
match !status with
| Status.Queued
| Status.Queued
| Status.Running ->
begin
String.concat " " [ "Register :" ;
Compute_node.to_string w ;
String.concat " " [ "Register :" ;
Compute_node.to_string w ;
string_of_int pid ]
|> send_log "req" msg_size t0;
add_worker w pid;
if (!status = Status.Queued) then
change_status Status.Running ;
Zmq.Socket.send rep_socket "OK";
send_log "rep" 2 t0 "Register : OK"
send_log "rep" 2 t0 "Register : OK"
end
| Status.Stopping
| Status.Stopping
| Status.Stopped ->
Zmq.Socket.send rep_socket "Failed";
end
| Message.Unregister (w,pid) ->
| Message.Unregister (w,pid) ->
begin
String.concat " " [ "Unregister :" ;
(Compute_node.to_string w) ;
(Compute_node.to_string w) ;
(string_of_int pid) ]
|> send_log "req" msg_size t0;
Zmq.Socket.send rep_socket "OK";
del_worker w pid;
String.concat " " [ "Unregister :";
String.concat " " [ "Unregister :";
(Hashtbl.length workers_hash) |> string_of_int ;
"remaining" ]
"remaining" ]
|> send_log "rep" 2 t0 ;
let n_connect =
let n_connect =
n_connected workers_hash t0
in
match (daemon,n_connect) with
match (daemon,n_connect) with
| (false,0) -> change_status Status.Stopped
| (true ,0) -> change_status Status.Queued
| (true ,0) -> change_status Status.Queued
| _ -> ()
end
| Message.Test ->
| Message.Test ->
begin
Zmq.Socket.send rep_socket "OK";
send_log "rep" 2 t0 "Test"
@ -719,17 +717,17 @@ let run ?(daemon=true) ezfio_filename =
let raw_msg =
Zmq.Socket.recv_all ~block:false pull_socket
in
let t0 =
let t0 =
Unix.gettimeofday ()
in
let msg =
List.rev_map String.trim raw_msg
|> List.rev
|> Message.create
|> Message.create
and msg_size =
List.fold_left (fun accu x -> accu + (String.length x)) 0 raw_msg
in
let recv_log =
let recv_log =
send_log "pull" msg_size t0
in
@ -739,7 +737,7 @@ let run ?(daemon=true) ezfio_filename =
begin
if (status = Status.Running) then
touch_worker h pid ;
let log_msg =
let log_msg =
Printf.sprintf "Walkers from %s : %d / %d / %d"
(key h pid) (Array.length w) (!last_walker) walk_num_tot
in
@ -754,10 +752,10 @@ let run ?(daemon=true) ezfio_filename =
(Unix.gettimeofday () -. wall0)
1. (Property.to_string Property.Wall)
hostname (string_of_int dataserver_pid) 1
|> Block.of_string
|> Block.of_string
in
match wall with
| Some wall ->
| Some wall ->
begin
output_string !block_channel (Block.to_string wall);
output_char !block_channel '\n';
@ -777,7 +775,7 @@ let run ?(daemon=true) ezfio_filename =
| Message.Ezfio _
| Message.Register (_, _)
| Message.Unregister (_, _)
-> failwith "Bad message"
-> failwith "Bad message"
in handle msg
in
@ -785,18 +783,18 @@ let run ?(daemon=true) ezfio_filename =
while (!status <> Status.Stopped)
do
let polling =
Zmq.Poll.poll ~timeout:1000 pollitem
Zmq.Poll.poll ~timeout:1000 pollitem
in
match polling.(1) with
| Some Zmq.Poll.In -> handle_pull !status
| _ ->
| _ ->
begin
match polling.(0) with
| Some Zmq.Poll.In -> handle_rep ()
| _ ->
| _ ->
begin
flush !block_channel ;
let file_size =
let file_size =
(Unix.stat block_channel_filename_locked).Unix.st_size
in
if (file_size > !max_file_size) then
@ -811,13 +809,13 @@ let run ?(daemon=true) ezfio_filename =
List.iter (fun socket ->
Zmq.Socket.set_linger_period socket 1000 ;
Zmq.Socket.close socket)
[ rep_socket ; pull_socket ]
[ rep_socket ; pull_socket ]
in
Thread.create f
in
(** {2 Finalization} *)
(** Cleans all the open files, sockets, etc.
@ -842,16 +840,16 @@ let run ?(daemon=true) ezfio_filename =
(** {3 Main function} *)
let t0 =
let t0 =
Unix.gettimeofday ()
in
(* Handle signals *)
let handler s =
let handler s =
Printf.printf "Dataserver received signal %d... killing\n%!" s;
Watchdog.kill ();
in
List.iter (fun s -> ignore @@ Sys.signal s (Sys.Signal_handle handler))
List.iter (fun s -> ignore @@ Sys.signal s (Sys.Signal_handle handler))
[
Sys.sigint ;
Sys.sigterm ;
@ -864,7 +862,7 @@ let run ?(daemon=true) ezfio_filename =
begin
try
(List.iter Thread.join
(List.iter Thread.join
[ start_status_thread () ;
start_log_thread () ;
start_main_thread () ;

View File

@ -1,13 +1,13 @@
open Qptypes
type t =
type t =
{ property : Property.t ;
data : Block.t list;
}
module Average = struct
include Sample
include Sample
end
module Error = struct
@ -19,42 +19,42 @@ module Variance = struct
end
module Skewness: sig
type t
type t
val to_float : t -> float
val of_float : float -> t
val to_string : t -> string
end = struct
type t = float
type t = float
let to_string = string_of_float
let to_float x = x
let of_float x = x
end
module Kurtosis: sig
type t
type t
val to_float : t -> float
val of_float : float -> t
val to_string : t -> string
end = struct
type t = float
type t = float
let to_string = string_of_float
let to_float x = x
let of_float x = x
end
module GaussianDist: sig
type t
type t
val create : mu:Average.t -> sigma2:Variance.t -> t
val eval : g:t -> x:float -> float
end = struct
type t = { mu: Average.t ; sigma2: Variance.t }
type t = { mu: Average.t ; sigma2: Variance.t }
let create ~mu ~sigma2 =
{ mu ; sigma2 }
let eval ~g ~x =
let { mu ; sigma2 } =
let { mu ; sigma2 } =
g
in
let mu =
let mu =
Average.to_float mu
and sigma2 =
Variance.to_float sigma2
@ -62,10 +62,10 @@ end = struct
let x2 =
(x -. mu) *. ( x -. mu) /. sigma2
in
let pi =
let pi =
acos (-1.)
in
let c =
let c =
1. /. (sqrt (sigma2 *. (pi +. pi)))
in
c *. exp ( -0.5 *. x2)
@ -78,7 +78,7 @@ let hashtbl_to_alist table =
let hashtbl_change table key f =
let elt =
try
try
Some (Hashtbl.find table key)
with
| Not_found -> None
@ -91,7 +91,7 @@ let hashtbl_change table key f =
(** Build from raw data. Range values are given in percent. *)
let of_raw_data ?(locked=true) ~range property =
let data =
let data =
Block.raw_data ~locked ()
|> List.filter (fun x -> x.Block.property = property)
|> List.sort (fun x y ->
@ -109,7 +109,7 @@ let of_raw_data ?(locked=true) ~range property =
(Weight.to_float x.Block.weight) +. accu
) 0. data
in
let wmin, wmax =
rmin *. total_weight *. 0.01,
rmax *. total_weight *. 0.01
@ -128,13 +128,13 @@ let of_raw_data ?(locked=true) ~range property =
(wsum_new, x::l)
else
(wsum_new, l)
end
end
) (0.,[]) data
in
List.rev new_data
in
let result =
let result =
match range with
| (0.,100.) -> { property ; data }
| (rmin,rmax) -> { property ; data=data_in_range rmin rmax }
@ -146,7 +146,7 @@ let of_raw_data ?(locked=true) ~range property =
(** Compute average *)
let average { property ; data } =
if Property.is_scalar property then
let (num,denom) =
let (num,denom) =
List.fold_left (fun (an, ad) x ->
let num =
(Weight.to_float x.Block.weight) *. (Sample.to_float x.Block.value)
@ -154,7 +154,7 @@ let average { property ; data } =
(Weight.to_float x.Block.weight)
in (an +. num, ad +. den)
) (0., 0.) data
in
in
num /. denom
|> Average.of_float
else
@ -163,15 +163,15 @@ let average { property ; data } =
| [] -> 1
| x :: tl -> Sample.dimension x.Block.value
in
let (num,denom) =
let (num,denom) =
List.fold_left (fun (an, ad) x ->
let num =
Array.map (fun y -> (Weight.to_float x.Block.weight) *. y)
(Sample.to_float_array x.Block.value)
(Sample.to_float_array x.Block.value)
and den = (Weight.to_float x.Block.weight)
in ( Array.mapi (fun i y -> y +. num.(i)) an , ad +. den)
in ( Array.mapi (fun i y -> y +. num.(i)) an , ad +. den)
) (Array.make dim 0. , 0.) data
in
in
let denom_inv =
1. /. denom
in
@ -180,22 +180,28 @@ let average { property ; data } =
(** Compute sum (for CPU/Wall time) *)
let sum { property ; data } =
List.fold_left (fun accu x ->
let num = (Weight.to_float x.Block.weight) *. (Sample.to_float x.Block.value)
in accu +. num
) 0. data
) 0. data
(** Calculation of the average and error bar *)
let ave_error { property ; data } =
let rec loop ~sum ~avsq ~ansum ~avsum ~n ?idx = function
| [] ->
(* sum: \sum_k x_k *. w_k
ansum: \sum_k w_k
avsum: \sum_k x_k *. w_k
avcu0: avsum / ansum
avsq: \sum_k (1. -. (w_k /. ansum_k)) *. (x_k -. avcu0)^2 *. w_k)
*)
let rec loop ~sum ~avsq ~ansum ~avsum ~n ?idx = function
| [] ->
begin
if (n > 0.) then
( Average.of_float (sum /. ansum),
@ -205,12 +211,8 @@ let ave_error { property ; data } =
end
| (x,w) :: tail ->
begin
let avcu0 =
avsum /. ansum
in
let xw =
x *. w
in
let avcu0 = avsum /. ansum in
let xw = x *. w in
let ansum, avsum, sum =
ansum +. w ,
avsum +. xw ,
@ -220,9 +222,9 @@ let ave_error { property ; data } =
~sum:sum
~avsq:(avsq +. (1. -. (w /. ansum)) *. (x -. avcu0)
*. (x -. avcu0) *. w)
~avsum:avsum
~avsum:avsum
~ansum:ansum
~n:(n +. 1.)
~n:(n +. 1.)
end
in
@ -242,7 +244,7 @@ let ave_error { property ; data } =
(Sample.to_float x.Block.value,
Weight.to_float x.Block.weight)
) data
|> ave_error_scalar
|> ave_error_scalar
else
match data with
| [] -> (Average.of_float 0., None)
@ -251,7 +253,7 @@ let ave_error { property ; data } =
head.Block.value
|> Sample.dimension
in
let result =
let result =
Array.init dim (fun idx ->
List.rev_map (fun x ->
(Sample.to_float ~idx x.Block.value,
@ -260,16 +262,16 @@ let ave_error { property ; data } =
|> ave_error_scalar
)
in
( Array.map (fun (x,_) -> Average.to_float x) result
|> Average.of_float_array ~dim ,
( Array.map (fun (x,_) -> Average.to_float x) result
|> Average.of_float_array ~dim ,
if (Array.length result < 2) then
None
else
Some (Array.map (function
Some (Array.map (function
| (_,Some y) -> Error.to_float y
| (_,None) -> 0.) result
| (_,None) -> 0.) result
|> Average.of_float_array ~dim)
)
)
@ -286,14 +288,14 @@ let fold_blocks ~f { property ; data } =
List.fold_left (fun accu block ->
let x = Sample.to_float block.Block.value
in f accu x
) init data
) init data
(** Convergence plot *)
let convergence { property ; data } =
let rec loop ~sum ~avsq ~ansum ~avsum ~n ~accu = function
let rec loop ~sum ~avsq ~ansum ~avsum ~n ~accu = function
| [] -> List.rev accu
| head :: tail ->
begin
@ -307,7 +309,7 @@ let convergence { property ; data } =
and avsum = avsum +. xw
and sum = sum +. xw
in
let accu =
let accu =
if (n > 0.) then
(sum /. ansum, sqrt ( abs_float ( avsq /.( ansum *. n))))::accu
else
@ -317,9 +319,9 @@ let convergence { property ; data } =
~sum:sum
~avsq:(avsq +. (1. -. (w /. ansum)) *. (x -. avcu0)
*. (x -. avcu0) *. w)
~avsum:avsum
~avsum:avsum
~ansum:ansum
~n:(n +. 1.)
~n:(n +. 1.)
~accu:accu
end
in
@ -365,7 +367,7 @@ let max_block =
(** Create a hash table for merging *)
let create_hash ~create_key ?(update_block_id=(fun x->x))
let create_hash ~create_key ?(update_block_id=(fun x->x))
?(update_value=(fun wc vc wb vb sw -> (wc *. vc +. wb *. vb) /. sw) )
?(update_weight=(fun wc wb -> wc +. wb) ) t =
let table = Hashtbl.create 63
@ -374,7 +376,7 @@ let create_hash ~create_key ?(update_block_id=(fun x->x))
let key = create_key block
in
let open Block in
hashtbl_change table key (function
hashtbl_change table key (function
| Some current ->
let wc, wb =
Weight.to_float current.weight,
@ -384,7 +386,7 @@ let create_hash ~create_key ?(update_block_id=(fun x->x))
update_weight wc wb
in
if (Property.is_scalar current.property) then
let vc, vb =
let vc, vb =
Sample.to_float current.value,
Sample.to_float block.value
in Some
@ -396,15 +398,15 @@ let create_hash ~create_key ?(update_block_id=(fun x->x))
compute_node = block.compute_node;
}
else
let vc, vb =
let vc, vb =
Sample.to_float_array current.value,
Sample.to_float_array block.value
and dim =
and dim =
Sample.dimension current.value
in Some
{ property = current.property ;
weight = Weight.of_float sw ;
value =
value =
Array.init dim (fun i -> update_value wc vc.(i) wb vb.(i) sw)
|> Sample.of_float_array ~dim ;
block_id = update_block_id block.block_id;
@ -443,15 +445,15 @@ let merge ~create_key ?update_block_id ?update_value ?update_weight t =
(** Merge per block id *)
let merge_per_block_id =
merge
merge
~create_key:(fun block -> Block_id.to_string block.Block.block_id)
(** Merge per compute_node *)
let merge_per_compute_node =
merge
~create_key:(fun block ->
Printf.sprintf "%s"
~create_key:(fun block ->
Printf.sprintf "%s"
(Compute_node.to_string block.Block.compute_node) )
@ -459,8 +461,8 @@ let merge_per_compute_node =
(** Merge per Compute_node and PID *)
let merge_per_compute_node_and_pid =
merge
~create_key:(fun block ->
Printf.sprintf "%s %10.10d"
~create_key:(fun block ->
Printf.sprintf "%s %10.10d"
(Compute_node.to_string block.Block.compute_node)
(block.Block.pid) )
@ -469,8 +471,8 @@ let merge_per_compute_node_and_pid =
(** Merge per Compute_node and BlockId *)
let merge_per_compute_node_and_block_id =
merge
~create_key:(fun block ->
Printf.sprintf "%s %10.10d"
~create_key:(fun block ->
Printf.sprintf "%s %10.10d"
(Compute_node.to_string block.Block.compute_node)
(Block_id.to_int block.Block.block_id) )
@ -510,48 +512,48 @@ let error_x_over_y = function
(** Create float, variable operators *)
let one_variable_operator ~update_value p f =
{ p with
data = List.rev @@ List.rev_map (fun b -> { b with
let one_variable_operator ~update_value p f =
{ p with
data = List.rev @@ List.rev_map (fun b -> { b with
Block.value = Sample.of_float (update_value (Sample.to_float b.Block.value) ) }
) p.data }
let ( +@ ) p f = one_variable_operator p f
~update_value: (fun x -> x +. f )
let ( *@ ) p f = one_variable_operator p f
let ( *@ ) p f = one_variable_operator p f
~update_value: (fun x -> x *. f )
let ( -@ ) p f = one_variable_operator p f
let ( -@ ) p f = one_variable_operator p f
~update_value: (fun x -> x -. f )
let ( /@ ) p f = one_variable_operator p f
let ( /@ ) p f = one_variable_operator p f
~update_value: (fun x -> x /. f )
(** Create two variable operators *)
let two_variable_operator ~update_value p1 p2 =
let two_variable_operator ~update_value p1 p2 =
merge
~update_value
~create_key:(fun block ->
Printf.sprintf "%s %10.10d %10.10d"
Printf.sprintf "%s %10.10d %10.10d"
(Compute_node.to_string block.Block.compute_node)
(Block_id.to_int block.Block.block_id)
(Block_id.to_int block.Block.block_id)
(block.Block.pid) )
~update_weight:(fun wc wb -> wc )
{ property = p1.property ;
data = List.concat [ p1.data ; p2.data ] }
data = List.concat [ p1.data ; p2.data ] }
let ( +! ) = two_variable_operator
let ( +! ) = two_variable_operator
~update_value: (fun wc vc wb vb sw -> (vc +. vb) )
let ( *! ) = two_variable_operator
let ( *! ) = two_variable_operator
~update_value: (fun wc vc wb vb sw -> (vc *. vb) )
let ( -! ) = two_variable_operator
let ( -! ) = two_variable_operator
~update_value: (fun wc vc wb vb sw -> (vc -. vb) )
let ( /! ) = two_variable_operator
let ( /! ) = two_variable_operator
~update_value: (fun wc vc wb vb sw -> (vc /. vb) )
@ -560,11 +562,11 @@ let ( /! ) = two_variable_operator
(** Merge two consecutive blocks *)
let compress =
merge
~create_key:(fun block ->
~create_key:(fun block ->
Printf.sprintf "%s %10.10d %10.10d"
(Compute_node.to_string block.Block.compute_node) block.Block.pid
(((Block_id.to_int block.Block.block_id)+1)/2))
~update_block_id:(fun block_id ->
~update_block_id:(fun block_id ->
((Block_id.to_int block_id)+1)/2
|> Block_id.of_int )
@ -576,15 +578,15 @@ let max_value_per_compute_node t =
let table = Hashtbl.create 63
in
let create_key block =
Printf.sprintf "%s %10.10d"
Printf.sprintf "%s %10.10d"
(Compute_node.to_string block.Block.compute_node)
(block.Block.pid)
(block.Block.pid)
in
List.iter (fun block ->
let key = create_key block
in
let open Block in
hashtbl_change table key (function
hashtbl_change table key (function
| Some current ->
let vc = Sample.to_float current.value
and vb = Sample.to_float block.value
@ -610,36 +612,36 @@ let max_value_per_compute_node t =
(** String representation *)
let to_string p =
let to_string p =
match p.property with
| Property.Cpu -> Printf.sprintf "%s" (Time.string_of_sec (sum p))
| Property.Wall -> Printf.sprintf "%s" (Time.string_of_sec (sum (max_value_per_compute_node p)))
| Property.Accep -> Printf.sprintf "%16.10f" (average p |> Average.to_float)
| _ ->
| _ ->
begin
if Property.is_scalar p.property then
match ave_error p with
| (ave, Some error) ->
let (ave, error) =
Average.to_float ave,
| (ave, Some error) ->
let (ave, error) =
Average.to_float ave,
Error.to_float error
in
Printf.sprintf "%16.10f +/- %16.10f" ave error
| (ave, None) ->
let ave =
| (ave, None) ->
let ave =
Average.to_float ave
in
Printf.sprintf "%16.10f" ave
else
match ave_error p with
| (ave, Some error) ->
| (ave, Some error) ->
let idxmax =
Average.dimension ave
in
let rec f accu idx =
if (idx < idxmax) then
let (ave, error) =
Average.to_float ~idx ave,
let (ave, error) =
Average.to_float ~idx ave,
Error.to_float ~idx error
in
let s =
@ -650,9 +652,9 @@ let to_string p =
accu
in
(f "[ \n" 0) ^ " ]"
| (ave, None) ->
| (ave, None) ->
Average.to_float ave
|> Printf.sprintf "%16.10f"