src/Pure/Concurrent/future.ML
author wenzelm
Mon Sep 08 00:10:41 2008 +0200 (2008-09-08 ago)
changeset 28162 55772e4e95e0
parent 28156 5205f7979b4f
child 28163 8bf8c21296ca
permissions -rw-r--r--
tuned check_cache;
removed broken self_synchronized, which cannot be used in conjunction with condition variables;
more precise use of SYNCHRONIZED vs. wait;
tuned worker_loop;
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@28156
    10
  type 'a T
wenzelm@28156
    11
  eqtype id
wenzelm@28156
    12
  val id_of: 'a T -> id
wenzelm@28156
    13
  val interrupt: id -> unit
wenzelm@28156
    14
  val dependent_future: id list -> (unit -> 'a) -> 'a T
wenzelm@28156
    15
  val future: (unit -> 'a) -> 'a T
wenzelm@28156
    16
  val await: 'a T -> 'a
wenzelm@28156
    17
end;
wenzelm@28156
    18
wenzelm@28156
    19
structure Future: FUTURE =
wenzelm@28156
    20
struct
wenzelm@28156
    21
wenzelm@28156
    22
(* synchronized execution *)
wenzelm@28156
    23
wenzelm@28156
    24
local
wenzelm@28156
    25
  val lock = Mutex.mutex ();
wenzelm@28156
    26
  val cond = ConditionVar.conditionVar ();
wenzelm@28156
    27
in
wenzelm@28156
    28
wenzelm@28162
    29
fun SYNCHRONIZED e = uninterruptible (fn restore_attributes => fn () =>
wenzelm@28162
    30
  let
wenzelm@28162
    31
    val _ = Mutex.lock lock;
wenzelm@28162
    32
    val result = Exn.capture (restore_attributes e) ();
wenzelm@28162
    33
    val _ = Mutex.unlock lock;
wenzelm@28162
    34
  in Exn.release result end) ();
wenzelm@28156
    35
wenzelm@28156
    36
fun wait () = ConditionVar.wait (cond, lock);
wenzelm@28156
    37
fun wait_timeout timeout = ConditionVar.waitUntil (cond, lock, Time.+ (Time.now (), timeout));
wenzelm@28156
    38
wenzelm@28156
    39
fun notify_all () = ConditionVar.broadcast cond;
wenzelm@28156
    40
wenzelm@28156
    41
end;
wenzelm@28156
    42
wenzelm@28156
    43
wenzelm@28156
    44
(* typed futures, unytped ids *)
wenzelm@28156
    45
wenzelm@28156
    46
datatype 'a T = Future of serial * 'a Exn.result option ref;
wenzelm@28156
    47
wenzelm@28156
    48
datatype id = Id of serial;
wenzelm@28156
    49
fun id_of (Future (id, _)) = Id id;
wenzelm@28156
    50
wenzelm@28156
    51
local val tag = Universal.tag () : serial Universal.tag in
wenzelm@28156
    52
  fun get_id () = Thread.getLocal tag;
wenzelm@28156
    53
  fun put_id id = Thread.setLocal (tag, id);
wenzelm@28156
    54
end;
wenzelm@28156
    55
wenzelm@28156
    56
wenzelm@28156
    57
(* ordered queue of tasks *)
wenzelm@28156
    58
wenzelm@28156
    59
datatype task =
wenzelm@28156
    60
  Task of (unit -> unit) |
wenzelm@28156
    61
  Running of Thread.thread;
wenzelm@28156
    62
wenzelm@28156
    63
datatype queue = Queue of task IntGraph.T * (serial * (unit -> unit)) Queue.T;
wenzelm@28156
    64
wenzelm@28156
    65
val empty_queue = Queue (IntGraph.empty, Queue.empty);
wenzelm@28156
    66
wenzelm@28156
    67
fun check_cache (queue as Queue (tasks, cache)) =
wenzelm@28156
    68
  if not (Queue.is_empty cache) then queue
wenzelm@28156
    69
  else
wenzelm@28162
    70
    let fun ready (id, (Task task, ([], _))) = Queue.enqueue (id, task) | ready _ = I
wenzelm@28162
    71
    in Queue (tasks, IntGraph.fold ready tasks Queue.empty) end;
wenzelm@28156
    72
wenzelm@28156
    73
val next_task = check_cache #> (fn queue as Queue (tasks, cache) =>
wenzelm@28156
    74
  if Queue.is_empty cache then (NONE, queue)
wenzelm@28156
    75
  else
wenzelm@28156
    76
    let val (task, cache') = Queue.dequeue cache
wenzelm@28156
    77
    in (SOME task, Queue (tasks, cache')) end);
wenzelm@28156
    78
wenzelm@28156
    79
fun get_task (Queue (tasks, _)) id = IntGraph.get_node tasks id;
wenzelm@28156
    80
wenzelm@28156
    81
fun new_task deps id task (Queue (tasks, _)) =
wenzelm@28156
    82
  let
wenzelm@28156
    83
    fun add_dep (Id dep) G = IntGraph.add_edge_acyclic (dep, id) G
wenzelm@28156
    84
      handle IntGraph.UNDEF _ => G;  (*dep already finished*)
wenzelm@28156
    85
    val tasks' = tasks |> IntGraph.new_node (id, Task task) |> fold add_dep deps;
wenzelm@28156
    86
  in Queue (tasks', Queue.empty) end;
wenzelm@28156
    87
wenzelm@28156
    88
fun running_task id thread (Queue (tasks, cache)) =
wenzelm@28156
    89
  Queue (IntGraph.map_node id (K (Running thread)) tasks, cache);
wenzelm@28156
    90
wenzelm@28156
    91
fun finished_task id (Queue (tasks, _)) =
wenzelm@28156
    92
  Queue (IntGraph.del_nodes [id] tasks, Queue.empty);
wenzelm@28156
    93
wenzelm@28156
    94
wenzelm@28156
    95
(* global state *)
wenzelm@28156
    96
wenzelm@28156
    97
val tasks = ref empty_queue;
wenzelm@28156
    98
val scheduler = ref (NONE: Thread.thread option);
wenzelm@28156
    99
val workers = ref ([]: Thread.thread list);
wenzelm@28156
   100
wenzelm@28156
   101
wenzelm@28156
   102
fun interrupt (Id id) = SYNCHRONIZED (fn () =>
wenzelm@28156
   103
  (case try (get_task (! tasks)) id of
wenzelm@28156
   104
    SOME (Running thread) => Thread.interrupt thread
wenzelm@28156
   105
  | _ => ()));
wenzelm@28156
   106
wenzelm@28156
   107
wenzelm@28156
   108
(* worker thread *)
wenzelm@28156
   109
wenzelm@28162
   110
local val active = ref 0 in
wenzelm@28162
   111
wenzelm@28162
   112
fun change_active b = (*requires SYNCHRONIZED*)
wenzelm@28162
   113
  let
wenzelm@28162
   114
    val _ = change active (fn n => if b then n + 1 else n - 1);
wenzelm@28162
   115
    val n = ! active;
wenzelm@28162
   116
    val _ = Multithreading.tracing 1 (fn () => "SCHEDULE: " ^ string_of_int n ^ " active");
wenzelm@28162
   117
  in () end;
wenzelm@28162
   118
wenzelm@28162
   119
end;
wenzelm@28162
   120
wenzelm@28156
   121
fun excessive_threads () = false;  (* FIXME *)
wenzelm@28156
   122
wenzelm@28162
   123
fun worker_stop () = SYNCHRONIZED (fn () =>
wenzelm@28162
   124
  (change_active false; change workers (filter (fn t => not (Thread.equal (t, Thread.self ()))))));
wenzelm@28156
   125
wenzelm@28162
   126
fun worker_wait () = SYNCHRONIZED (fn () =>
wenzelm@28162
   127
  (change_active false; wait (); change_active true));
wenzelm@28156
   128
wenzelm@28156
   129
fun worker_loop () =
wenzelm@28162
   130
  if excessive_threads () then worker_stop ()
wenzelm@28162
   131
  else
wenzelm@28162
   132
    (case SYNCHRONIZED (fn () => change_result tasks next_task) of
wenzelm@28162
   133
      NONE => (worker_wait (); worker_loop ())
wenzelm@28162
   134
    | SOME (id, task) =>
wenzelm@28162
   135
        let
wenzelm@28162
   136
          val _ = SYNCHRONIZED (fn () => change tasks (running_task id (Thread.self ())));
wenzelm@28162
   137
          val _ = task ();
wenzelm@28162
   138
          val _ = SYNCHRONIZED (fn () => change tasks (finished_task id));
wenzelm@28162
   139
          val _ = notify_all ();
wenzelm@28162
   140
        in worker_loop () end);
wenzelm@28156
   141
wenzelm@28162
   142
fun worker_start () = SYNCHRONIZED (fn () =>
wenzelm@28156
   143
 (change_active true;
wenzelm@28162
   144
  change workers (cons (Thread.fork (worker_loop, Multithreading.no_interrupts)))));
wenzelm@28156
   145
wenzelm@28156
   146
wenzelm@28156
   147
(* scheduler *)
wenzelm@28156
   148
wenzelm@28162
   149
fun scheduler_loop () = (*requires SYNCHRONIZED*)
wenzelm@28156
   150
  let
wenzelm@28156
   151
    val m = Multithreading.max_threads_value ();
wenzelm@28156
   152
    val k = m - length (! workers);
wenzelm@28156
   153
    val _ = if k > 0 then funpow k worker_start () else ();
wenzelm@28156
   154
  in wait_timeout (Time.fromSeconds 1); scheduler_loop () end;
wenzelm@28156
   155
wenzelm@28156
   156
fun check_scheduler () = SYNCHRONIZED (fn () =>
wenzelm@28156
   157
  let
wenzelm@28156
   158
    val scheduler_active =
wenzelm@28156
   159
      (case ! scheduler of
wenzelm@28156
   160
        NONE => false
wenzelm@28156
   161
      | SOME t => Thread.isActive t);
wenzelm@28156
   162
  in
wenzelm@28156
   163
    if scheduler_active then ()
wenzelm@28162
   164
    else scheduler :=
wenzelm@28162
   165
      SOME (Thread.fork (SYNCHRONIZED o scheduler_loop, Multithreading.no_interrupts))
wenzelm@28156
   166
  end);
wenzelm@28156
   167
wenzelm@28156
   168
wenzelm@28156
   169
(* future values *)
wenzelm@28156
   170
wenzelm@28156
   171
fun dependent_future deps (e: unit -> 'a) =
wenzelm@28156
   172
  let
wenzelm@28156
   173
    val _ = check_scheduler ();
wenzelm@28156
   174
wenzelm@28156
   175
    val r = ref (NONE: 'a Exn.result option);
wenzelm@28156
   176
    val task = Multithreading.with_attributes (Thread.getAttributes ())
wenzelm@28156
   177
      (fn _ => fn () => r := SOME (Exn.capture e ()));
wenzelm@28162
   178
wenzelm@28156
   179
    val id = serial ();
wenzelm@28156
   180
    val _ = SYNCHRONIZED (fn () => change tasks (new_task deps id task));
wenzelm@28156
   181
    val _ = notify_all ();
wenzelm@28162
   182
wenzelm@28156
   183
  in Future (id, r) end;
wenzelm@28156
   184
wenzelm@28156
   185
fun future e = dependent_future [] e;
wenzelm@28156
   186
wenzelm@28156
   187
fun await (Future (_, r)) =
wenzelm@28156
   188
  let
wenzelm@28156
   189
    val _ = check_scheduler ();
wenzelm@28156
   190
wenzelm@28156
   191
    fun loop () =
wenzelm@28156
   192
      (case SYNCHRONIZED (fn () => ! r) of
wenzelm@28156
   193
        NONE => (wait (); loop ())
wenzelm@28156
   194
      | SOME res => Exn.release res);
wenzelm@28156
   195
  in loop () end;
wenzelm@28156
   196
wenzelm@28156
   197
end;