Parallelism: clean-up this commit

Author: arkocal

src/common/util/cilfacade.ml

@@ -114,8 +114,8 @@ end
 let cleanCilPrinter = new cleanCilPrinterClass
 
 let cleanDumpFile (pp: cilPrinter) (out : out_channel) (outfile: string) file =
-  Domain.DLS.set Pretty.printDepth 99999;
-  Domain.DLS.set Pretty.fastMode true;
+  Pretty.printDepth := 99999;
+  Pretty.fastMode := true;
   iterGlobals file (fun g -> dumpGlobal pp out g);
   flush out
 

src/solver/parallelStats.ml

@@ -21,11 +21,15 @@ struct
   let instant_returns = Atomic.make 0
   let thread_starts = Atomic.make 0
   let thread_creates = Atomic.make 0
+  let thread_start_times = Array.init maximal_number_of_threads (fun _ -> 0.0)
+  let thread_end_times = Array.init maximal_number_of_threads (fun _ -> 0.0)
 
   let active_threads = Atomic.make 0
   let first_thread_activation_time = Atomic.make 0.0
   let last_thread_activation_update_time = Atomic.make 0.0
   let total_thread_activation_time = Atomic.make 0.0
+  
+  let iterations = Atomic.make 0
 
   let updates = Atomic.make 0
   let updates_by_thread = Array.init maximal_number_of_threads (fun _ -> Atomic.make 0)
@@ -88,44 +92,46 @@ struct
   let rec thread_starts_solve_event thread_id =
     Atomic.incr thread_starts;
     let t = Unix.gettimeofday () in
-    if (Atomic.get first_thread_activation_time) = 0.0 then 
-      begin
-        let success = Atomic.compare_and_set first_thread_activation_time 0.0 t in
-        if (not success) then thread_starts_solve_event thread_id
-        else (
-          Atomic.set last_thread_activation_update_time t;
-          Atomic.incr active_threads;
-        )
-      end
-    else (
-      let last_thread_activation_update_time_f = Atomic.get last_thread_activation_update_time in
-      let success = Atomic.compare_and_set last_thread_activation_update_time last_thread_activation_update_time_f t in
-      if (not success) then thread_starts_solve_event thread_id
-      else
-        begin
-          let time_diff = t -. last_thread_activation_update_time_f in
-          let cpu_time_since_last_update = time_diff *. (float_of_int @@ Atomic.get active_threads) in
-          let current_total_thread_activation_time = Atomic.get total_thread_activation_time in
-          let new_total_thread_activation_time = current_total_thread_activation_time +. cpu_time_since_last_update in
-          Atomic.set total_thread_activation_time new_total_thread_activation_time;
-          Atomic.incr active_threads;
-        end
-    )
+    thread_start_times.(thread_id) <- t
+    (* if (Atomic.get first_thread_activation_time) = 0.0 then  *)
+    (*   begin *)
+    (*     let success = Atomic.compare_and_set first_thread_activation_time 0.0 t in *)
+    (*     if (not success) then thread_starts_solve_event thread_id *)
+    (*     else ( *)
+    (*       Atomic.set last_thread_activation_update_time t; *)
+    (*       Atomic.incr active_threads; *)
+    (*     ) *)
+    (*   end *)
+    (* else ( *)
+    (*   let last_thread_activation_update_time_f = Atomic.get last_thread_activation_update_time in *)
+    (*   let success = Atomic.compare_and_set last_thread_activation_update_time last_thread_activation_update_time_f t in *)
+    (*   if (not success) then thread_starts_solve_event thread_id *)
+    (*   else *)
+    (*     begin *)
+    (*       let time_diff = t -. last_thread_activation_update_time_f in *)
+    (*       let cpu_time_since_last_update = time_diff *. (float_of_int @@ Atomic.get active_threads) in *)
+    (*       let current_total_thread_activation_time = Atomic.get total_thread_activation_time in *)
+    (*       let new_total_thread_activation_time = current_total_thread_activation_time +. cpu_time_since_last_update in *)
+    (*       Atomic.set total_thread_activation_time new_total_thread_activation_time; *)
+    (*       Atomic.incr active_threads; *)
+    (*     end *)
+    (* ) *)
 
   let rec thread_ends_solve_event thread_id =
     let t = Unix.gettimeofday () in
