src/HOL/Tools/Sledgehammer/async_manager.ML
author blanchet
Mon Jun 02 17:34:26 2014 +0200 (2014-06-02 ago)
changeset 57158 f028d93798e6
parent 56303 4cc3f4db3447
child 57556 6180d81be977
permissions -rw-r--r--
simplified counterexample handling
     1 (*  Title:      HOL/Tools/Sledgehammer/async_manager.ML
     2     Author:     Fabian Immler, TU Muenchen
     3     Author:     Makarius
     4     Author:     Jasmin Blanchette, TU Muenchen
     5 
     6 Central manager for asynchronous diagnosis tool threads.
     7 *)
     8 
     9 (*Proof General legacy*)
    10 
    11 signature ASYNC_MANAGER =
    12 sig
    13   val break_into_chunks : string -> string list
    14   val thread :
    15     string -> Time.time -> Time.time -> string * string
    16     -> (unit -> bool * string) -> unit
    17   val kill_threads : string -> string -> unit
    18   val has_running_threads : string -> bool
    19   val running_threads : string -> string -> unit
    20   val thread_messages : string -> string -> int option -> unit
    21 end;
    22 
    23 structure Async_Manager : ASYNC_MANAGER =
    24 struct
    25 
    26 (** preferences **)
    27 
    28 val message_store_limit = 20
    29 val message_display_limit = 10
    30 
    31 
    32 (** thread management **)
    33 
    34 fun implode_message (workers, work) =
    35   space_implode " " (Try.serial_commas "and" workers) ^ work
    36 
    37 
    38 (* data structures over threads *)
    39 
    40 structure Thread_Heap = Heap
    41 (
    42   type elem = Time.time * Thread.thread
    43   fun ord ((a, _), (b, _)) = Time.compare (a, b)
    44 )
    45 
    46 fun lookup_thread xs = AList.lookup Thread.equal xs
    47 fun delete_thread xs = AList.delete Thread.equal xs
    48 fun update_thread xs = AList.update Thread.equal xs
    49 
    50 
    51 (* state of thread manager *)
    52 
    53 type state =
    54   {manager: Thread.thread option,
    55    timeout_heap: Thread_Heap.T,
    56    active:
    57      (Thread.thread
    58       * (string * Time.time * Time.time * (string * string))) list,
    59    canceling:  (Thread.thread * (string * Time.time * (string * string))) list,
    60    messages: (bool * (string * (string * string))) list,
    61    store: (string * (string * string)) list}
    62 
    63 fun make_state manager timeout_heap active canceling messages store : state =
    64   {manager = manager, timeout_heap = timeout_heap, active = active,
    65    canceling = canceling, messages = messages, store = store}
    66 
    67 val global_state = Synchronized.var "async_manager"
    68   (make_state NONE Thread_Heap.empty [] [] [] [])
    69 
    70 
    71 (* unregister thread *)
    72 
    73 fun unregister (urgent, message) thread =
    74   Synchronized.change global_state
    75   (fn state as {manager, timeout_heap, active, canceling, messages, store} =>
    76     (case lookup_thread active thread of
    77       SOME (tool, _, _, desc as (worker, its_desc)) =>
    78         let
    79           val active' = delete_thread thread active
    80           val now = Time.now ()
    81           val canceling' = (thread, (tool, now, desc)) :: canceling
    82           val message' =
    83             (worker, its_desc ^ (if message = "" then "" else "\n" ^ message))
    84           val messages' = (urgent, (tool, message')) :: messages
    85           val store' = (tool, message') ::
    86             (if length store <= message_store_limit then store
    87              else #1 (chop message_store_limit store))
    88         in make_state manager timeout_heap active' canceling' messages' store' end
    89     | NONE => state))
    90 
    91 
    92 (* main manager thread -- only one may exist *)
    93 
    94 val min_wait_time = seconds 0.3
    95 val max_wait_time = seconds 10.0
    96 
    97 fun replace_all bef aft =
    98   let
    99     fun aux seen "" = String.implode (rev seen)
   100       | aux seen s =
   101         if String.isPrefix bef s then
   102           aux seen "" ^ aft ^ aux [] (unprefix bef s)
   103         else
   104           aux (String.sub (s, 0) :: seen) (String.extract (s, 1, NONE))
   105   in aux [] end
   106 
   107 (* This is a workaround for Proof General's off-by-a-few sendback display bug,
   108    whereby "pr" in "proof" is not highlighted. *)
   109 val break_into_chunks = space_explode "\000" o replace_all "\n\n" "\000"
   110 
   111 fun print_new_messages () =
   112   Synchronized.change_result global_state
   113       (fn {manager, timeout_heap, active, canceling, messages, store} =>
   114           messages
   115           |> List.partition
   116                  (fn (urgent, _) =>
   117                      (null active andalso null canceling) orelse urgent)
   118           ||> (fn postponed_messages =>
   119                   make_state manager timeout_heap active canceling
   120                                      postponed_messages store))
   121   |> map (fn (_, (tool, (worker, work))) => ((tool, work), worker))
   122   |> AList.group (op =)
   123   |> List.app (fn ((_, ""), _) => ()
   124                 | ((tool, work), workers) =>
   125                   tool ^ ": " ^
   126                   implode_message (workers |> sort_distinct string_ord, work)
   127                   |> break_into_chunks
   128                   |> List.app Output.urgent_message)
   129 
   130 fun check_thread_manager () = Synchronized.change global_state
   131   (fn state as {manager, timeout_heap, active, canceling, messages, store} =>
   132     if (case manager of SOME thread => Thread.isActive thread | NONE => false) then state
   133     else let val manager = SOME (Runtime.thread false (fn () =>
   134       let
   135         fun time_limit timeout_heap =
   136           (case try Thread_Heap.min timeout_heap of
   137             NONE => Time.+ (Time.now (), max_wait_time)
   138           | SOME (time, _) => time)
   139 
   140         (*action: find threads whose timeout is reached, and interrupt canceling threads*)
   141         fun action {manager, timeout_heap, active, canceling, messages, store} =
   142           let val (timeout_threads, timeout_heap') =
   143             Thread_Heap.upto (Time.now (), Thread.self ()) timeout_heap
   144           in
   145             if null timeout_threads andalso null canceling then
   146               NONE
   147             else
   148               let
   149                 val _ = List.app (Simple_Thread.interrupt_unsynchronized o #1) canceling
   150                 val canceling' = filter (Thread.isActive o #1) canceling
   151                 val state' = make_state manager timeout_heap' active canceling' messages store
   152               in SOME (map #2 timeout_threads, state') end
   153           end
   154       in
   155         while Synchronized.change_result global_state
   156           (fn state as {timeout_heap, active, canceling, messages, store, ...} =>
   157             if null active andalso null canceling andalso null messages
   158             then (false, make_state NONE timeout_heap active canceling messages store)
   159             else (true, state))
   160         do
   161           (Synchronized.timed_access global_state
   162                (SOME o time_limit o #timeout_heap) action
   163            |> these
   164            |> List.app (unregister (false, "Timed out."));
   165            print_new_messages ();
   166            (* give threads some time to respond to interrupt *)
   167            OS.Process.sleep min_wait_time)
   168       end))
   169     in make_state manager timeout_heap active canceling messages store end)
   170 
   171 
   172 (* register thread *)
   173 
   174 fun register tool birth_time death_time desc thread =
   175  (Synchronized.change global_state
   176     (fn {manager, timeout_heap, active, canceling, messages, store} =>
   177       let
   178         val timeout_heap' = Thread_Heap.insert (death_time, thread) timeout_heap
   179         val active' = update_thread (thread, (tool, birth_time, death_time, desc)) active
   180         val state' = make_state manager timeout_heap' active' canceling messages store
   181       in state' end);
   182   check_thread_manager ())
   183 
   184 
   185 fun thread tool birth_time death_time desc f =
   186   (Runtime.thread true
   187        (fn () =>
   188            let
   189              val self = Thread.self ()
   190              val _ = register tool birth_time death_time desc self
   191            in unregister (f ()) self end);
   192    ())
   193 
   194 
   195 (** user commands **)
   196 
   197 (* kill threads *)
   198 
   199 fun kill_threads tool das_wort_worker = Synchronized.change global_state
   200   (fn {manager, timeout_heap, active, canceling, messages, store} =>
   201     let
   202       val killing =
   203         map_filter (fn (th, (tool', _, _, desc)) =>
   204                        if tool' = tool then SOME (th, (tool', Time.now (), desc))
   205                        else NONE) active
   206       val state' = make_state manager timeout_heap [] (killing @ canceling) messages store
   207       val _ =
   208         if null killing then ()
   209         else Output.urgent_message ("Interrupted active " ^ das_wort_worker ^ "s.")
   210     in state' end)
   211 
   212 
   213 (* running threads *)
   214 
   215 fun seconds time = string_of_int (Time.toSeconds time) ^ " s"
   216 
   217 fun has_running_threads tool =
   218   exists (fn (_, (tool', _, _, _)) => tool' = tool)
   219          (#active (Synchronized.value global_state))
   220 
   221 fun running_threads tool das_wort_worker =
   222   let
   223     val {active, canceling, ...} = Synchronized.value global_state
   224     val now = Time.now ()
   225     fun running_info (_, (tool', birth_time, death_time, desc)) =
   226       if tool' = tool then
   227         SOME ("Running: " ^ seconds (Time.- (now, birth_time)) ^ " -- " ^
   228               seconds (Time.- (death_time, now)) ^ " to live:\n" ^
   229               op ^ desc)
   230       else
   231         NONE
   232     fun canceling_info (_, (tool', death_time, desc)) =
   233       if tool' = tool then
   234         SOME ("Trying to interrupt " ^ das_wort_worker ^ " since " ^
   235               seconds (Time.- (now, death_time)) ^ ":\n" ^ op ^ desc)
   236       else
   237         NONE
   238     val running =
   239       case map_filter running_info active of
   240         [] => ["No " ^ das_wort_worker ^ "s running."]
   241       | ss => "Running " ^ das_wort_worker ^ "s " :: ss
   242     val interrupting =
   243       case map_filter canceling_info canceling of
   244         [] => []
   245       | ss => "Interrupting " ^ das_wort_worker ^ "s " :: ss
   246   in Output.urgent_message (space_implode "\n\n" (running @ interrupting)) end
   247 
   248 fun thread_messages tool das_wort_worker opt_limit =
   249   let
   250     val limit = the_default message_display_limit opt_limit
   251     val tool_store = Synchronized.value global_state
   252                      |> #store |> filter (curry (op =) tool o fst)
   253     val header =
   254       "Recent " ^ das_wort_worker ^ " messages" ^
   255         (if length tool_store <= limit then ":"
   256          else " (" ^ string_of_int limit ^ " displayed):")
   257     val ss = tool_store |> chop limit |> #1 |> map (op ^ o snd)
   258   in List.app Output.urgent_message (header :: maps break_into_chunks ss) end
   259 
   260 end;