src/Pure/Concurrent/future.ML
author wenzelm
Tue, 06 Jan 2009 13:46:48 +0100
changeset 29366 1ffc8cbf39ec
parent 29341 6bb007a0f9f2
child 29384 a3c7e9ae9b71
permissions -rw-r--r--
tuned map: reduced overhead due to bulk jobs; tuned join_results: reduced overhead for finished futures; tuned;
Ignore whitespace changes - Everywhere: Within whitespace: At end of lines:
28156
5205f7979b4f Functional threads as future values.
wenzelm
parents:
diff changeset
     1
(*  Title:      Pure/Concurrent/future.ML
5205f7979b4f Functional threads as future values.
wenzelm
parents:
diff changeset
     2
    Author:     Makarius
5205f7979b4f Functional threads as future values.
wenzelm
parents:
diff changeset
     3
28201
7ae5cdb7b122 some general notes on future values;
wenzelm
parents: 28197
diff changeset
     4
Future values.
7ae5cdb7b122 some general notes on future values;
wenzelm
parents: 28197
diff changeset
     5
7ae5cdb7b122 some general notes on future values;
wenzelm
parents: 28197
diff changeset
     6
Notes:
7ae5cdb7b122 some general notes on future values;
wenzelm
parents: 28197
diff changeset
     7
7ae5cdb7b122 some general notes on future values;
wenzelm
parents: 28197
diff changeset
     8
  * Futures are similar to delayed evaluation, i.e. delay/force is
7ae5cdb7b122 some general notes on future values;
wenzelm
parents: 28197
diff changeset
     9
    generalized to fork/join (and variants).  The idea is to model
7ae5cdb7b122 some general notes on future values;
wenzelm
parents: 28197
diff changeset
    10
    parallel value-oriented computations, but *not* communicating
7ae5cdb7b122 some general notes on future values;
wenzelm
parents: 28197
diff changeset
    11
    processes.
7ae5cdb7b122 some general notes on future values;
wenzelm
parents: 28197
diff changeset
    12
7ae5cdb7b122 some general notes on future values;
wenzelm
parents: 28197
diff changeset
    13
  * Futures are grouped; failure of one group member causes the whole
7ae5cdb7b122 some general notes on future values;
wenzelm
parents: 28197
diff changeset
    14
    group to be interrupted eventually.
7ae5cdb7b122 some general notes on future values;
wenzelm
parents: 28197
diff changeset
    15
7ae5cdb7b122 some general notes on future values;
wenzelm
parents: 28197
diff changeset
    16
  * Forked futures are evaluated spontaneously by a farm of worker
7ae5cdb7b122 some general notes on future values;
wenzelm
parents: 28197
diff changeset
    17
    threads in the background; join resynchronizes the computation and
7ae5cdb7b122 some general notes on future values;
wenzelm
parents: 28197
diff changeset
    18
    delivers results (values or exceptions).
7ae5cdb7b122 some general notes on future values;
wenzelm
parents: 28197
diff changeset
    19
7ae5cdb7b122 some general notes on future values;
wenzelm
parents: 28197
diff changeset
    20
  * The pool of worker threads is limited, usually in correlation with
7ae5cdb7b122 some general notes on future values;
wenzelm
parents: 28197
diff changeset
    21
    the number of physical cores on the machine.  Note that allocation
7ae5cdb7b122 some general notes on future values;
wenzelm
parents: 28197
diff changeset
    22
    of runtime resources is distorted either if workers yield CPU time
7ae5cdb7b122 some general notes on future values;
wenzelm
parents: 28197
diff changeset
    23
    (e.g. via system sleep or wait operations), or if non-worker
7ae5cdb7b122 some general notes on future values;
wenzelm
parents: 28197
diff changeset
    24
    threads contend for significant runtime resources independently.
28156
5205f7979b4f Functional threads as future values.
wenzelm
parents:
diff changeset
    25
*)
5205f7979b4f Functional threads as future values.
wenzelm
parents:
diff changeset
    26
5205f7979b4f Functional threads as future values.
wenzelm
parents:
diff changeset
    27
signature FUTURE =
5205f7979b4f Functional threads as future values.
wenzelm
parents:
diff changeset
    28
sig
28645
605a3b1ef6ba added Future.enabled check;
wenzelm
parents: 28575
diff changeset
    29
  val enabled: unit -> bool
29119
99941fd0cb0e renamed structure TaskQueue to Task_Queue;
wenzelm
parents: 29118
diff changeset
    30
  type task = Task_Queue.task
99941fd0cb0e renamed structure TaskQueue to Task_Queue;
wenzelm
parents: 29118
diff changeset
    31
  type group = Task_Queue.group
28979
3ce619d8d432 fork/map: no inheritance of group (structure is nested, not parallel);
wenzelm
parents: 28972
diff changeset
    32
  val thread_data: unit -> (string * task) option
28972
cb8a2c3e188f renamed type Future.T to future;
wenzelm
parents: 28647
diff changeset
    33
  type 'a future
cb8a2c3e188f renamed type Future.T to future;
wenzelm
parents: 28647
diff changeset
    34
  val task_of: 'a future -> task
cb8a2c3e188f renamed type Future.T to future;
wenzelm
parents: 28647
diff changeset
    35
  val group_of: 'a future -> group
cb8a2c3e188f renamed type Future.T to future;
wenzelm
parents: 28647
diff changeset
    36
  val peek: 'a future -> 'a Exn.result option
cb8a2c3e188f renamed type Future.T to future;
wenzelm
parents: 28647
diff changeset
    37
  val is_finished: 'a future -> bool
28997
1b99dcae2156 added constant value;
wenzelm
parents: 28979
diff changeset
    38
  val value: 'a -> 'a future