-    let last_thread_activation_update_time_f = Atomic.get last_thread_activation_update_time in
-    let success = Atomic.compare_and_set last_thread_activation_update_time last_thread_activation_update_time_f t in
-    if (not success) then thread_ends_solve_event thread_id
-    else
-      begin
-        let time_diff = t -. last_thread_activation_update_time_f in
-        let cpu_time_since_last_update = time_diff *. (float_of_int @@ Atomic.get active_threads) in
-        let current_total_thread_activation_time = Atomic.get total_thread_activation_time in
-        let new_total_thread_activation_time = current_total_thread_activation_time +. cpu_time_since_last_update in
-        Atomic.set total_thread_activation_time new_total_thread_activation_time;
-        Atomic.decr active_threads;
-      end
+    thread_end_times.(thread_id) <- t
+    (* let last_thread_activation_update_time_f = Atomic.get last_thread_activation_update_time in *)
+    (* let success = Atomic.compare_and_set last_thread_activation_update_time last_thread_activation_update_time_f t in *)
+    (* if (not success) then thread_ends_solve_event thread_id *)
+    (* else *)
+    (*   begin *)
+    (*     let time_diff = t -. last_thread_activation_update_time_f in *)
+    (*     let cpu_time_since_last_update = time_diff *. (float_of_int @@ Atomic.get active_threads) in *)
+    (*     let current_total_thread_activation_time = Atomic.get total_thread_activation_time in *)
+    (*     let new_total_thread_activation_time = current_total_thread_activation_time +. cpu_time_since_last_update in *)
+    (*     Atomic.set total_thread_activation_time new_total_thread_activation_time; *)
+    (*     Atomic.decr active_threads; *)
+    (*   end *)
 
 
   (* solvers can assign this to print solver specific statistics using their data structures *)
@@ -169,21 +175,38 @@ struct
     let threads_data = Seq.zip threads_data (Array.to_seq updates_by_thread) in
     let threads_with_update = ref 0 in
     let threads_with_anything = ref 0 in
-    Seq.iteri (fun i ((vars, evals), updates) -> 
-        if (Atomic.get vars) <> 0 || (Atomic.get evals) > 0 || (Atomic.get updates) <> 0 then
-          begin
-            Logs.info "Thread %d: vars: %d, evals: %d, updates: %d" i (Atomic.get vars) (Atomic.get evals) (Atomic.get updates);
-            threads_with_anything := !threads_with_anything + 1;
-          end;
-        if (Atomic.get updates) <> 0 then threads_with_update := !threads_with_update + 1
-      ) threads_data;
+    (* The following is an important statistics, here left out for simplicity *)
+    (* Seq.iteri (fun i ((vars, evals), updates) ->  *)
+    (*     if (Atomic.get vars) <> 0 || (Atomic.get evals) > 0 || (Atomic.get updates) <> 0 then *)
+    (*       begin *)
+    (*         Logs.info "Thread %d: vars: %d, evals: %d, updates: %d" i (Atomic.get vars) (Atomic.get evals) (Atomic.get updates); *)
+    (*         threads_with_anything := !threads_with_anything + 1; *)
+    (*       end; *)
+    (*     if (Atomic.get updates) <> 0 then threads_with_update := !threads_with_update + 1 *)
+    (*   ) threads_data; *)
     Logs.info "Threads with updates: %d" !threads_with_update;
     Logs.info "Threads with anything: %d" !threads_with_anything;
     Logs.info "Threads returned instantly: %d" (Atomic.get instant_returns);
     Logs.info "Threads started: %d" (Atomic.get thread_starts);
     Logs.info "Threads created: %d" (Atomic.get thread_creates);
     Logs.info "Threads active: %d" (Atomic.get active_threads);
 
+    let non_zero_start_times = Array.filter (fun t -> t <> 0.0) thread_start_times in
+    let first_thread_start = Array.fold_left min infinity non_zero_start_times in
+    let current_time = Unix.gettimeofday () in
+    let total_runtime = Seq.zip (Array.to_seq thread_start_times) (Array.to_seq thread_end_times) 
+      |> Seq.filter (fun (start, end_) -> start <> 0.0) 
+      |> Seq.map (fun (start, end_) -> if end_ = 0.0 then (start, current_time) else (start, end_))
+      |> Seq.fold_left (fun acc (start, end_) -> acc +. (end_ -. start)) 0.0 in
+    let walltime = current_time -. first_thread_start in
+    Logs.info "Total runtime: %f" total_runtime;
+    Logs.info "Walltime: %f" walltime;
+    let average_active_threads = total_runtime /. walltime in
+    Logs.info "Average active threads: %f" average_active_threads;
+      
+
+    
+
     (* Logs.info "vars: %d" (Atomic.get vars); *)
 
     (* Array.iteri (fun i v -> if Atomic.get v <> 0 then Logs.info "vars (%d): %d" i (Atomic.get v)) vars_by_thread; *)

