src/Pure/Concurrent/future.ML
author wenzelm
Wed Sep 10 21:50:30 2008 +0200 (2008-09-10 ago)
changeset 28193 7ed74d0ba607
parent 28192 6d977729c8fa
child 28197 7053c539ecd8
permissions -rw-r--r--
replaced join_all by join_results, which returns Exn.results;
join: disallow Multithreading.self_critical, which is prone to deadlocks due to context change via fork;
wenzelm@28156
     1
(*  Title:      Pure/Concurrent/future.ML
wenzelm@28156
     2
    ID:         $Id$
wenzelm@28156
     3
    Author:     Makarius
wenzelm@28156
     4
wenzelm@28156
     5
Functional threads as future values.
wenzelm@28156
     6
*)
wenzelm@28156
     7
wenzelm@28156
     8
signature FUTURE =
wenzelm@28156
     9
sig
wenzelm@28166
    10
  type task = TaskQueue.task
wenzelm@28166
    11
  type group = TaskQueue.group
wenzelm@28156
    12
  type 'a T
wenzelm@28166
    13
  val task_of: 'a T -> task
wenzelm@28177
    14
  val group_of: 'a T -> group
wenzelm@28177
    15
  val shutdown_request: unit -> unit
wenzelm@28191
    16
  val future: group option -> task list -> (unit -> 'a) -> 'a T
wenzelm@28166
    17
  val fork: (unit -> 'a) -> 'a T
wenzelm@28186
    18
  val cancel: 'a T -> unit
wenzelm@28193
    19
  val join_results: 'a T list -> 'a Exn.result list
wenzelm@28166
    20
  val join: 'a T -> 'a
wenzelm@28156
    21
end;
wenzelm@28156
    22
wenzelm@28156
    23
structure Future: FUTURE =
wenzelm@28156
    24
struct
wenzelm@28156
    25
wenzelm@28177
    26
(** future values **)
wenzelm@28177
    27
wenzelm@28167
    28
(* identifiers *)
wenzelm@28167
    29
wenzelm@28167
    30
type task = TaskQueue.task;
wenzelm@28167
    31
type group = TaskQueue.group;
wenzelm@28167
    32
wenzelm@28177
    33
local val tag = Universal.tag () : (task * group) option Universal.tag in
wenzelm@28177
    34
  fun thread_data () = the_default NONE (Thread.getLocal tag);
wenzelm@28177
    35
  fun set_thread_data x = Thread.setLocal (tag, x);
wenzelm@28167
    36
end;
wenzelm@28167
    37
wenzelm@28167
    38
wenzelm@28167
    39
(* datatype future *)
wenzelm@28167
    40
wenzelm@28167
    41
datatype 'a T = Future of
wenzelm@28167
    42
 {task: task,
wenzelm@28177
    43
  group: group,
wenzelm@28167
    44
  result: 'a Exn.result option ref};
wenzelm@28167
    45
wenzelm@28167
    46
fun task_of (Future {task, ...}) = task;
wenzelm@28167
    47
fun group_of (Future {group, ...}) = group;
wenzelm@28167
    48
wenzelm@28167
    49
wenzelm@28177
    50
wenzelm@28177
    51
(** scheduling **)
wenzelm@28177
    52
wenzelm@28177
    53
(* global state *)
wenzelm@28177
    54
wenzelm@28177
    55
val queue = ref TaskQueue.empty;
wenzelm@28192
    56
val workers = ref ([]: (Thread.thread * bool) list);
wenzelm@28177
    57
val scheduler = ref (NONE: Thread.thread option);
wenzelm@28177
    58
wenzelm@28177
    59
val excessive = ref 0;
wenzelm@28177
    60
wenzelm@28177
    61
fun trace_active () =
wenzelm@28192
    62
  let
wenzelm@28192
    63
    val ws = ! workers;
wenzelm@28192
    64
    val m = string_of_int (length ws);
wenzelm@28192
    65
    val n = string_of_int (length (filter #2 ws));
wenzelm@28192
    66
  in Multithreading.tracing 1 (fn () => "SCHEDULE: " ^ m ^ " workers, " ^ n ^ " active") end;
wenzelm@28177
    67
wenzelm@28177
    68
wenzelm@28177
    69
(* requests *)
wenzelm@28177
    70
wenzelm@28177
    71
datatype request = Shutdown | Cancel of group;
wenzelm@28177
    72
val requests = Mailbox.create () : request Mailbox.T;
wenzelm@28177
    73
wenzelm@28177
    74
fun shutdown_request () = Mailbox.send requests Shutdown;
wenzelm@28177
    75
fun cancel_request group = Mailbox.send requests (Cancel group);
wenzelm@28177
    76
wenzelm@28177
    77
wenzelm@28177
    78
(* synchronization *)
wenzelm@28156
    79
wenzelm@28156
    80
local
wenzelm@28156
    81
  val lock = Mutex.mutex ();
wenzelm@28156
    82
  val cond = ConditionVar.conditionVar ();
wenzelm@28156
    83
in
wenzelm@28156
    84
wenzelm@28192
    85
fun SYNCHRONIZED name e = uninterruptible (fn restore_attributes => fn () =>
wenzelm@28162
    86
  let
wenzelm@28192
    87
    val _ = Multithreading.tracing 4 (fn () => name ^ ": locking");
wenzelm@28162
    88
    val _ = Mutex.lock lock;
wenzelm@28192
    89
    val _ = Multithreading.tracing 4 (fn () => name ^ ": locked");
wenzelm@28162
    90
    val result = Exn.capture (restore_attributes e) ();
wenzelm@28162
    91
    val _ = Mutex.unlock lock;
wenzelm@28192
    92
    val _ = Multithreading.tracing 4 (fn () => name ^ ": unlocked");
wenzelm@28162
    93
  in Exn.release result end) ();
wenzelm@28156
    94
wenzelm@28167
    95
fun wait name = (*requires SYNCHRONIZED*)
wenzelm@28167
    96
  let
wenzelm@28192
    97
    val _ = Multithreading.tracing 4 (fn () => name ^ ": waiting");
wenzelm@28167
    98
    val _ = ConditionVar.wait (cond, lock);
wenzelm@28192
    99
    val _ = Multithreading.tracing 4 (fn () => name ^ ": notified");
wenzelm@28167
   100
  in () end;
wenzelm@28166
   101
wenzelm@28166
   102
fun notify_all () = (*requires SYNCHRONIZED*)
wenzelm@28166
   103
  ConditionVar.broadcast cond;
wenzelm@28156
   104
wenzelm@28156
   105
end;
wenzelm@28156
   106
wenzelm@28156
   107
wenzelm@28177
   108
(* execute *)
wenzelm@28156
   109
wenzelm@28167
   110
fun execute name (task, group, run) =
wenzelm@28167
   111
  let
wenzelm@28177
   112
    val _ = set_thread_data (SOME (task, group));
wenzelm@28167
   113
    val _ = Multithreading.tracing 4 (fn () => name ^ ": running");
wenzelm@28167
   114
    val ok = run ();
wenzelm@28167
   115
    val _ = Multithreading.tracing 4 (fn () => name ^ ": finished");
wenzelm@28177
   116
    val _ = set_thread_data NONE;
wenzelm@28192
   117
    val _ = SYNCHRONIZED "execute" (fn () =>
wenzelm@28177
   118
     (change queue (TaskQueue.finish task);
wenzelm@28186
   119
      if ok then ()
wenzelm@28191
   120
      else if TaskQueue.cancel (! queue) group then ()
wenzelm@28186
   121
      else cancel_request group;
wenzelm@28177
   122
      notify_all ()));
wenzelm@28167
   123
  in () end;
wenzelm@28167
   124
wenzelm@28167
   125
wenzelm@28167
   126
(* worker threads *)
wenzelm@28167
   127
wenzelm@28192
   128
fun change_active active = (*requires SYNCHRONIZED*)
wenzelm@28192
   129
  (change workers (AList.update Thread.equal (Thread.self (), active)); trace_active ());
wenzelm@28186
   130
wenzelm@28186
   131
fun worker_wait name = (*requires SYNCHRONIZED*)
wenzelm@28186
   132
  (change_active false; wait name; change_active true);
wenzelm@28162
   133
wenzelm@28167
   134
fun worker_next name = (*requires SYNCHRONIZED*)
wenzelm@28167
   135
  if ! excessive > 0 then
wenzelm@28167
   136
    (dec excessive;
wenzelm@28192
   137
     change workers (filter_out (fn (thread, _) => Thread.equal (thread, Thread.self ())));
wenzelm@28167
   138
     NONE)
wenzelm@28166
   139
  else
wenzelm@28186
   140
    (case change_result queue TaskQueue.dequeue of
wenzelm@28186
   141
      NONE => (worker_wait name; worker_next name)
wenzelm@28166
   142
    | some => some);
wenzelm@28156
   143
wenzelm@28167
   144
fun worker_loop name =
wenzelm@28192
   145
  (case SYNCHRONIZED name (fn () => worker_next name) of
wenzelm@28166
   146
    NONE => ()
wenzelm@28167
   147
  | SOME work => (execute name work; worker_loop name));
wenzelm@28156
   148
wenzelm@28167
   149
fun worker_start name = (*requires SYNCHRONIZED*)
wenzelm@28192
   150
  change workers
wenzelm@28192
   151
    (cons (Thread.fork (fn () => worker_loop name, Multithreading.no_interrupts), true));
wenzelm@28156
   152
wenzelm@28156
   153
wenzelm@28156
   154
(* scheduler *)
wenzelm@28156
   155
wenzelm@28192
   156
fun scheduler_fork shutdown = SYNCHRONIZED "scheduler_fork" (fn () =>
wenzelm@28156
   157
  let
wenzelm@28177
   158
    val _ = trace_active ();
wenzelm@28191
   159
    val _ =
wenzelm@28192
   160
      (case List.partition (Thread.isActive o #1) (! workers) of
wenzelm@28191
   161
        (_, []) => ()
wenzelm@28191
   162
      | (active, inactive) =>
wenzelm@28191
   163
          (workers := active; Multithreading.tracing 0 (fn () =>
wenzelm@28192
   164
            "SCHEDULE: disposed " ^ string_of_int (length inactive) ^ " dead worker threads")));
wenzelm@28191
   165
wenzelm@28191
   166
    val m = if shutdown then 0 else Multithreading.max_threads_value ();
wenzelm@28167
   167
    val l = length (! workers);
wenzelm@28167
   168
    val _ = excessive := l - m;
wenzelm@28191
   169
    val _ = List.app (fn i => worker_start ("worker " ^ string_of_int i)) (l upto m - 1);
wenzelm@28192
   170
    val _ = if shutdown then notify_all () else ();
wenzelm@28192
   171
  in shutdown andalso null (! workers) end);
wenzelm@28167
   172
wenzelm@28191
   173
fun scheduler_loop (shutdown, canceled) =
wenzelm@28191
   174
  if scheduler_fork shutdown then ()
wenzelm@28191
   175
  else
wenzelm@28192
   176
    let
wenzelm@28192
   177
      val canceled' = SYNCHRONIZED "scheduler"
wenzelm@28192
   178
        (fn () => filter_out (TaskQueue.cancel (! queue)) canceled);
wenzelm@28192
   179
    in
wenzelm@28191
   180
      (case Mailbox.receive_timeout (Time.fromSeconds 1) requests of
wenzelm@28191
   181
        SOME Shutdown => scheduler_loop (true, canceled')
wenzelm@28191
   182
      | SOME (Cancel group) => scheduler_loop (shutdown, group :: canceled')
wenzelm@28191
   183
      | NONE => scheduler_loop (shutdown, canceled'))
wenzelm@28191
   184
    end;
wenzelm@28156
   185
wenzelm@28192
   186
fun scheduler_check () = SYNCHRONIZED "scheduler_check" (fn () =>
wenzelm@28167
   187
  if (case ! scheduler of NONE => false | SOME thread => Thread.isActive thread) then ()
wenzelm@28191
   188
  else scheduler :=
wenzelm@28191
   189
    SOME (Thread.fork (fn () => scheduler_loop (false, []), Multithreading.no_interrupts)));
wenzelm@28156
   190
wenzelm@28156
   191
wenzelm@28191
   192
(* future values: fork independent computation *)
wenzelm@28156
   193
wenzelm@28191
   194
fun future opt_group deps (e: unit -> 'a) =
wenzelm@28156
   195
  let
wenzelm@28191
   196
    val _ = scheduler_check ();
wenzelm@28177
   197
wenzelm@28191
   198
    val group = (case opt_group of SOME group => group | NONE => TaskQueue.new_group ());
wenzelm@28177
   199
wenzelm@28166
   200
    val result = ref (NONE: 'a Exn.result option);
wenzelm@28177
   201
    val run = Multithreading.with_attributes (Thread.getAttributes ())
wenzelm@28177
   202
      (fn _ => fn ok =>
wenzelm@28177
   203
        let val res = if ok then Exn.capture e () else Exn.Exn Interrupt
wenzelm@28177
   204
        in result := SOME res; is_some (Exn.get_result res) end);
wenzelm@28177
   205
wenzelm@28192
   206
    val task = SYNCHRONIZED "future" (fn () =>
wenzelm@28166
   207
      change_result queue (TaskQueue.enqueue group deps run) before notify_all ());
wenzelm@28166
   208
  in Future {task = task, group = group, result = result} end;
wenzelm@28162
   209
wenzelm@28191
   210
fun fork e = future (Option.map #2 (thread_data ())) [] e;
wenzelm@28186
   211
wenzelm@28186
   212
wenzelm@28191
   213
(* join: retrieve results *)
wenzelm@28186
   214
wenzelm@28193
   215
fun join_results xs =
wenzelm@28156
   216
  let
wenzelm@28193
   217
    val _ = Multithreading.self_critical () andalso
wenzelm@28193
   218
      error "Cannot join future values within critical section";
wenzelm@28191
   219
    val _ = scheduler_check ();
wenzelm@28177
   220
wenzelm@28186
   221
    fun unfinished () =
wenzelm@28186
   222
      xs |> map_filter (fn Future {task, result = ref NONE, ...} => SOME task | _ => NONE);
wenzelm@28186
   223
wenzelm@28186
   224
    (*alien thread -- refrain from contending for resources*)
wenzelm@28186
   225
    fun passive_join () = (*requires SYNCHRONIZED*)
wenzelm@28186
   226
      (case unfinished () of [] => ()
wenzelm@28186
   227
      | _ => (wait "join"; passive_join ()));
wenzelm@28186
   228
wenzelm@28186
   229
    (*proper worker thread -- actively work towards results*)
wenzelm@28186
   230
    fun active_join () = (*requires SYNCHRONIZED*)
wenzelm@28186
   231
      (case unfinished () of [] => ()
wenzelm@28186
   232
      | tasks =>
wenzelm@28186
   233
          (case change_result queue (TaskQueue.dequeue_towards tasks) of
wenzelm@28186
   234
            NONE => (worker_wait "join"; active_join ())
wenzelm@28186
   235
          | SOME work => (execute "join" work; active_join ())));
wenzelm@28186
   236
wenzelm@28186
   237
    val _ =
wenzelm@28186
   238
      (case thread_data () of
wenzelm@28192
   239
        NONE => SYNCHRONIZED "join" passive_join
wenzelm@28192
   240
      | SOME (task, _) => SYNCHRONIZED "join" (fn () =>
wenzelm@28186
   241
         (change queue (TaskQueue.depend (unfinished ()) task); active_join ())));
wenzelm@28186
   242
wenzelm@28193
   243
  in xs |> map (fn Future {result = ref (SOME res), ...} => res) end;
wenzelm@28186
   244
wenzelm@28193
   245
fun join x = Exn.release (singleton join_results x);
wenzelm@28156
   246
wenzelm@28191
   247
wenzelm@28191
   248
(* termination *)
wenzelm@28191
   249
wenzelm@28191
   250
(*cancel: present and future group members will be interrupted eventually*)
wenzelm@28191
   251
fun cancel x = (scheduler_check (); cancel_request (group_of x));
wenzelm@28191
   252
wenzelm@28191
   253
(*interrupt: adhoc signal, permissive, may get ignored*)
wenzelm@28192
   254
fun interrupt_task id = SYNCHRONIZED "interrupt" (fn () => TaskQueue.interrupt (! queue) id);
wenzelm@28191
   255
wenzelm@28156
   256
end;