src/Pure/Concurrent/future.ML
author wenzelm
Mon Sep 08 20:33:24 2008 +0200 (2008-09-08 ago)
changeset 28167 27e2ca41b58c
parent 28166 43087721a66e
child 28170 a18cf8a0e656
permissions -rw-r--r--
more interrupt operations;
maintain thread data (formerly in task_queue.ML);
wait: added tracing;
added requests mailbox;
handle excessive workers
fork: inherit group;
added rudimentary shutdown;
misc tuning;
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@28166
    14
  val group_of: 'a T -> group option
wenzelm@28167
    15
  val interrupt_task: task -> unit
wenzelm@28167
    16
  val interrupt_group: group -> unit
wenzelm@28167
    17
  val interrupt_task_group: task -> unit
wenzelm@28167
    18
  val interrupt: 'a T -> unit
wenzelm@28167
    19
  val shutdown: unit -> unit
wenzelm@28166
    20
  val future: group option -> task list -> (unit -> 'a) -> 'a T
wenzelm@28166
    21
  val fork: (unit -> 'a) -> 'a T
wenzelm@28166
    22
  val join: 'a T -> 'a
wenzelm@28156
    23
end;
wenzelm@28156
    24
wenzelm@28156
    25
structure Future: FUTURE =
wenzelm@28156
    26
struct
wenzelm@28156
    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@28167
    33
local val tag = Universal.tag () : task option Universal.tag in
wenzelm@28167
    34
  fun get_task () = the_default NONE (Thread.getLocal tag);
wenzelm@28167
    35
  fun set_task x = Thread.setLocal (tag, x);
wenzelm@28167
    36
end;
wenzelm@28167
    37
wenzelm@28167
    38
local val tag = Universal.tag () : group option Universal.tag in
wenzelm@28167
    39
  fun get_group () = the_default NONE (Thread.getLocal tag);
wenzelm@28167
    40
  fun set_group x = Thread.setLocal (tag, x);
wenzelm@28167
    41
end;
wenzelm@28167
    42
wenzelm@28167
    43
wenzelm@28167
    44
(* datatype future *)
wenzelm@28167
    45
wenzelm@28167
    46
datatype 'a T = Future of
wenzelm@28167
    47
 {task: task,
wenzelm@28167
    48
  group: group option,
wenzelm@28167
    49
  result: 'a Exn.result option ref};
wenzelm@28167
    50
wenzelm@28167
    51
fun task_of (Future {task, ...}) = task;
wenzelm@28167
    52
fun group_of (Future {group, ...}) = group;
wenzelm@28167
    53
wenzelm@28167
    54
wenzelm@28156
    55
(* synchronized execution *)
wenzelm@28156
    56
wenzelm@28156
    57
local
wenzelm@28156
    58
  val lock = Mutex.mutex ();
wenzelm@28156
    59
  val cond = ConditionVar.conditionVar ();
wenzelm@28156
    60
in
wenzelm@28156
    61
wenzelm@28162
    62
fun SYNCHRONIZED e = uninterruptible (fn restore_attributes => fn () =>
wenzelm@28162
    63
  let
wenzelm@28162
    64
    val _ = Mutex.lock lock;
wenzelm@28162
    65
    val result = Exn.capture (restore_attributes e) ();
wenzelm@28162
    66
    val _ = Mutex.unlock lock;
wenzelm@28162
    67
  in Exn.release result end) ();
wenzelm@28156
    68
wenzelm@28167
    69
fun wait name = (*requires SYNCHRONIZED*)
wenzelm@28167
    70
  let
wenzelm@28167
    71
    val _ = Multithreading.tracing 4 (fn () => name ^ " : waiting");
wenzelm@28167
    72
    val _ = ConditionVar.wait (cond, lock);
wenzelm@28167
    73
    val _ = Multithreading.tracing 4 (fn () => name ^ " : notified");
wenzelm@28167
    74
  in () end;
wenzelm@28166
    75
wenzelm@28166
    76
fun notify_all () = (*requires SYNCHRONIZED*)
wenzelm@28166
    77
  ConditionVar.broadcast cond;
wenzelm@28156
    78
wenzelm@28156
    79
end;
wenzelm@28156
    80
wenzelm@28156
    81
wenzelm@28167
    82
(** scheduling **)
wenzelm@28156
    83
wenzelm@28167
    84
datatype request = Shutdown | CancelGroup of group;
wenzelm@28167
    85
val requests = Mailbox.create () : request Mailbox.T;
wenzelm@28156
    86
wenzelm@28166
    87
val queue = ref TaskQueue.empty;
wenzelm@28156
    88
val scheduler = ref (NONE: Thread.thread option);
wenzelm@28156
    89
val workers = ref ([]: Thread.thread list);
wenzelm@28156
    90
wenzelm@28156
    91
wenzelm@28167
    92
(* signals *)
wenzelm@28167
    93
wenzelm@28167
    94
fun interrupt_task x = SYNCHRONIZED (fn () => TaskQueue.interrupt_task (! queue) x);
wenzelm@28167
    95
fun interrupt_group x = SYNCHRONIZED (fn () => TaskQueue.interrupt_group (! queue) x);
wenzelm@28167
    96
fun interrupt_task_group x = SYNCHRONIZED (fn () => TaskQueue.interrupt_task_group (! queue) x);
wenzelm@28167
    97
wenzelm@28167
    98
fun interrupt (Future {task, ...}) = interrupt_task_group task;
wenzelm@28167
    99
wenzelm@28167
   100
fun shutdown () = Mailbox.send Shutdown requests;
wenzelm@28167
   101
wenzelm@28167
   102
wenzelm@28167
   103
(* execute *)
wenzelm@28156
   104
wenzelm@28167
   105
fun execute name (task, group, run) =
wenzelm@28167
   106
  let
wenzelm@28167
   107
    val _ = set_task (SOME task);
wenzelm@28167
   108
    val _ = set_group group;
wenzelm@28167
   109
    val _ = Multithreading.tracing 4 (fn () => name ^ ": running");
wenzelm@28167
   110
    val ok = run ();
wenzelm@28167
   111
    val _ = Multithreading.tracing 4 (fn () => name ^ ": finished");
wenzelm@28167
   112
    val _ = set_task NONE;
wenzelm@28167
   113
    val _ = set_group NONE;
wenzelm@28167
   114
    val _ = SYNCHRONIZED (fn () => (change queue (TaskQueue.finished task); notify_all ()));
wenzelm@28167
   115
    val _ = (case (ok, group) of (false, SOME g) => Mailbox.send (CancelGroup g) requests | _ => ());
wenzelm@28167
   116
  in () end;
wenzelm@28167
   117
wenzelm@28167
   118
wenzelm@28167
   119
(* worker threads *)
wenzelm@28167
   120
wenzelm@28167
   121
val excessive = ref 0;
wenzelm@28167
   122
val active = ref 0;
wenzelm@28162
   123
wenzelm@28162
   124
fun change_active b = (*requires SYNCHRONIZED*)
wenzelm@28166
   125
 (change active (fn n => if b then n + 1 else n - 1);
wenzelm@28166
   126
  Multithreading.tracing 1 (fn () => "SCHEDULE: " ^ string_of_int (! active) ^ " active"));
wenzelm@28162
   127
wenzelm@28167
   128
fun worker_next name = (*requires SYNCHRONIZED*)
wenzelm@28167
   129
  if ! excessive > 0 then
wenzelm@28167
   130
    (dec excessive;
wenzelm@28167
   131
     change_active false;
wenzelm@28167
   132
     change workers (remove Thread.equal (Thread.self ()));
wenzelm@28167
   133
     NONE)
wenzelm@28166
   134
  else
wenzelm@28166
   135
    (case change_result queue (TaskQueue.dequeue (Thread.self ())) of
wenzelm@28167
   136
      NONE => (change_active false; wait name; change_active true; worker_next name)
wenzelm@28166
   137
    | some => some);
wenzelm@28156
   138
wenzelm@28167
   139
fun worker_loop name =
wenzelm@28167
   140
  (case SYNCHRONIZED (fn () => worker_next name) of
wenzelm@28166
   141
    NONE => ()
wenzelm@28167
   142
  | SOME work => (execute name work; worker_loop name));
wenzelm@28156
   143
wenzelm@28167
   144
fun worker_start name = (*requires SYNCHRONIZED*)
wenzelm@28156
   145
 (change_active true;
wenzelm@28167
   146
  change workers (cons (Thread.fork (fn () => worker_loop name, Multithreading.no_interrupts))));
wenzelm@28156
   147
wenzelm@28156
   148
wenzelm@28156
   149
(* scheduler *)
wenzelm@28156
   150
wenzelm@28167
   151
fun scheduler_fork () = SYNCHRONIZED (fn () =>
wenzelm@28156
   152
  let
wenzelm@28156
   153
    val m = Multithreading.max_threads_value ();
wenzelm@28167
   154
    val l = length (! workers);
wenzelm@28167
   155
    val _ = excessive := l - m;
wenzelm@28167
   156
  in List.app (fn i => worker_start ("worker " ^ string_of_int i)) (l upto m - 1) end);
wenzelm@28167
   157
wenzelm@28167
   158
fun scheduler_loop () =
wenzelm@28167
   159
  (scheduler_fork ();
wenzelm@28167
   160
    (case Mailbox.receive_timeout (Time.fromMilliseconds 300) requests of
wenzelm@28167
   161
      SOME Shutdown => ()   (* FIXME *)
wenzelm@28167
   162
    | SOME (CancelGroup group) => (interrupt_group group; scheduler_loop ())  (* FIXME *)
wenzelm@28167
   163
    | NONE => scheduler_loop ()));
wenzelm@28156
   164
wenzelm@28156
   165
fun check_scheduler () = SYNCHRONIZED (fn () =>
wenzelm@28167
   166
  if (case ! scheduler of NONE => false | SOME thread => Thread.isActive thread) then ()
wenzelm@28167
   167
  else scheduler := SOME (Thread.fork (scheduler_loop, Multithreading.no_interrupts)));
wenzelm@28156
   168
wenzelm@28156
   169
wenzelm@28156
   170
(* future values *)
wenzelm@28156
   171
wenzelm@28166
   172
fun future group deps (e: unit -> 'a) =
wenzelm@28156
   173
  let
wenzelm@28156
   174
    val _ = check_scheduler ();
wenzelm@28166
   175
    val result = ref (NONE: 'a Exn.result option);
wenzelm@28167
   176
    val run = Multithreading.with_attributes (Thread.getAttributes ()) (fn _ => fn () =>
wenzelm@28167
   177
      let val res = Exn.capture e () in result := SOME res; is_some (Exn.get_result res) end);
wenzelm@28166
   178
    val task = SYNCHRONIZED (fn () =>
wenzelm@28166
   179
      change_result queue (TaskQueue.enqueue group deps run) before notify_all ());
wenzelm@28166
   180
  in Future {task = task, group = group, result = result} end;
wenzelm@28162
   181
wenzelm@28167
   182
fun fork e = future (get_group ()) [] e;
wenzelm@28162
   183
wenzelm@28166
   184
fun join (Future {result, ...}) =
wenzelm@28156
   185
  let
wenzelm@28156
   186
    val _ = check_scheduler ();
wenzelm@28166
   187
    fun loop () =
wenzelm@28166
   188
      (case ! result of
wenzelm@28167
   189
        NONE => (wait "join"; loop ())
wenzelm@28166
   190
      | SOME res => res);
wenzelm@28166
   191
  in Exn.release (SYNCHRONIZED loop) end;
wenzelm@28156
   192
wenzelm@28156
   193
end;