src/solver/td_parallel_base.ml

@@ -30,7 +30,10 @@ module CasWithStatAndException = struct
     if Atomic.compare_and_set key old new_ then
       Atomic.incr count_success
     else
-      raise CasFailException
+      begin
+        Atomic.incr count_failure;
+        raise CasFailException
+      end
 end
 
 module Base : GenericCreatingEqSolver =
@@ -54,14 +57,17 @@ module Base : GenericCreatingEqSolver =
         stable: bool;
         called: bool;
         root: bool; (** If this variable was the starting point of a solver thread *)
+        id: int; (** Just for experimentation *)
       }
-      let default () = {value = S.Dom.bot (); infl = VS.empty; called = false; stable = false; wpoint = false; root = false}
+      let default () = {value = S.Dom.bot (); infl = VS.empty; called = false; stable = false; wpoint = false; root = false; id=0}
       let to_string s = S.Dom.show s.value
     end
 
     (** Concurrency safe hashmap for the state of the unknowns. *)
     module CM = Data.SafeHashmap (S.Var) (DefaultState) (HM)
 
+    let var_count = Atomic.make 0
+
     let create_empty_data () = CM.create () 
 
     (* TODO this map should also be thread safe, or secured with a lock *)
@@ -77,9 +83,9 @@ module Base : GenericCreatingEqSolver =
     let cas = Data.CasWithStat.cas
 
     let print_data data =
-      Logs.debug "CAS success: %d" (Atomic.get Data.CasWithStat.count_success);
-      Logs.debug "CAS failure: %d" (Atomic.get Data.CasWithStat.count_failure);
-      Logs.debug "CAS success rate: %f" (float_of_int (Atomic.get Data.CasWithStat.count_success) /. (float_of_int (Atomic.get Data.CasWithStat.count_success + Atomic.get Data.CasWithStat.count_failure)))
+      Logs.info "CAS success: %d" (Atomic.get CasWithStatAndException.count_success);
+      Logs.info "CAS failure: %d" (Atomic.get CasWithStatAndException.count_failure);
+      Logs.info "CAS success rate: %f" (float_of_int (Atomic.get CasWithStatAndException.count_success) /. (float_of_int (Atomic.get CasWithStatAndException.count_success + Atomic.get CasWithStatAndException.count_failure)))
 
     let print_data_verbose data str =
       if Logs.Level.should_log Debug then (
@@ -108,6 +114,9 @@ module Base : GenericCreatingEqSolver =
       let init x thread_id =
         let value, was_created = CM.find_create data x in
         if (was_created) then new_var_event thread_id x;
+        let var_id = Atomic.fetch_and_add var_count 1 in
+        let s = Atomic.get value in
+        Atomic.set value {s with id = var_id};
         value
       in
 
@@ -176,7 +185,7 @@ module Base : GenericCreatingEqSolver =
               if tracing then trace "thread_pool" "starting task %d to iterate %a" job_id S.Var.pretty_trace y;
               thread_starts_solve_event job_id;
               let inner_prom = ref [] in
-              iterate None inner_prom y job_id y_atom;
+              iterate None inner_prom y [] job_id y_atom;
               HM.remove unknowns_with_running_jobs y;
               thread_ends_solve_event job_id;
               Thread_pool.await_all pool (!inner_prom);
@@ -199,13 +208,14 @@ module Base : GenericCreatingEqSolver =
           @param job_id The id of the thread that is solving for x.
           @param x_atom The atomic reference to the state of x, to prevent unnecessary lookups.
       *)
-      and iterate orig prom x job_id x_atom = (* ~(inner) solve in td3*)
+      and iterate orig prom x ichain job_id x_atom = (* ~(inner) solve in td3*)
 
         (** Get the value for y, triggering an iteration if necessary, and performing a lookup otherwise. 
             @param x The variable whose query led to the query for y, so that the infl of y can be updated.
             @param y The variable to get the value for.
             @return The value of y.
         *)