28972
cb8a2c3e188f renamed type Future.T to future;
wenzelm
parents: 28647
diff changeset
    39
  val fork: (unit -> 'a) -> 'a future
28979
3ce619d8d432 fork/map: no inheritance of group (structure is nested, not parallel);
wenzelm
parents: 28972
diff changeset
    40
  val fork_group: group -> (unit -> 'a) -> 'a future
3ce619d8d432 fork/map: no inheritance of group (structure is nested, not parallel);
wenzelm
parents: 28972
diff changeset
    41
  val fork_deps: 'b future list -> (unit -> 'a) -> 'a future
29119
99941fd0cb0e renamed structure TaskQueue to Task_Queue;
wenzelm
parents: 29118
diff changeset
    42
  val fork_pri: int -> (unit -> 'a) -> 'a future
28972
cb8a2c3e188f renamed type Future.T to future;
wenzelm
parents: 28647
diff changeset
    43
  val join_results: 'a future list -> 'a Exn.result list
cb8a2c3e188f renamed type Future.T to future;
wenzelm
parents: 28647
diff changeset
    44
  val join_result: 'a future -> 'a Exn.result
cb8a2c3e188f renamed type Future.T to future;
wenzelm
parents: 28647
diff changeset
    45
  val join: 'a future -> 'a
cb8a2c3e188f renamed type Future.T to future;
wenzelm
parents: 28647
diff changeset
    46
  val map: ('a -> 'b) -> 'a future -> 'b future
28206
bcd48c6897d4 eliminated requests, use global state variables uniformly;
wenzelm
parents: 28203
diff changeset
    47
  val interrupt_task: string -> unit
28972
cb8a2c3e188f renamed type Future.T to future;
wenzelm
parents: 28647
diff changeset
    48
  val cancel: 'a future -> unit
28203
88f18387f1c9 shutdown: global join-and-shutdown operation;
wenzelm
parents: 28202
diff changeset
    49
  val shutdown: unit -> unit
28156
5205f7979b4f Functional threads as future values.
wenzelm
parents:
diff changeset
    50
end;
5205f7979b4f Functional threads as future values.
wenzelm
parents:
diff changeset
    51
5205f7979b4f Functional threads as future values.
wenzelm
parents:
diff changeset
    52
structure Future: FUTURE =
5205f7979b4f Functional threads as future values.
wenzelm
parents:
diff changeset
    53
struct
5205f7979b4f Functional threads as future values.
wenzelm
parents:
diff changeset
    54
28177
8c0335bc9336 inherit group from running thread, or create a new one -- make it harder to re-use canceled groups;
wenzelm
parents: 28170
diff changeset
    55
(** future values **)
8c0335bc9336 inherit group from running thread, or create a new one -- make it harder to re-use canceled groups;
wenzelm
parents: 28170
diff changeset
    56
28645
605a3b1ef6ba added Future.enabled check;
wenzelm
parents: 28575
diff changeset
    57
fun enabled () =
29118
8f2481aa363d removed old scheduler;
wenzelm
parents: 28997
diff changeset
    58
  Multithreading.enabled () andalso
28645
605a3b1ef6ba added Future.enabled check;
wenzelm
parents: 28575
diff changeset
    59
    not (Multithreading.self_critical ());
605a3b1ef6ba added Future.enabled check;
wenzelm
parents: 28575
diff changeset
    60
605a3b1ef6ba added Future.enabled check;
wenzelm
parents: 28575
diff changeset
    61
28167
27e2ca41b58c more interrupt operations;
wenzelm
parents: 28166
diff changeset
    62
(* identifiers *)
27e2ca41b58c more interrupt operations;
wenzelm
parents: 28166
diff changeset
    63
29119
99941fd0cb0e renamed structure TaskQueue to Task_Queue;
wenzelm
parents: 29118
diff changeset
    64
type task = Task_Queue.task;
99941fd0cb0e renamed structure TaskQueue to Task_Queue;
wenzelm
parents: 29118
diff changeset
    65
type group = Task_Queue.group;
28167
27e2ca41b58c more interrupt operations;
wenzelm
parents: 28166
diff changeset
    66
28979
3ce619d8d432 fork/map: no inheritance of group (structure is nested, not parallel);
wenzelm
parents: 28972
diff changeset
    67
local val tag = Universal.tag () : (string * task) option Universal.tag in
28177
8c0335bc9336 inherit group from running thread, or create a new one -- make it harder to re-use canceled groups;
wenzelm
parents: 28170
diff changeset
    68
  fun thread_data () = the_default NONE (Thread.getLocal tag);
28390
0b9fb63b8e1d proper setmp_thread_data for nested execute (cf. join_loop);
wenzelm
parents: 28386
diff changeset
    69
  fun setmp_thread_data data f x = Library.setmp_thread_data tag (thread_data ()) (SOME data) f x;
28167
27e2ca41b58c more interrupt operations;
wenzelm
parents: 28166
diff changeset
    70
end;
27e2ca41b58c more interrupt operations;
wenzelm
parents: 28166
diff changeset
    71
27e2ca41b58c more interrupt operations;
wenzelm
parents: 28166
diff changeset
    72
27e2ca41b58c more interrupt operations;
wenzelm
parents: 28166
diff changeset
    73
(* datatype future *)
27e2ca41b58c more interrupt operations;
wenzelm
parents: 28166
diff changeset
    74
28972
cb8a2c3e188f renamed type Future.T to future;
wenzelm
parents: 28647
diff changeset
    75
datatype 'a future = Future of
28167
27e2ca41b58c more interrupt operations;
wenzelm
parents: 28166
diff changeset
    76
 {task: task,
28177
8c0335bc9336 inherit group from running thread, or create a new one -- make it harder to re-use canceled groups;
wenzelm
parents: 28170
diff changeset
    77
  group: group,
28167
27e2ca41b58c more interrupt operations;
wenzelm
parents: 28166
diff changeset
    78
  result: 'a Exn.result option ref};
27e2ca41b58c more interrupt operations;
wenzelm
parents: 28166
diff changeset
    79
27e2ca41b58c more interrupt operations;
wenzelm
parents: 28166
diff changeset
    80
fun task_of (Future {task, ...}) = task;
27e2ca41b58c more interrupt operations;
wenzelm
parents: 28166
diff changeset
    81
fun group_of (Future {group, ...}) = group;
27e2ca41b58c more interrupt operations;
wenzelm
parents: 28166
diff changeset
    82
28558
2a6297b4273c replaced str_of by general peek;
wenzelm
parents: 28548
diff changeset
    83
fun peek (Future {result, ...}) = ! result;
2a6297b4273c replaced str_of by general peek;
wenzelm
parents: 28548
diff changeset
    84
fun is_finished x = is_some (peek x);
28320
c6aef67f964d added is_finished;
wenzelm
parents: 28304
diff changeset
    85
28997
1b99dcae2156 added constant value;
wenzelm
parents: 28979
diff changeset
    86
fun value x = Future
29119
99941fd0cb0e renamed structure TaskQueue to Task_Queue;
wenzelm
parents: 29118
diff changeset
    87
 {task = Task_Queue.new_task 0,
99941fd0cb0e renamed structure TaskQueue to Task_Queue;
wenzelm
parents: 29118
diff changeset
    88
  group = Task_Queue.new_group (),
28997
1b99dcae2156 added constant value;
wenzelm
parents: 28979
diff changeset
    89
  result = ref (SOME (Exn.Result x))};
1b99dcae2156 added constant value;
wenzelm
parents: 28979
diff changeset
    90
28167
27e2ca41b58c more interrupt operations;
wenzelm
parents: 28166
diff changeset
    91
28177
8c0335bc9336 inherit group from running thread, or create a new one -- make it harder to re-use canceled groups;
wenzelm
parents: 28170
diff changeset
    92
8c0335bc9336 inherit group from running thread, or create a new one -- make it harder to re-use canceled groups;
wenzelm
parents: 28170
diff changeset
    93
(** scheduling **)
8c0335bc9336 inherit group from running thread, or create a new one -- make it harder to re-use canceled groups;
wenzelm
parents: 28170
diff changeset
    94
8c0335bc9336 inherit group from running thread, or create a new one -- make it harder to re-use canceled groups;
wenzelm
parents: 28170
diff changeset
    95
(* global state *)
8c0335bc9336 inherit group from running thread, or create a new one -- make it harder to re-use canceled groups;
wenzelm
parents: 28170
diff changeset
    96
29119
99941fd0cb0e renamed structure TaskQueue to Task_Queue;
wenzelm
parents: 29118
diff changeset
    97
val queue = ref Task_Queue.empty;
28468
7c80ab57f56d more tuning of tracing messages;
wenzelm
parents: 28464
diff changeset
    98
val next = ref 0;
28192
6d977729c8fa workers: explicit activity flag;
wenzelm
parents: 28191
diff changeset
    99
val workers = ref ([]: (Thread.thread * bool) list);
28177
8c0335bc9336 inherit group from running thread, or create a new one -- make it harder to re-use canceled groups;
wenzelm
parents: 28170
diff changeset
   100
val scheduler = ref (NONE: Thread.thread option);
8c0335bc9336 inherit group from running thread, or create a new one -- make it harder to re-use canceled groups;
wenzelm
parents: 28170
diff changeset
   101
val excessive = ref 0;
29119
99941fd0cb0e renamed structure TaskQueue to Task_Queue;
wenzelm
parents: 29118
diff changeset
   102
val canceled = ref ([]: Task_Queue.group list);
28206
bcd48c6897d4 eliminated requests, use global state variables uniformly;
wenzelm
parents: 28203
diff changeset
   103
val do_shutdown = ref false;
28177
8c0335bc9336 inherit group from running thread, or create a new one -- make it harder to re-use canceled groups;
wenzelm
parents: 28170
diff changeset
   104
8c0335bc9336 inherit group from running thread, or create a new one -- make it harder to re-use canceled groups;
wenzelm
parents: 28170
diff changeset
   105
8c0335bc9336 inherit group from running thread, or create a new one -- make it harder to re-use canceled groups;
wenzelm
parents: 28170
diff changeset
   106
(* synchronization *)
28156
5205f7979b4f Functional threads as future values.
wenzelm
parents:
diff changeset
   107
5205f7979b4f Functional threads as future values.
wenzelm
parents:
diff changeset
   108
local
5205f7979b4f Functional threads as future values.
wenzelm
parents:
diff changeset
   109
  val lock = Mutex.mutex ();
5205f7979b4f Functional threads as future values.
wenzelm
parents:
diff changeset
   110
  val cond = ConditionVar.conditionVar ();
5205f7979b4f Functional threads as future values.
wenzelm
parents:
diff changeset
   111
in
5205f7979b4f Functional threads as future values.
wenzelm
parents:
diff changeset
   112
28575
ed869f019642 SimpleThread.synchronized;
wenzelm
parents: 28558
diff changeset
   113
fun SYNCHRONIZED name = SimpleThread.synchronized name lock;
28156
5205f7979b4f Functional threads as future values.
wenzelm
parents:
diff changeset
   114
29119
99941fd0cb0e renamed structure TaskQueue to Task_Queue;
wenzelm
parents: 29118
diff changeset
   115
fun wait () = (*requires SYNCHRONIZED*)
28206
bcd48c6897d4 eliminated requests, use global state variables uniformly;
wenzelm
parents: 28203
diff changeset
   116
  ConditionVar.wait (cond, lock);
bcd48c6897d4 eliminated requests, use global state variables uniformly;
wenzelm
parents: 28203
diff changeset
   117
29119
99941fd0cb0e renamed structure TaskQueue to Task_Queue;
wenzelm
parents: 29118
diff changeset
   118
fun wait_timeout timeout = (*requires SYNCHRONIZED*)
29341
6bb007a0f9f2 more reactive scheduler: reduced loop timeout, propagate broadcast interrupt via TaskQueue.cancel_all;
wenzelm
parents: 29119
diff changeset
   119
  ignore (ConditionVar.waitUntil (cond, lock, Time.+ (Time.now (), timeout)));
28166
43087721a66e moved task, thread_data, group, queue to task_queue.ML;
wenzelm
parents: 28163
diff changeset
   120
43087721a66e moved task, thread_data, group, queue to task_queue.ML;
wenzelm
parents: 28163
diff changeset
   121
fun notify_all () = (*requires SYNCHRONIZED*)
43087721a66e moved task, thread_data, group, queue to task_queue.ML;
wenzelm
parents: 28163
diff changeset
   122
  ConditionVar.broadcast cond;
28156
5205f7979b4f Functional threads as future values.
wenzelm
parents:
diff changeset
   123
5205f7979b4f Functional threads as future values.
wenzelm
parents:
diff changeset
   124
end;
5205f7979b4f Functional threads as future values.
wenzelm
parents:
diff changeset
   125
5205f7979b4f Functional threads as future values.
wenzelm
parents:
diff changeset
   126
28382
a97fa342540d added release_results (formerly in par_list.ML);
wenzelm
parents: 28380
diff changeset
   127
(* worker activity *)
a97fa342540d added release_results (formerly in par_list.ML);
wenzelm
parents: 28380
diff changeset
   128
a97fa342540d added release_results (formerly in par_list.ML);
wenzelm
parents: 28380
diff changeset
   129
fun trace_active () =
a97fa342540d added release_results (formerly in par_list.ML);
wenzelm
parents: 28380
diff changeset
   130
  let
a97fa342540d added release_results (formerly in par_list.ML);
wenzelm
parents: 28380
diff changeset
   131
    val ws = ! workers;
a97fa342540d added release_results (formerly in par_list.ML);
wenzelm
parents: 28380
diff changeset
   132
    val m = string_of_int (length ws);
a97fa342540d added release_results (formerly in par_list.ML);
wenzelm
parents: 28380
diff changeset
   133
    val n = string_of_int (length (filter #2 ws));
a97fa342540d added release_results (formerly in par_list.ML);
wenzelm
parents: 28380
diff changeset
   134
  in Multithreading.tracing 1 (fn () => "SCHEDULE: " ^ m ^ " workers, " ^ n ^ " active") end;
a97fa342540d added release_results (formerly in par_list.ML);
wenzelm
parents: 28380
diff changeset
   135
a97fa342540d added release_results (formerly in par_list.ML);
wenzelm
parents: 28380
diff changeset
   136
fun change_active active = (*requires SYNCHRONIZED*)
a97fa342540d added release_results (formerly in par_list.ML);
wenzelm
parents: 28380
diff changeset
   137
  change workers (AList.update Thread.equal (Thread.self (), active));
a97fa342540d added release_results (formerly in par_list.ML);
wenzelm
parents: 28380
diff changeset
   138
a97fa342540d added release_results (formerly in par_list.ML);
wenzelm
parents: 28380
diff changeset
   139
29366
1ffc8cbf39ec tuned map: reduced overhead due to bulk jobs;
wenzelm
parents: 29341
diff changeset
   140
(* execute jobs *)
28156
5205f7979b4f Functional threads as future values.
wenzelm
parents:
diff changeset
   141
29341
6bb007a0f9f2 more reactive scheduler: reduced loop timeout, propagate broadcast interrupt via TaskQueue.cancel_all;
wenzelm
parents: 29119
diff changeset
   142
fun do_cancel group = (*requires SYNCHRONIZED*)
6bb007a0f9f2 more reactive scheduler: reduced loop timeout, propagate broadcast interrupt via TaskQueue.cancel_all;
wenzelm
parents: 29119
diff changeset
   143
  change canceled (insert Task_Queue.eq_group group);
6bb007a0f9f2 more reactive scheduler: reduced loop timeout, propagate broadcast interrupt via TaskQueue.cancel_all;
wenzelm
parents: 29119
diff changeset
   144
29366
1ffc8cbf39ec tuned map: reduced overhead due to bulk jobs;
wenzelm
parents: 29341
diff changeset
   145
fun execute name (task, group, jobs) =
28167
27e2ca41b58c more interrupt operations;
wenzelm
parents: 28166
diff changeset
   146
  let
28382
a97fa342540d added release_results (formerly in par_list.ML);
wenzelm
parents: 28380
diff changeset
   147
    val _ = trace_active ();
29366
1ffc8cbf39ec tuned map: reduced overhead due to bulk jobs;
wenzelm
parents: 29341
diff changeset
   148
    val ok = setmp_thread_data (name, task) (fn () =>
1ffc8cbf39ec tuned map: reduced overhead due to bulk jobs;
wenzelm
parents: 29341
diff changeset
   149
      fold (fn job => fn ok => job ok) jobs (Task_Queue.is_valid group)) ();
28192
6d977729c8fa workers: explicit activity flag;
wenzelm
parents: 28191
diff changeset
   150
    val _ = SYNCHRONIZED "execute" (fn () =>
29119
99941fd0cb0e renamed structure TaskQueue to Task_Queue;
wenzelm
parents: 29118
diff changeset
   151
     (change queue (Task_Queue.finish task);
28186
6a8417f36837 cancel: check_scheduler;
wenzelm
parents: 28177
diff changeset
   152
      if ok then ()
29119
99941fd0cb0e renamed structure TaskQueue to Task_Queue;
wenzelm
parents: 29118
diff changeset
   153
      else if Task_Queue.cancel (! queue) group then ()
29341
6bb007a0f9f2 more reactive scheduler: reduced loop timeout, propagate broadcast interrupt via TaskQueue.cancel_all;
wenzelm
parents: 29119
diff changeset
   154
      else do_cancel group;
28177
8c0335bc9336 inherit group from running thread, or create a new one -- make it harder to re-use canceled groups;
wenzelm
parents: 28170
diff changeset
   155
      notify_all ()));
28167
27e2ca41b58c more interrupt operations;
wenzelm
parents: 28166
diff changeset
   156
  in () end;
27e2ca41b58c more interrupt operations;
wenzelm
parents: 28166
diff changeset
   157
27e2ca41b58c more interrupt operations;
wenzelm
parents: 28166
diff changeset
   158
27e2ca41b58c more interrupt operations;
wenzelm
parents: 28166
diff changeset
   159
(* worker threads *)
27e2ca41b58c more interrupt operations;
wenzelm
parents: 28166
diff changeset
   160
29119
99941fd0cb0e renamed structure TaskQueue to Task_Queue;
wenzelm
parents: 29118
diff changeset
   161
fun worker_wait () = (*requires SYNCHRONIZED*)
99941fd0cb0e renamed structure TaskQueue to Task_Queue;
wenzelm
parents: 29118
diff changeset
   162
  (change_active false; wait (); change_active true);
28162
55772e4e95e0 tuned check_cache;
wenzelm
parents: 28156
diff changeset
   163
29119
99941fd0cb0e renamed structure TaskQueue to Task_Queue;
wenzelm
parents: 29118
diff changeset
   164
fun worker_next () = (*requires SYNCHRONIZED*)
28167
27e2ca41b58c more interrupt operations;
wenzelm
parents: 28166
diff changeset
   165
  if ! excessive > 0 then
27e2ca41b58c more interrupt operations;
wenzelm
parents: 28166
diff changeset
   166
    (dec excessive;
28192
6d977729c8fa workers: explicit activity flag;
wenzelm
parents: 28191
diff changeset
   167
     change workers (filter_out (fn (thread, _) => Thread.equal (thread, Thread.self ())));
28203
88f18387f1c9 shutdown: global join-and-shutdown operation;
wenzelm
parents: 28202
diff changeset
   168
     notify_all ();
28167
27e2ca41b58c more interrupt operations;
wenzelm
parents: 28166
diff changeset
   169
     NONE)
28166
43087721a66e moved task, thread_data, group, queue to task_queue.ML;
wenzelm
parents: 28163
diff changeset
   170
  else
29119
99941fd0cb0e renamed structure TaskQueue to Task_Queue;
wenzelm
parents: 29118
diff changeset
   171
    (case change_result queue Task_Queue.dequeue of
99941fd0cb0e renamed structure TaskQueue to Task_Queue;
wenzelm
parents: 29118
diff changeset
   172
      NONE => (worker_wait (); worker_next ())
28166
43087721a66e moved task, thread_data, group, queue to task_queue.ML;
wenzelm
parents: 28163
diff changeset
   173
    | some => some);
28156
5205f7979b4f Functional threads as future values.
wenzelm
parents:
diff changeset
   174
28167
27e2ca41b58c more interrupt operations;
wenzelm
parents: 28166
diff changeset
   175
fun worker_loop name =
29119
99941fd0cb0e renamed structure TaskQueue to Task_Queue;
wenzelm
parents: 29118
diff changeset
   176
  (case SYNCHRONIZED name worker_next of
99941fd0cb0e renamed structure TaskQueue to Task_Queue;
wenzelm
parents: 29118
diff changeset
   177
    NONE => ()
28167
27e2ca41b58c more interrupt operations;
wenzelm
parents: 28166
diff changeset
   178
  | SOME work => (execute name work; worker_loop name));
28156
5205f7979b4f Functional threads as future values.
wenzelm
parents:
diff changeset
   179
28167
27e2ca41b58c more interrupt operations;
wenzelm
parents: 28166
diff changeset
   180
fun worker_start name = (*requires SYNCHRONIZED*)
28242
f978c8e75118 SimpleThread.fork;
wenzelm
parents: 28208
diff changeset
   181
  change workers (cons (SimpleThread.fork false (fn () => worker_loop name), true));
28156
5205f7979b4f Functional threads as future values.
wenzelm
parents:
diff changeset
   182
5205f7979b4f Functional threads as future values.
wenzelm
parents:
diff changeset
   183
5205f7979b4f Functional threads as future values.
wenzelm
parents:
diff changeset
   184
(* scheduler *)
5205f7979b4f Functional threads as future values.
wenzelm
parents:
diff changeset
   185
28206
bcd48c6897d4 eliminated requests, use global state variables uniformly;
wenzelm
parents: 28203
diff changeset
   186
fun scheduler_next () = (*requires SYNCHRONIZED*)
28156
5205f7979b4f Functional threads as future values.
wenzelm
parents:
diff changeset
   187
  let
28206
bcd48c6897d4 eliminated requests, use global state variables uniformly;
wenzelm
parents: 28203
diff changeset
   188
    (*worker threads*)
28191
9e5f556409c6 future: allow explicit group;
wenzelm
parents: 28186
diff changeset
   189
    val _ =
28192
6d977729c8fa workers: explicit activity flag;
wenzelm
parents: 28191
diff changeset
   190
      (case List.partition (Thread.isActive o #1) (! workers) of
28191
9e5f556409c6 future: allow explicit group;
wenzelm
parents: 28186
diff changeset
   191
        (_, []) => ()
9e5f556409c6 future: allow explicit group;
wenzelm
parents: 28186
diff changeset
   192
      | (active, inactive) =>
9e5f556409c6 future: allow explicit group;
wenzelm
parents: 28186
diff changeset
   193
          (workers := active; Multithreading.tracing 0 (fn () =>
28192
6d977729c8fa workers: explicit activity flag;
wenzelm
parents: 28191
diff changeset
   194
            "SCHEDULE: disposed " ^ string_of_int (length inactive) ^ " dead worker threads")));
28382
a97fa342540d added release_results (formerly in par_list.ML);
wenzelm
parents: 28380
diff changeset
   195
    val _ = trace_active ();
28191
9e5f556409c6 future: allow explicit group;
wenzelm
parents: 28186
diff changeset
   196
28206
bcd48c6897d4 eliminated requests, use global state variables uniformly;
wenzelm
parents: 28203
diff changeset
   197
    val m = if ! do_shutdown then 0 else Multithreading.max_threads_value ();
28167
27e2ca41b58c more interrupt operations;
wenzelm
parents: 28166
diff changeset
   198
    val l = length (! workers);
27e2ca41b58c more interrupt operations;
wenzelm
parents: 28166
diff changeset
   199
    val _ = excessive := l - m;
28203
88f18387f1c9 shutdown: global join-and-shutdown operation;
wenzelm
parents: 28202
diff changeset
   200
    val _ =
28468
7c80ab57f56d more tuning of tracing messages;
wenzelm
parents: 28464
diff changeset
   201
      if m > l then funpow (m - l) (fn () => worker_start ("worker " ^ string_of_int (inc next))) ()
28203
88f18387f1c9 shutdown: global join-and-shutdown operation;
wenzelm
parents: 28202
diff changeset
   202
      else ();
28206
bcd48c6897d4 eliminated requests, use global state variables uniformly;
wenzelm
parents: 28203
diff changeset
   203
bcd48c6897d4 eliminated requests, use global state variables uniformly;
wenzelm
parents: 28203
diff changeset
   204
    (*canceled groups*)
29119
99941fd0cb0e renamed structure TaskQueue to Task_Queue;
wenzelm
parents: 29118
diff changeset
   205
    val _ =  change canceled (filter_out (Task_Queue.cancel (! queue)));
28206
bcd48c6897d4 eliminated requests, use global state variables uniformly;
wenzelm
parents: 28203
diff changeset
   206
bcd48c6897d4 eliminated requests, use global state variables uniformly;
wenzelm
parents: 28203
diff changeset
   207
    (*shutdown*)
bcd48c6897d4 eliminated requests, use global state variables uniformly;
wenzelm
parents: 28203
diff changeset
   208
    val continue = not (! do_shutdown andalso null (! workers));
bcd48c6897d4 eliminated requests, use global state variables uniformly;
wenzelm
parents: 28203
diff changeset
   209
    val _ = if continue then () else scheduler := NONE;
28167
27e2ca41b58c more interrupt operations;
wenzelm
parents: 28166
diff changeset
   210
28206
bcd48c6897d4 eliminated requests, use global state variables uniformly;
wenzelm
parents: 28203
diff changeset
   211
    val _ = notify_all ();
29341
6bb007a0f9f2 more reactive scheduler: reduced loop timeout, propagate broadcast interrupt via TaskQueue.cancel_all;
wenzelm
parents: 29119
diff changeset
   212
    val _ = interruptible (fn () => wait_timeout (Time.fromSeconds 1)) ()
6bb007a0f9f2 more reactive scheduler: reduced loop timeout, propagate broadcast interrupt via TaskQueue.cancel_all;
wenzelm
parents: 29119
diff changeset
   213
      handle Exn.Interrupt => List.app do_cancel (Task_Queue.cancel_all (! queue));
28206
bcd48c6897d4 eliminated requests, use global state variables uniformly;
wenzelm
parents: 28203
diff changeset
   214
  in continue end;
bcd48c6897d4 eliminated requests, use global state variables uniformly;
wenzelm
parents: 28203
diff changeset
   215
bcd48c6897d4 eliminated requests, use global state variables uniformly;
wenzelm
parents: 28203
diff changeset
   216
fun scheduler_loop () =
29119
99941fd0cb0e renamed structure TaskQueue to Task_Queue;
wenzelm
parents: 29118
diff changeset
   217
  while SYNCHRONIZED "scheduler" scheduler_next do ();
28156
5205f7979b4f Functional threads as future values.
wenzelm
parents:
diff changeset
   218
28203
88f18387f1c9 shutdown: global join-and-shutdown operation;
wenzelm
parents: 28202
diff changeset
   219
fun scheduler_active () = (*requires SYNCHRONIZED*)
88f18387f1c9 shutdown: global join-and-shutdown operation;
wenzelm
parents: 28202
diff changeset
   220
  (case ! scheduler of NONE => false | SOME thread => Thread.isActive thread);
88f18387f1c9 shutdown: global join-and-shutdown operation;
wenzelm
parents: 28202
diff changeset
   221
28464
dcc030b52583 tuned SYNCHRONIZED: outermost Exn.release;
wenzelm
parents: 28442
diff changeset
   222
fun scheduler_check name = SYNCHRONIZED name (fn () =>
28206
bcd48c6897d4 eliminated requests, use global state variables uniformly;
wenzelm
parents: 28203
diff changeset
   223
  if not (scheduler_active ()) then
29119
99941fd0cb0e renamed structure TaskQueue to Task_Queue;
wenzelm
parents: 29118
diff changeset
   224
    (do_shutdown := false; scheduler := SOME (SimpleThread.fork false scheduler_loop))
28206
bcd48c6897d4 eliminated requests, use global state variables uniformly;
wenzelm
parents: 28203
diff changeset
   225
  else if ! do_shutdown then error "Scheduler shutdown in progress"
bcd48c6897d4 eliminated requests, use global state variables uniformly;
wenzelm
parents: 28203
diff changeset
   226
  else ());
28156
5205f7979b4f Functional threads as future values.
wenzelm
parents:
diff changeset
   227
5205f7979b4f Functional threads as future values.
wenzelm
parents:
diff changeset
   228
29366
1ffc8cbf39ec tuned map: reduced overhead due to bulk jobs;
wenzelm
parents: 29341
diff changeset
   229
1ffc8cbf39ec tuned map: reduced overhead due to bulk jobs;
wenzelm
parents: 29341
diff changeset
   230
(** futures **)
28156
5205f7979b4f Functional threads as future values.
wenzelm
parents:
diff changeset
   231
29366
1ffc8cbf39ec tuned map: reduced overhead due to bulk jobs;
wenzelm
parents: 29341
diff changeset
   232
(* future job: fill result *)
1ffc8cbf39ec tuned map: reduced overhead due to bulk jobs;
wenzelm
parents: 29341
diff changeset
   233
1ffc8cbf39ec tuned map: reduced overhead due to bulk jobs;
wenzelm
parents: 29341
diff changeset
   234
fun future_job group (e: unit -> 'a) =
28156
5205f7979b4f Functional threads as future values.
wenzelm
parents:
diff changeset
   235
  let
28166
43087721a66e moved task, thread_data, group, queue to task_queue.ML;
wenzelm
parents: 28163
diff changeset
   236
    val result = ref (NONE: 'a Exn.result option);
29366
1ffc8cbf39ec tuned map: reduced overhead due to bulk jobs;
wenzelm
parents: 29341
diff changeset
   237
    val job = Multithreading.with_attributes (Thread.getAttributes ())
28177
8c0335bc9336 inherit group from running thread, or create a new one -- make it harder to re-use canceled groups;
wenzelm
parents: 28170
diff changeset
   238
      (fn _ => fn ok =>
28532
16c6ae7d1aa6 more careful handling of group interrupts;
wenzelm
parents: 28472
diff changeset
   239
        let
16c6ae7d1aa6 more careful handling of group interrupts;
wenzelm
parents: 28472
diff changeset
   240
          val res = if ok then Exn.capture e () else Exn.Exn Exn.Interrupt;
28548
003f52c2bb8f future result: Interrupt invalidates group, but pretends success otherwise;
wenzelm
parents: 28534
diff changeset
   241
          val _ = result := SOME res;
28532
16c6ae7d1aa6 more careful handling of group interrupts;
wenzelm
parents: 28472
diff changeset
   242
          val res_ok =
16c6ae7d1aa6 more careful handling of group interrupts;
wenzelm
parents: 28472
diff changeset
   243
            (case res of
16c6ae7d1aa6 more careful handling of group interrupts;
wenzelm
parents: 28472
diff changeset
   244
              Exn.Result _ => true
29119
99941fd0cb0e renamed structure TaskQueue to Task_Queue;
wenzelm
parents: 29118
diff changeset
   245
            | Exn.Exn Exn.Interrupt => (Task_Queue.invalidate_group group; true)
28532
16c6ae7d1aa6 more careful handling of group interrupts;
wenzelm
parents: 28472
diff changeset
   246
            | _ => false);
28548
003f52c2bb8f future result: Interrupt invalidates group, but pretends success otherwise;
wenzelm
parents: 28534
diff changeset
   247
        in res_ok end);
29366
1ffc8cbf39ec tuned map: reduced overhead due to bulk jobs;
wenzelm
parents: 29341
diff changeset
   248
  in (result, job) end;
28177
8c0335bc9336 inherit group from running thread, or create a new one -- make it harder to re-use canceled groups;
wenzelm
parents: 28170
diff changeset
   249
29366
1ffc8cbf39ec tuned map: reduced overhead due to bulk jobs;
wenzelm
parents: 29341
diff changeset
   250
1ffc8cbf39ec tuned map: reduced overhead due to bulk jobs;
wenzelm
parents: 29341
diff changeset
   251
(* fork *)
1ffc8cbf39ec tuned map: reduced overhead due to bulk jobs;
wenzelm
parents: 29341
diff changeset
   252
1ffc8cbf39ec tuned map: reduced overhead due to bulk jobs;
wenzelm
parents: 29341
diff changeset
   253
fun fork_future opt_group deps pri e =
1ffc8cbf39ec tuned map: reduced overhead due to bulk jobs;
wenzelm
parents: 29341
diff changeset
   254
  let
1ffc8cbf39ec tuned map: reduced overhead due to bulk jobs;
wenzelm
parents: 29341
diff changeset
   255
    val _ = scheduler_check "future check";
1ffc8cbf39ec tuned map: reduced overhead due to bulk jobs;
wenzelm
parents: 29341
diff changeset
   256
1ffc8cbf39ec tuned map: reduced overhead due to bulk jobs;
wenzelm
parents: 29341
diff changeset
   257
    val group = (case opt_group of SOME group => group | NONE => Task_Queue.new_group ());
1ffc8cbf39ec tuned map: reduced overhead due to bulk jobs;
wenzelm
parents: 29341
diff changeset
   258
    val (result, job) = future_job group e;
28192
6d977729c8fa workers: explicit activity flag;
wenzelm
parents: 28191
diff changeset
   259
    val task = SYNCHRONIZED "future" (fn () =>
29366
1ffc8cbf39ec tuned map: reduced overhead due to bulk jobs;
wenzelm
parents: 29341
diff changeset
   260
      change_result queue (Task_Queue.enqueue group deps pri job) before notify_all ());
28166
43087721a66e moved task, thread_data, group, queue to task_queue.ML;
wenzelm
parents: 28163
diff changeset
   261
  in Future {task = task, group = group, result = result} end;
28162
55772e4e95e0 tuned check_cache;
wenzelm
parents: 28156
diff changeset
   262
29366
1ffc8cbf39ec tuned map: reduced overhead due to bulk jobs;
wenzelm
parents: 29341
diff changeset
   263
fun fork e = fork_future NONE [] 0 e;
1ffc8cbf39ec tuned map: reduced overhead due to bulk jobs;
wenzelm
parents: 29341
diff changeset
   264
fun fork_group group e = fork_future (SOME group) [] 0 e;
1ffc8cbf39ec tuned map: reduced overhead due to bulk jobs;
wenzelm
parents: 29341
diff changeset
   265
fun fork_deps deps e = fork_future NONE (map task_of deps) 0 e;
1ffc8cbf39ec tuned map: reduced overhead due to bulk jobs;
wenzelm
parents: 29341
diff changeset
   266
fun fork_pri pri e = fork_future NONE [] pri e;
28186
6a8417f36837 cancel: check_scheduler;
wenzelm
parents: 28177
diff changeset
   267
6a8417f36837 cancel: check_scheduler;
wenzelm
parents: 28177
diff changeset
   268
29366
1ffc8cbf39ec tuned map: reduced overhead due to bulk jobs;
wenzelm
parents: 29341
diff changeset
   269
(* join *)
1ffc8cbf39ec tuned map: reduced overhead due to bulk jobs;
wenzelm
parents: 29341
diff changeset
   270
1ffc8cbf39ec tuned map: reduced overhead due to bulk jobs;
wenzelm
parents: 29341
diff changeset
   271
fun get_result x = the_default (Exn.Exn (SYS_ERROR "unfinished future")) (peek x);
28186
6a8417f36837 cancel: check_scheduler;
wenzelm
parents: 28177
diff changeset
   272
29366
1ffc8cbf39ec tuned map: reduced overhead due to bulk jobs;
wenzelm
parents: 29341
diff changeset
   273
fun join_results xs =
1ffc8cbf39ec tuned map: reduced overhead due to bulk jobs;
wenzelm
parents: 29341
diff changeset
   274
  if forall is_finished xs then map get_result xs
1ffc8cbf39ec tuned map: reduced overhead due to bulk jobs;
wenzelm
parents: 29341
diff changeset
   275
  else uninterruptible (fn _ => fn () =>
1ffc8cbf39ec tuned map: reduced overhead due to bulk jobs;
wenzelm
parents: 29341
diff changeset
   276
    let
1ffc8cbf39ec tuned map: reduced overhead due to bulk jobs;
wenzelm
parents: 29341
diff changeset
   277
      val _ = scheduler_check "join check";
1ffc8cbf39ec tuned map: reduced overhead due to bulk jobs;
wenzelm
parents: 29341
diff changeset
   278
      val _ = Multithreading.self_critical () andalso
1ffc8cbf39ec tuned map: reduced overhead due to bulk jobs;
wenzelm
parents: 29341
diff changeset
   279
        error "Cannot join future values within critical section";
28177
8c0335bc9336 inherit group from running thread, or create a new one -- make it harder to re-use canceled groups;
wenzelm
parents: 28170
diff changeset
   280
29366
1ffc8cbf39ec tuned map: reduced overhead due to bulk jobs;
wenzelm
parents: 29341
diff changeset
   281
      fun join_loop _ [] = ()
1ffc8cbf39ec tuned map: reduced overhead due to bulk jobs;
wenzelm
parents: 29341
diff changeset
   282
        | join_loop name deps =
1ffc8cbf39ec tuned map: reduced overhead due to bulk jobs;
wenzelm
parents: 29341
diff changeset
   283
            (case SYNCHRONIZED name (fn () =>
1ffc8cbf39ec tuned map: reduced overhead due to bulk jobs;
wenzelm
parents: 29341
diff changeset
   284
                change_result queue (Task_Queue.dequeue_towards deps)) of
1ffc8cbf39ec tuned map: reduced overhead due to bulk jobs;
wenzelm
parents: 29341
diff changeset
   285
              NONE => ()
1ffc8cbf39ec tuned map: reduced overhead due to bulk jobs;
wenzelm
parents: 29341
diff changeset
   286
            | SOME (work, deps') => (execute name work; join_loop name deps'));
1ffc8cbf39ec tuned map: reduced overhead due to bulk jobs;
wenzelm
parents: 29341
diff changeset
   287
      val _ =
1ffc8cbf39ec tuned map: reduced overhead due to bulk jobs;
wenzelm
parents: 29341
diff changeset
   288
        (case thread_data () of
1ffc8cbf39ec tuned map: reduced overhead due to bulk jobs;
wenzelm
parents: 29341
diff changeset
   289
          NONE =>
1ffc8cbf39ec tuned map: reduced overhead due to bulk jobs;
wenzelm
parents: 29341
diff changeset
   290
            (*alien thread -- refrain from contending for resources*)
1ffc8cbf39ec tuned map: reduced overhead due to bulk jobs;
wenzelm
parents: 29341
diff changeset
   291
            while not (forall is_finished xs)
1ffc8cbf39ec tuned map: reduced overhead due to bulk jobs;
wenzelm
parents: 29341
diff changeset
   292
            do SYNCHRONIZED "join_thread" (fn () => wait ())
1ffc8cbf39ec tuned map: reduced overhead due to bulk jobs;
wenzelm
parents: 29341
diff changeset
   293
        | SOME (name, task) =>
1ffc8cbf39ec tuned map: reduced overhead due to bulk jobs;
wenzelm
parents: 29341
diff changeset
   294
            (*proper task -- actively work towards results*)
1ffc8cbf39ec tuned map: reduced overhead due to bulk jobs;
wenzelm
parents: 29341
diff changeset
   295
            let
1ffc8cbf39ec tuned map: reduced overhead due to bulk jobs;
wenzelm
parents: 29341
diff changeset
   296
              val pending = filter_out is_finished xs;
1ffc8cbf39ec tuned map: reduced overhead due to bulk jobs;
wenzelm
parents: 29341
diff changeset
   297
              val deps = map task_of pending;
1ffc8cbf39ec tuned map: reduced overhead due to bulk jobs;
wenzelm
parents: 29341
diff changeset
   298
              val _ = SYNCHRONIZED "join" (fn () =>
1ffc8cbf39ec tuned map: reduced overhead due to bulk jobs;
wenzelm
parents: 29341
diff changeset
   299
                (change queue (Task_Queue.depend deps task); notify_all ()));
1ffc8cbf39ec tuned map: reduced overhead due to bulk jobs;
wenzelm
parents: 29341
diff changeset
   300
              val _ = join_loop ("join_loop: " ^ name) deps;
1ffc8cbf39ec tuned map: reduced overhead due to bulk jobs;
wenzelm
parents: 29341
diff changeset
   301
              val _ =
1ffc8cbf39ec tuned map: reduced overhead due to bulk jobs;
wenzelm
parents: 29341
diff changeset
   302
                while not (forall is_finished pending)
1ffc8cbf39ec tuned map: reduced overhead due to bulk jobs;
wenzelm
parents: 29341
diff changeset
   303
                do SYNCHRONIZED "join_task" (fn () => worker_wait ());
1ffc8cbf39ec tuned map: reduced overhead due to bulk jobs;
wenzelm
parents: 29341
diff changeset
   304
            in () end);
28186
6a8417f36837 cancel: check_scheduler;
wenzelm
parents: 28177
diff changeset
   305
29366
1ffc8cbf39ec tuned map: reduced overhead due to bulk jobs;
wenzelm
parents: 29341
diff changeset
   306
    in map get_result xs end) ();
28186
6a8417f36837 cancel: check_scheduler;
wenzelm
parents: 28177
diff changeset
   307
28647
8068cdc84e7e join_results: allow CRITICAL join of finished futures;
wenzelm
parents: 28645
diff changeset
   308
fun join_result x = singleton join_results x;
8068cdc84e7e join_results: allow CRITICAL join of finished futures;
wenzelm
parents: 28645
diff changeset
   309
fun join x = Exn.release (join_result x);
28156
5205f7979b4f Functional threads as future values.
wenzelm
parents:
diff changeset
   310
29366
1ffc8cbf39ec tuned map: reduced overhead due to bulk jobs;
wenzelm
parents: 29341
diff changeset
   311
1ffc8cbf39ec tuned map: reduced overhead due to bulk jobs;
wenzelm
parents: 29341
diff changeset
   312
(* map *)
1ffc8cbf39ec tuned map: reduced overhead due to bulk jobs;
wenzelm
parents: 29341
diff changeset
   313
1ffc8cbf39ec tuned map: reduced overhead due to bulk jobs;
wenzelm
parents: 29341
diff changeset
   314
fun map_future f (x as Future {task, group, ...}) =
1ffc8cbf39ec tuned map: reduced overhead due to bulk jobs;
wenzelm
parents: 29341
diff changeset
   315
  let
1ffc8cbf39ec tuned map: reduced overhead due to bulk jobs;
wenzelm
parents: 29341
diff changeset
   316
    val _ = scheduler_check "map_future check";
1ffc8cbf39ec tuned map: reduced overhead due to bulk jobs;
wenzelm
parents: 29341
diff changeset
   317
1ffc8cbf39ec tuned map: reduced overhead due to bulk jobs;
wenzelm
parents: 29341
diff changeset
   318
    val (result', job) = future_job group (fn () => f (join x));
1ffc8cbf39ec tuned map: reduced overhead due to bulk jobs;
wenzelm
parents: 29341
diff changeset
   319
    val extended = SYNCHRONIZED "map_future" (fn () =>
1ffc8cbf39ec tuned map: reduced overhead due to bulk jobs;
wenzelm
parents: 29341
diff changeset
   320
      (case Task_Queue.extend task job (! queue) of
1ffc8cbf39ec tuned map: reduced overhead due to bulk jobs;
wenzelm
parents: 29341
diff changeset
   321
        SOME queue' => (queue := queue'; true)
1ffc8cbf39ec tuned map: reduced overhead due to bulk jobs;
wenzelm
parents: 29341
diff changeset
   322
      | NONE => false));
1ffc8cbf39ec tuned map: reduced overhead due to bulk jobs;
wenzelm
parents: 29341
diff changeset
   323
  in
1ffc8cbf39ec tuned map: reduced overhead due to bulk jobs;
wenzelm
parents: 29341
diff changeset
   324
    if extended then Future {task = task, group = group, result = result'}
1ffc8cbf39ec tuned map: reduced overhead due to bulk jobs;
wenzelm
parents: 29341
diff changeset
   325
    else fork_future (SOME group) [task] (Task_Queue.pri_of_task task) (fn () => f (join x))
1ffc8cbf39ec tuned map: reduced overhead due to bulk jobs;
wenzelm
parents: 29341
diff changeset
   326
  end;
28979
3ce619d8d432 fork/map: no inheritance of group (structure is nested, not parallel);
wenzelm
parents: 28972
diff changeset
   327
28191
9e5f556409c6 future: allow explicit group;
wenzelm
parents: 28186
diff changeset
   328
29366
1ffc8cbf39ec tuned map: reduced overhead due to bulk jobs;
wenzelm
parents: 29341
diff changeset
   329
(* cancel *)
28202
23cb9a974630 added focus, which indicates a particular collection of high-priority tasks;
wenzelm
parents: 28201
diff changeset
   330
23cb9a974630 added focus, which indicates a particular collection of high-priority tasks;
wenzelm
parents: 28201
diff changeset
   331
(*interrupt: permissive signal, may get ignored*)
28197
7053c539ecd8 added interrupt_task (external id);
wenzelm
parents: 28193
diff changeset
   332
fun interrupt_task id = SYNCHRONIZED "interrupt"
29119
99941fd0cb0e renamed structure TaskQueue to Task_Queue;
wenzelm
parents: 29118
diff changeset
   333
  (fn () => Task_Queue.interrupt_external (! queue) id);
28191
9e5f556409c6 future: allow explicit group;
wenzelm
parents: 28186
diff changeset
   334
28206
bcd48c6897d4 eliminated requests, use global state variables uniformly;
wenzelm
parents: 28203
diff changeset
   335
(*cancel: present and future group members will be interrupted eventually*)
bcd48c6897d4 eliminated requests, use global state variables uniformly;
wenzelm
parents: 28203
diff changeset
   336
fun cancel x =
28464
dcc030b52583 tuned SYNCHRONIZED: outermost Exn.release;
wenzelm
parents: 28442
diff changeset
   337
 (scheduler_check "cancel check";
29341
6bb007a0f9f2 more reactive scheduler: reduced loop timeout, propagate broadcast interrupt via TaskQueue.cancel_all;
wenzelm
parents: 29119
diff changeset
   338
  SYNCHRONIZED "cancel" (fn () => (do_cancel (group_of x); notify_all ())));
28206
bcd48c6897d4 eliminated requests, use global state variables uniformly;
wenzelm
parents: 28203
diff changeset
   339
bcd48c6897d4 eliminated requests, use global state variables uniformly;
wenzelm
parents: 28203
diff changeset
   340
29366
1ffc8cbf39ec tuned map: reduced overhead due to bulk jobs;
wenzelm
parents: 29341
diff changeset
   341
1ffc8cbf39ec tuned map: reduced overhead due to bulk jobs;
wenzelm
parents: 29341
diff changeset
   342
(** global join and shutdown **)
1ffc8cbf39ec tuned map: reduced overhead due to bulk jobs;
wenzelm
parents: 29341
diff changeset
   343
28203
88f18387f1c9 shutdown: global join-and-shutdown operation;
wenzelm
parents: 28202
diff changeset
   344
fun shutdown () =
28276
fbc707811203 shutdown only if Multithreading.available;
wenzelm
parents: 28242
diff changeset
   345
  if Multithreading.available then
28464
dcc030b52583 tuned SYNCHRONIZED: outermost Exn.release;
wenzelm
parents: 28442
diff changeset
   346
   (scheduler_check "shutdown check";
28276
fbc707811203 shutdown only if Multithreading.available;
wenzelm
parents: 28242
diff changeset
   347
    SYNCHRONIZED "shutdown" (fn () =>
29119
99941fd0cb0e renamed structure TaskQueue to Task_Queue;
wenzelm
parents: 29118
diff changeset
   348
     (while not (scheduler_active ()) do wait ();
99941fd0cb0e renamed structure TaskQueue to Task_Queue;
wenzelm
parents: 29118
diff changeset
   349
      while not (Task_Queue.is_empty (! queue)) do wait ();
28276
fbc707811203 shutdown only if Multithreading.available;
wenzelm
parents: 28242
diff changeset
   350
      do_shutdown := true;
fbc707811203 shutdown only if Multithreading.available;
wenzelm
parents: 28242
diff changeset
   351
      notify_all ();
29119
99941fd0cb0e renamed structure TaskQueue to Task_Queue;
wenzelm
parents: 29118
diff changeset
   352
      while not (null (! workers)) do wait ();
99941fd0cb0e renamed structure TaskQueue to Task_Queue;
wenzelm
parents: 29118
diff changeset
   353
      while scheduler_active () do wait ();
28470
409fedeece30 added simple heartbeat thread;
wenzelm
parents: 28468
diff changeset
   354
      OS.Process.sleep (Time.fromMilliseconds 300))))
28276
fbc707811203 shutdown only if Multithreading.available;
wenzelm
parents: 28242
diff changeset
   355
  else ();
28203
88f18387f1c9 shutdown: global join-and-shutdown operation;
wenzelm
parents: 28202
diff changeset
   356
29366
1ffc8cbf39ec tuned map: reduced overhead due to bulk jobs;
wenzelm
parents: 29341
diff changeset
   357
1ffc8cbf39ec tuned map: reduced overhead due to bulk jobs;
wenzelm
parents: 29341
diff changeset
   358
(*final declarations of this structure!*)
1ffc8cbf39ec tuned map: reduced overhead due to bulk jobs;
wenzelm
parents: 29341
diff changeset
   359
val map = map_future;
1ffc8cbf39ec tuned map: reduced overhead due to bulk jobs;
wenzelm
parents: 29341
diff changeset
   360
28156
5205f7979b4f Functional threads as future values.
wenzelm
parents:
diff changeset
   361
end;
28972
cb8a2c3e188f renamed type Future.T to future;
wenzelm
parents: 28647
diff changeset
   362
cb8a2c3e188f renamed type Future.T to future;
wenzelm
parents: 28647
diff changeset
   363
type 'a future = 'a Future.future;
cb8a2c3e188f renamed type Future.T to future;
wenzelm
parents: 28647
diff changeset
   364