+
         let rec query x y = (* ~eval in td3 *)
           (* Query with atomics: if anything is changed, query is repeated and the initial call *)
           (* has no side effects. Thus, imitating that the query just happend in a later point in time.*)
@@ -233,7 +243,8 @@ module Base : GenericCreatingEqSolver =
               ) else (
                 if tracing then trace "infl" "add_infl %a %a" S.Var.pretty_trace y S.Var.pretty_trace x;
                 cas y_atom y_state {y_state_with_infl with stable = true; called=true};
-                iterate (Some x) prom y job_id y_atom;
+                let ichain = y_state.id :: ichain in
+                iterate (Some x) prom y ichain job_id y_atom;
                 (Atomic.get y_atom).value
               )
             ) ) with CasFailException -> query x y
@@ -281,9 +292,19 @@ module Base : GenericCreatingEqSolver =
         in
 
         (* begining of iteration to update the value for x *)
+        (* let ichain_as_string = List.fold_left (fun acc x -> acc ^ (string_of_int x) ^ ".") "" in  *)
+        (* if tracing then trace "ichain" "%s" (ichain_as_string ichain); *)
         assert (not @@ is_global x);
         let cas = CasWithStatAndException.cas in
         let x_state = Atomic.get x_atom in
+
+        (match orig with
+           Some orig -> begin
+             let orig_atom = CM.find data orig in
+             let orig_state = Atomic.get orig_atom in
+             if tracing then trace "ilink" "%d,%d" orig_state.id x_state.id;
+           end
+         | None -> ());
         if tracing then trace "iter" "%d iterate %a, stable: %b, wpoint: %b" job_id S.Var.pretty_trace x x_state.stable x_state.wpoint;
         let x_is_widening_point = x_state.wpoint in (* if x becomes a wpoint during eq, checking this will delay widening until next iterate *)
         eval_rhs_event job_id x;
@@ -302,13 +323,13 @@ module Base : GenericCreatingEqSolver =
               | Some z -> (VS.add z x_state.infl)
               | None -> x_state.infl in
             let x_state_new = {x_state with infl = infl; called = false; wpoint = false} in
-            try (cas x_atom x_state x_state_new ) with CasFailException -> (iterate[@tailcall]) orig prom x job_id x_atom;
+            try (cas x_atom x_state x_state_new ) with CasFailException -> (iterate[@tailcall]) orig prom x ichain job_id x_atom;
           ) else (
             let x_state_new = {x_state with stable = true} in
             (* No need to track cas success, as we will iterate again anyway. *)
             ignore @@ Atomic.compare_and_set x_atom x_state x_state_new;
             if tracing then trace "iter" "iterate still unstable %a" S.Var.pretty_trace x;
-            (iterate[@tailcall]) orig prom x job_id x_atom
+            (iterate[@tailcall]) orig prom x ichain job_id x_atom
           )
         ) else (
           (* value has changed *)
@@ -336,11 +357,11 @@ module Base : GenericCreatingEqSolver =
                 let success = Atomic.compare_and_set x_atom x_state new_s in 
                 if success then (
                   if tracing then trace "iter" "iterate changed %a" S.Var.pretty_trace x;
-                  (iterate[@tailcall]) orig prom x job_id x_atom
+                  (iterate[@tailcall]) orig prom x ichain job_id x_atom
                 ) else (finalize[@tailcall]) ()
               ) in
             finalize ();
-          ) with CasFailException -> (iterate[@tailcall]) orig prom x job_id x_atom;
+          ) with CasFailException -> (iterate[@tailcall]) orig prom x ichain job_id x_atom;
         ) in
 
       let set_start (x,d) =
@@ -350,6 +371,7 @@ module Base : GenericCreatingEqSolver =
       in
 
       (* beginning of main solve *)
+
       start_event ();
 
       List.iter set_start st;
@@ -388,6 +410,7 @@ module Base : GenericCreatingEqSolver =
       (* After termination, only those variables are stable which are
        * - reachable from any of the queried variables vs, or
        * - effected by side-effects and have no constraints on their own (this should be the case for all of our analyses). *)
+      print_data ();
       print_stats ();
       stop_event ();