src/Pure/Concurrent/future.ML
author wenzelm
Tue, 21 Jul 2009 20:37:31 +0200
changeset 32102 81d03a29980c
parent 32099 5382c93108db
child 32107 47d0da617fcc
permissions -rw-r--r--
added worker_group; fork: default to subgroup of worker_group; removed obsolete fork_local; join/get_result: cumulative flattened exceptions; map: subgroup of worker_group;
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
32058
c76fd93b3b99 more abstract Future.is_worker;
wenzelm
parents: 32055
diff changeset
    32
  val is_worker: unit -> bool
32102
81d03a29980c added worker_group;
wenzelm
parents: 32099
diff changeset
    33
  val worker_group: unit -> Task_Queue.group option
28972
cb8a2c3e188f renamed type Future.T to future;
wenzelm
parents: 28647
diff changeset
    34
  type 'a future
cb8a2c3e188f renamed type Future.T to future;
wenzelm
parents: 28647
diff changeset
    35
  val task_of: 'a future -> task
cb8a2c3e188f renamed type Future.T to future;
wenzelm
parents: 28647
diff changeset
    36
  val group_of: 'a future -> group
cb8a2c3e188f renamed type Future.T to future;
wenzelm
parents: 28647
diff changeset
    37
  val peek: 'a future -> 'a Exn.result option
cb8a2c3e188f renamed type Future.T to future;
wenzelm
parents: 28647
diff changeset
    38
  val is_finished: 'a future -> bool
28997
1b99dcae2156 added constant value;
wenzelm
parents: 28979
diff changeset
    39
  val value: 'a -> 'a future
28972
cb8a2c3e188f renamed type Future.T to future;
wenzelm
parents: 28647
diff changeset
    40
  val fork: (unit -> 'a) -> 'a future
28979
3ce619d8d432 fork/map: no inheritance of group (structure is nested, not parallel);
wenzelm
parents: 28972
diff changeset
    41
  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
    42
  val fork_deps: 'b future list -> (unit -> 'a) -> 'a future
29119
99941fd0cb0e renamed structure TaskQueue to Task_Queue;
wenzelm
parents: 29118
diff changeset
    43
  val fork_pri: int -> (unit -> 'a) -> 'a future
28972
cb8a2c3e188f renamed type Future.T to future;
wenzelm
parents: 28647
diff changeset
    44
  val join_results: 'a future list -> 'a Exn.result list
cb8a2c3e188f renamed type Future.T to future;
wenzelm
parents: 28647
diff changeset
    45
  val join_result: 'a future -> 'a Exn.result
cb8a2c3e188f renamed type Future.T to future;
wenzelm
parents: 28647
diff changeset
    46
  val join: 'a future -> 'a
cb8a2c3e188f renamed type Future.T to future;
wenzelm
parents: 28647
diff changeset
    47
  val map: ('a -> 'b) -> 'a future -> 'b future
30618
046f4f986fb5 restricted interrupts for tasks running as future worker thread -- attempt to prevent interrupt race conditions;
wenzelm
parents: 30612
diff changeset
    48
  val interruptible_task: ('a -> 'b) -> 'a -> 'b
28206
bcd48c6897d4 eliminated requests, use global state variables uniformly;
wenzelm
parents: 28203
diff changeset
    49
  val interrupt_task: string -> unit
29431
0ebe652bfd5a added cancel_group;
wenzelm
parents: 29384
diff changeset
    50
  val cancel_group: group -> unit
28972
cb8a2c3e188f renamed type Future.T to future;
wenzelm
parents: 28647
diff changeset
    51
  val cancel: 'a future -> unit
28203
88f18387f1c9 shutdown: global join-and-shutdown operation;
wenzelm
parents: 28202
diff changeset
    52
  val shutdown: unit -> unit
28156
5205f7979b4f Functional threads as future values.
wenzelm
parents:
diff changeset
    53
end;
5205f7979b4f Functional threads as future values.
wenzelm
parents:
diff changeset
    54
5205f7979b4f Functional threads as future values.
wenzelm
parents:
diff changeset
    55
structure Future: FUTURE =
5205f7979b4f Functional threads as future values.
wenzelm
parents:
diff changeset
    56
struct
5205f7979b4f Functional threads as future values.
wenzelm
parents:
diff changeset
    57
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
    58
(** 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
    59
28645
605a3b1ef6ba added Future.enabled check;
wenzelm
parents: 28575
diff changeset
    60
fun enabled () =
29118
8f2481aa363d removed old scheduler;
wenzelm
parents: 28997
diff changeset
    61
  Multithreading.enabled () andalso
28645
605a3b1ef6ba added Future.enabled check;
wenzelm
parents: 28575
diff changeset
    62
    not (Multithreading.self_critical ());
605a3b1ef6ba added Future.enabled check;
wenzelm
parents: 28575
diff changeset
    63
605a3b1ef6ba added Future.enabled check;
wenzelm
parents: 28575
diff changeset
    64
28167
27e2ca41b58c more interrupt operations;
wenzelm
parents: 28166
diff changeset
    65
(* identifiers *)
27e2ca41b58c more interrupt operations;
wenzelm
parents: 28166
diff changeset
    66
29119
99941fd0cb0e renamed structure TaskQueue to Task_Queue;
wenzelm
parents: 29118
diff changeset
    67
type task = Task_Queue.task;
99941fd0cb0e renamed structure TaskQueue to Task_Queue;
wenzelm
parents: 29118
diff changeset
    68
type group = Task_Queue.group;
28167
27e2ca41b58c more interrupt operations;
wenzelm
parents: 28166
diff changeset
    69
32058
c76fd93b3b99 more abstract Future.is_worker;
wenzelm
parents: 32055
diff changeset
    70
local
c76fd93b3b99 more abstract Future.is_worker;
wenzelm
parents: 32055
diff changeset
    71
  val tag = Universal.tag () : (string * task * group) option Universal.tag;
c76fd93b3b99 more abstract Future.is_worker;
wenzelm
parents: 32055
diff changeset
    72
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
    73
  fun thread_data () = the_default NONE (Thread.getLocal tag);
32058
c76fd93b3b99 more abstract Future.is_worker;
wenzelm
parents: 32055
diff changeset
    74
  fun setmp_thread_data data f x =
c76fd93b3b99 more abstract Future.is_worker;
wenzelm
parents: 32055
diff changeset
    75
    Library.setmp_thread_data tag (thread_data ()) (SOME data) f x;
28167
27e2ca41b58c more interrupt operations;
wenzelm
parents: 28166
diff changeset
    76
end;
27e2ca41b58c more interrupt operations;
wenzelm
parents: 28166
diff changeset
    77
32058
c76fd93b3b99 more abstract Future.is_worker;
wenzelm
parents: 32055
diff changeset
    78
val is_worker = is_some o thread_data;
32102
81d03a29980c added worker_group;
wenzelm
parents: 32099
diff changeset
    79
val worker_group = Option.map #3 o thread_data;
32058
c76fd93b3b99 more abstract Future.is_worker;
wenzelm
parents: 32055
diff changeset
    80
28167
27e2ca41b58c more interrupt operations;
wenzelm
parents: 28166
diff changeset
    81
27e2ca41b58c more interrupt operations;
wenzelm
parents: 28166
diff changeset
    82
(* datatype future *)
27e2ca41b58c more interrupt operations;
wenzelm
parents: 28166
diff changeset
    83
28972
cb8a2c3e188f renamed type Future.T to future;
wenzelm
parents: 28647
diff changeset
    84
datatype 'a future = Future of
28167
27e2ca41b58c more interrupt operations;
wenzelm
parents: 28166
diff changeset
    85
 {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
    86
  group: group,
28167
27e2ca41b58c more interrupt operations;
wenzelm
parents: 28166
diff changeset
    87
  result: 'a Exn.result option ref};
27e2ca41b58c more interrupt operations;
wenzelm
parents: 28166
diff changeset
    88
27e2ca41b58c more interrupt operations;
wenzelm
parents: 28166
diff changeset
    89
fun task_of (Future {task, ...}) = task;
27e2ca41b58c more interrupt operations;
wenzelm
parents: 28166
diff changeset
    90
fun group_of (Future {group, ...}) = group;
27e2ca41b58c more interrupt operations;
wenzelm
parents: 28166
diff changeset
    91
28558
2a6297b4273c replaced str_of by general peek;
wenzelm
parents: 28548
diff changeset
    92
fun peek (Future {result, ...}) = ! result;
2a6297b4273c replaced str_of by general peek;
wenzelm
parents: 28548
diff changeset
    93
fun is_finished x = is_some (peek x);
28320
c6aef67f964d added is_finished;
wenzelm
parents: 28304
diff changeset
    94
28997
1b99dcae2156 added constant value;
wenzelm
parents: 28979
diff changeset
    95
fun value x = Future
29119
99941fd0cb0e renamed structure TaskQueue to Task_Queue;
wenzelm
parents: 29118
diff changeset
    96
 {task = Task_Queue.new_task 0,
32102
81d03a29980c added worker_group;
wenzelm
parents: 32099
diff changeset
    97
  group = Task_Queue.new_group NONE,
28997
1b99dcae2156 added constant value;
wenzelm
parents: 28979
diff changeset
    98
  result = ref (SOME (Exn.Result x))};
1b99dcae2156 added constant value;
wenzelm
parents: 28979
diff changeset
    99
28167
27e2ca41b58c more interrupt operations;
wenzelm
parents: 28166
diff changeset
   100
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
   101
8c0335bc9336 inherit group from running thread, or create a new one -- make it harder to re-use canceled groups;
wenzelm
parents: 28170
diff changeset
   102
(** 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
   103
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
(* 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
   105
29119
99941fd0cb0e renamed structure TaskQueue to Task_Queue;
wenzelm
parents: 29118
diff changeset
   106
val queue = ref Task_Queue.empty;
28468
7c80ab57f56d more tuning of tracing messages;
wenzelm
parents: 28464
diff changeset
   107
val next = ref 0;
28192
6d977729c8fa workers: explicit activity flag;
wenzelm
parents: 28191
diff changeset
   108
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
   109
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
   110
val excessive = ref 0;
29119
99941fd0cb0e renamed structure TaskQueue to Task_Queue;
wenzelm
parents: 29118
diff changeset
   111
val canceled = ref ([]: Task_Queue.group list);
28206
bcd48c6897d4 eliminated requests, use global state variables uniformly;
wenzelm
parents: 28203
diff changeset
   112
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
   113
8c0335bc9336 inherit group from running thread, or create a new one -- make it harder to re-use canceled groups;
wenzelm
parents: 28170
diff changeset
   114
8c0335bc9336 inherit group from running thread, or create a new one -- make it harder to re-use canceled groups;
wenzelm
parents: 28170
diff changeset
   115
(* synchronization *)
28156
5205f7979b4f Functional threads as future values.
wenzelm
parents:
diff changeset
   116
5205f7979b4f Functional threads as future values.
wenzelm
parents:
diff changeset
   117
local
5205f7979b4f Functional threads as future values.
wenzelm
parents:
diff changeset
   118
  val lock = Mutex.mutex ();
5205f7979b4f Functional threads as future values.
wenzelm
parents:
diff changeset
   119
  val cond = ConditionVar.conditionVar ();
5205f7979b4f Functional threads as future values.
wenzelm
parents:
diff changeset
   120
in
5205f7979b4f Functional threads as future values.
wenzelm
parents:
diff changeset
   121
28575
ed869f019642 SimpleThread.synchronized;
wenzelm
parents: 28558
diff changeset
   122
fun SYNCHRONIZED name = SimpleThread.synchronized name lock;
28156
5205f7979b4f Functional threads as future values.
wenzelm
parents:
diff changeset
   123
29119
99941fd0cb0e renamed structure TaskQueue to Task_Queue;
wenzelm
parents: 29118
diff changeset
   124
fun wait () = (*requires SYNCHRONIZED*)
28206
bcd48c6897d4 eliminated requests, use global state variables uniformly;
wenzelm
parents: 28203
diff changeset
   125
  ConditionVar.wait (cond, lock);
bcd48c6897d4 eliminated requests, use global state variables uniformly;
wenzelm
parents: 28203
diff changeset
   126
29119
99941fd0cb0e renamed structure TaskQueue to Task_Queue;
wenzelm
parents: 29118
diff changeset
   127
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
   128
  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
   129
43087721a66e moved task, thread_data, group, queue to task_queue.ML;
wenzelm
parents: 28163
diff changeset
   130
fun notify_all () = (*requires SYNCHRONIZED*)
43087721a66e moved task, thread_data, group, queue to task_queue.ML;
wenzelm
parents: 28163
diff changeset
   131
  ConditionVar.broadcast cond;
28156
5205f7979b4f Functional threads as future values.
wenzelm
parents:
diff changeset
   132
5205f7979b4f Functional threads as future values.
wenzelm
parents:
diff changeset
   133
end;
5205f7979b4f Functional threads as future values.
wenzelm
parents:
diff changeset
   134
5205f7979b4f Functional threads as future values.
wenzelm
parents:
diff changeset
   135
28382
a97fa342540d added release_results (formerly in par_list.ML);
wenzelm
parents: 28380
diff changeset
   136
(* worker activity *)
a97fa342540d added release_results (formerly in par_list.ML);
wenzelm
parents: 28380
diff changeset
   137
32095
ad4be204fdfe tuned tracing;
wenzelm
parents: 32058
diff changeset
   138
fun count_active ws =
ad4be204fdfe tuned tracing;
wenzelm
parents: 32058
diff changeset
   139
  fold (fn (_, active) => fn i => if active then i + 1 else i) ws 0;
ad4be204fdfe tuned tracing;
wenzelm
parents: 32058
diff changeset
   140
ad4be204fdfe tuned tracing;
wenzelm
parents: 32058
diff changeset
   141
fun trace_active () = Multithreading.tracing 1 (fn () =>
28382
a97fa342540d added release_results (formerly in par_list.ML);
wenzelm
parents: 28380
diff changeset
   142
  let
a97fa342540d added release_results (formerly in par_list.ML);
wenzelm
parents: 28380
diff changeset
   143
    val ws = ! workers;
a97fa342540d added release_results (formerly in par_list.ML);
wenzelm
parents: 28380
diff changeset
   144
    val m = string_of_int (length ws);
32095
ad4be204fdfe tuned tracing;
wenzelm
parents: 32058
diff changeset
   145
    val n = string_of_int (count_active ws);
ad4be204fdfe tuned tracing;
wenzelm
parents: 32058
diff changeset
   146
  in "SCHEDULE: " ^ m ^ " workers, " ^ n ^ " active" end);
28382
a97fa342540d added release_results (formerly in par_list.ML);
wenzelm
parents: 28380
diff changeset
   147
a97fa342540d added release_results (formerly in par_list.ML);
wenzelm
parents: 28380
diff changeset
   148
fun change_active active = (*requires SYNCHRONIZED*)
a97fa342540d added release_results (formerly in par_list.ML);
wenzelm
parents: 28380
diff changeset
   149
  change workers (AList.update Thread.equal (Thread.self (), active));
a97fa342540d added release_results (formerly in par_list.ML);
wenzelm
parents: 28380
diff changeset
   150
32095
ad4be204fdfe tuned tracing;
wenzelm
parents: 32058
diff changeset
   151
fun overloaded () =
ad4be204fdfe tuned tracing;
wenzelm
parents: 32058
diff changeset
   152
  count_active (! workers) > Multithreading.max_threads_value ();
ad4be204fdfe tuned tracing;
wenzelm
parents: 32058
diff changeset
   153
28382
a97fa342540d added release_results (formerly in par_list.ML);
wenzelm
parents: 28380
diff changeset
   154
32099
5382c93108db propagate exceptions within future groups;
wenzelm
parents: 32096
diff changeset
   155
(* execute future jobs *)
5382c93108db propagate exceptions within future groups;
wenzelm
parents: 32096
diff changeset
   156
5382c93108db propagate exceptions within future groups;
wenzelm
parents: 32096
diff changeset
   157
fun future_job group (e: unit -> 'a) =
5382c93108db propagate exceptions within future groups;
wenzelm
parents: 32096
diff changeset
   158
  let
5382c93108db propagate exceptions within future groups;
wenzelm
parents: 32096
diff changeset
   159
    val result = ref (NONE: 'a Exn.result option);
5382c93108db propagate exceptions within future groups;
wenzelm
parents: 32096
diff changeset
   160
    val job = Multithreading.with_attributes Multithreading.restricted_interrupts
5382c93108db propagate exceptions within future groups;
wenzelm
parents: 32096
diff changeset
   161
      (fn _ => fn ok =>
5382c93108db propagate exceptions within future groups;
wenzelm
parents: 32096
diff changeset
   162
        let
5382c93108db propagate exceptions within future groups;
wenzelm
parents: 32096
diff changeset
   163
          val res = if ok then Exn.capture e () else Exn.Exn Exn.Interrupt;
5382c93108db propagate exceptions within future groups;
wenzelm
parents: 32096
diff changeset
   164
          val _ = result := SOME res;
5382c93108db propagate exceptions within future groups;
wenzelm
parents: 32096
diff changeset
   165
          val _ = (case res of Exn.Exn exn => Task_Queue.cancel_group group exn | _ => ());
5382c93108db propagate exceptions within future groups;
wenzelm
parents: 32096
diff changeset
   166
          val res_ok = is_some (Exn.get_result res);
5382c93108db propagate exceptions within future groups;
wenzelm
parents: 32096
diff changeset
   167
        in res_ok end);
5382c93108db propagate exceptions within future groups;
wenzelm
parents: 32096
diff changeset
   168
  in (result, job) end;
28156
5205f7979b4f Functional threads as future values.
wenzelm
parents:
diff changeset
   169
29341
6bb007a0f9f2 more reactive scheduler: reduced loop timeout, propagate broadcast interrupt via TaskQueue.cancel_all;
wenzelm
parents: 29119
diff changeset
   170
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
   171
  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
   172
29366
1ffc8cbf39ec tuned map: reduced overhead due to bulk jobs;
wenzelm
parents: 29341
diff changeset
   173
fun execute name (task, group, jobs) =
28167
27e2ca41b58c more interrupt operations;
wenzelm
parents: 28166
diff changeset
   174
  let
28382
a97fa342540d added release_results (formerly in par_list.ML);
wenzelm
parents: 28380
diff changeset
   175
    val _ = trace_active ();
32102
81d03a29980c added worker_group;
wenzelm
parents: 32099
diff changeset
   176
    val valid = not (Task_Queue.is_canceled group);
32058
c76fd93b3b99 more abstract Future.is_worker;
wenzelm
parents: 32055
diff changeset
   177
    val ok = setmp_thread_data (name, task, group) (fn () =>
29384
a3c7e9ae9b71 more robust propagation of errors through bulk jobs;
wenzelm
parents: 29366
diff changeset
   178
      fold (fn job => fn ok => job valid andalso ok) jobs true) ();
28192
6d977729c8fa workers: explicit activity flag;
wenzelm
parents: 28191
diff changeset
   179
    val _ = SYNCHRONIZED "execute" (fn () =>
29119
99941fd0cb0e renamed structure TaskQueue to Task_Queue;
wenzelm
parents: 29118
diff changeset
   180
     (change queue (Task_Queue.finish task);
28186
6a8417f36837 cancel: check_scheduler;
wenzelm
parents: 28177
diff changeset
   181
      if ok then ()
29119
99941fd0cb0e renamed structure TaskQueue to Task_Queue;
wenzelm
parents: 29118
diff changeset
   182
      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
   183
      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
   184
      notify_all ()));
28167
27e2ca41b58c more interrupt operations;
wenzelm
parents: 28166
diff changeset
   185
  in () end;
27e2ca41b58c more interrupt operations;
wenzelm
parents: 28166
diff changeset
   186
27e2ca41b58c more interrupt operations;
wenzelm
parents: 28166
diff changeset
   187
27e2ca41b58c more interrupt operations;
wenzelm
parents: 28166
diff changeset
   188
(* worker threads *)
27e2ca41b58c more interrupt operations;
wenzelm
parents: 28166
diff changeset
   189
29119
99941fd0cb0e renamed structure TaskQueue to Task_Queue;
wenzelm
parents: 29118
diff changeset
   190
fun worker_wait () = (*requires SYNCHRONIZED*)
99941fd0cb0e renamed structure TaskQueue to Task_Queue;
wenzelm
parents: 29118
diff changeset
   191
  (change_active false; wait (); change_active true);
28162
55772e4e95e0 tuned check_cache;
wenzelm
parents: 28156
diff changeset
   192
29119
99941fd0cb0e renamed structure TaskQueue to Task_Queue;
wenzelm
parents: 29118
diff changeset
   193
fun worker_next () = (*requires SYNCHRONIZED*)
28167
27e2ca41b58c more interrupt operations;
wenzelm
parents: 28166
diff changeset
   194
  if ! excessive > 0 then
27e2ca41b58c more interrupt operations;
wenzelm
parents: 28166
diff changeset
   195
    (dec excessive;
28192
6d977729c8fa workers: explicit activity flag;
wenzelm
parents: 28191
diff changeset
   196
     change workers (filter_out (fn (thread, _) => Thread.equal (thread, Thread.self ())));
28203
88f18387f1c9 shutdown: global join-and-shutdown operation;
wenzelm
parents: 28202
diff changeset
   197
     notify_all ();
28167
27e2ca41b58c more interrupt operations;
wenzelm
parents: 28166
diff changeset
   198
     NONE)
32095
ad4be204fdfe tuned tracing;
wenzelm
parents: 32058
diff changeset
   199
  else if overloaded () then (worker_wait (); worker_next ())
28166
43087721a66e moved task, thread_data, group, queue to task_queue.ML;
wenzelm
parents: 28163
diff changeset
   200
  else
29119
99941fd0cb0e renamed structure TaskQueue to Task_Queue;
wenzelm
parents: 29118
diff changeset
   201
    (case change_result queue Task_Queue.dequeue of
99941fd0cb0e renamed structure TaskQueue to Task_Queue;
wenzelm
parents: 29118
diff changeset
   202
      NONE => (worker_wait (); worker_next ())
28166
43087721a66e moved task, thread_data, group, queue to task_queue.ML;
wenzelm
parents: 28163
diff changeset
   203
    | some => some);
28156
5205f7979b4f Functional threads as future values.
wenzelm
parents:
diff changeset
   204
28167
27e2ca41b58c more interrupt operations;
wenzelm
parents: 28166
diff changeset
   205
fun worker_loop name =
29119
99941fd0cb0e renamed structure TaskQueue to Task_Queue;
wenzelm
parents: 29118
diff changeset
   206
  (case SYNCHRONIZED name worker_next of
99941fd0cb0e renamed structure TaskQueue to Task_Queue;
wenzelm
parents: 29118
diff changeset
   207
    NONE => ()
28167
27e2ca41b58c more interrupt operations;
wenzelm
parents: 28166
diff changeset
   208
  | SOME work => (execute name work; worker_loop name));
28156
5205f7979b4f Functional threads as future values.
wenzelm
parents:
diff changeset
   209
28167
27e2ca41b58c more interrupt operations;
wenzelm
parents: 28166
diff changeset
   210
fun worker_start name = (*requires SYNCHRONIZED*)
28242
f978c8e75118 SimpleThread.fork;
wenzelm
parents: 28208
diff changeset
   211
  change workers (cons (SimpleThread.fork false (fn () => worker_loop name), true));
28156
5205f7979b4f Functional threads as future values.
wenzelm
parents:
diff changeset
   212
5205f7979b4f Functional threads as future values.
wenzelm
parents:
diff changeset
   213
5205f7979b4f Functional threads as future values.
wenzelm
parents:
diff changeset
   214
(* scheduler *)
5205f7979b4f Functional threads as future values.
wenzelm
parents:
diff changeset
   215
28206
bcd48c6897d4 eliminated requests, use global state variables uniformly;
wenzelm
parents: 28203
diff changeset
   216
fun scheduler_next () = (*requires SYNCHRONIZED*)
28156
5205f7979b4f Functional threads as future values.
wenzelm
parents:
diff changeset
   217
  let
32053
257eac3946e9 scheduler: tuned tracing (status);
wenzelm
parents: 31631
diff changeset
   218
    (*queue status*)
257eac3946e9 scheduler: tuned tracing (status);
wenzelm
parents: 31631
diff changeset
   219
    val _ = Multithreading.tracing 1 (fn () =>
257eac3946e9 scheduler: tuned tracing (status);
wenzelm
parents: 31631
diff changeset
   220
      let val {ready, pending, running} = Task_Queue.status (! queue) in
257eac3946e9 scheduler: tuned tracing (status);
wenzelm
parents: 31631
diff changeset
   221
        "SCHEDULE: " ^
257eac3946e9 scheduler: tuned tracing (status);
wenzelm
parents: 31631
diff changeset
   222
          string_of_int ready ^ " ready, " ^
257eac3946e9 scheduler: tuned tracing (status);
wenzelm
parents: 31631
diff changeset
   223
          string_of_int pending ^ " pending, " ^
257eac3946e9 scheduler: tuned tracing (status);
wenzelm
parents: 31631
diff changeset
   224
          string_of_int running ^ " running"
257eac3946e9 scheduler: tuned tracing (status);
wenzelm
parents: 31631
diff changeset
   225
      end);
257eac3946e9 scheduler: tuned tracing (status);
wenzelm
parents: 31631
diff changeset
   226
28206
bcd48c6897d4 eliminated requests, use global state variables uniformly;
wenzelm
parents: 28203
diff changeset
   227
    (*worker threads*)
32095
ad4be204fdfe tuned tracing;
wenzelm
parents: 32058
diff changeset
   228
    val ws = ! workers;
28191
9e5f556409c6 future: allow explicit group;
wenzelm
parents: 28186
diff changeset
   229
    val _ =
32095
ad4be204fdfe tuned tracing;
wenzelm
parents: 32058
diff changeset
   230
      if forall (Thread.isActive o #1) ws then ()
ad4be204fdfe tuned tracing;
wenzelm
parents: 32058
diff changeset
   231
      else
ad4be204fdfe tuned tracing;
wenzelm
parents: 32058
diff changeset
   232
        (case List.partition (Thread.isActive o #1) ws of
ad4be204fdfe tuned tracing;
wenzelm
parents: 32058
diff changeset
   233
          (_, []) => ()
ad4be204fdfe tuned tracing;
wenzelm
parents: 32058
diff changeset
   234
        | (active, inactive) =>
ad4be204fdfe tuned tracing;
wenzelm
parents: 32058
diff changeset
   235
            (workers := active; Multithreading.tracing 0 (fn () =>
ad4be204fdfe tuned tracing;
wenzelm
parents: 32058
diff changeset
   236
              "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
   237
    val _ = trace_active ();
28191
9e5f556409c6 future: allow explicit group;
wenzelm
parents: 28186
diff changeset
   238
28206
bcd48c6897d4 eliminated requests, use global state variables uniformly;
wenzelm
parents: 28203
diff changeset
   239
    val m = if ! do_shutdown then 0 else Multithreading.max_threads_value ();
32095
ad4be204fdfe tuned tracing;
wenzelm
parents: 32058
diff changeset
   240
    val mm = (m * 3) div 2;
ad4be204fdfe tuned tracing;
wenzelm
parents: 32058
diff changeset
   241
    val l = length ws;
ad4be204fdfe tuned tracing;
wenzelm
parents: 32058
diff changeset
   242
    val _ = excessive := l - mm;
28203
88f18387f1c9 shutdown: global join-and-shutdown operation;
wenzelm
parents: 28202
diff changeset
   243
    val _ =
32095
ad4be204fdfe tuned tracing;
wenzelm
parents: 32058
diff changeset
   244
      if mm > l then
ad4be204fdfe tuned tracing;
wenzelm
parents: 32058
diff changeset
   245
        funpow (mm - l) (fn () => worker_start ("worker " ^ string_of_int (inc next))) ()
28203
88f18387f1c9 shutdown: global join-and-shutdown operation;
wenzelm
parents: 28202
diff changeset
   246
      else ();
28206
bcd48c6897d4 eliminated requests, use global state variables uniformly;
wenzelm
parents: 28203
diff changeset
   247
bcd48c6897d4 eliminated requests, use global state variables uniformly;
wenzelm
parents: 28203
diff changeset
   248
    (*canceled groups*)
32095
ad4be204fdfe tuned tracing;
wenzelm
parents: 32058
diff changeset
   249
    val _ = change canceled (filter_out (Task_Queue.cancel (! queue)));
28206
bcd48c6897d4 eliminated requests, use global state variables uniformly;
wenzelm
parents: 28203
diff changeset
   250
bcd48c6897d4 eliminated requests, use global state variables uniformly;
wenzelm
parents: 28203
diff changeset
   251
    (*shutdown*)
32095
ad4be204fdfe tuned tracing;
wenzelm
parents: 32058
diff changeset
   252
    val continue = not (! do_shutdown andalso null ws);
28206
bcd48c6897d4 eliminated requests, use global state variables uniformly;
wenzelm
parents: 28203
diff changeset
   253
    val _ = if continue then () else scheduler := NONE;
28167
27e2ca41b58c more interrupt operations;
wenzelm
parents: 28166
diff changeset
   254
28206
bcd48c6897d4 eliminated requests, use global state variables uniformly;
wenzelm
parents: 28203
diff changeset
   255
    val _ = notify_all ();
30666
d6248d4508d5 future scheduler: reduced wait timeout if tasks need to be canceled -- to improve reactivity of interrupts;
wenzelm
parents: 30618
diff changeset
   256
    val _ = interruptible (fn () =>
d6248d4508d5 future scheduler: reduced wait timeout if tasks need to be canceled -- to improve reactivity of interrupts;
wenzelm
parents: 30618
diff changeset
   257
        wait_timeout (Time.fromMilliseconds (if null (! canceled) then 1000 else 50))) ()
29341
6bb007a0f9f2 more reactive scheduler: reduced loop timeout, propagate broadcast interrupt via TaskQueue.cancel_all;
wenzelm
parents: 29119
diff changeset
   258
      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
   259
  in continue end;
bcd48c6897d4 eliminated requests, use global state variables uniformly;
wenzelm
parents: 28203
diff changeset
   260
bcd48c6897d4 eliminated requests, use global state variables uniformly;
wenzelm
parents: 28203
diff changeset
   261
fun scheduler_loop () =
29119
99941fd0cb0e renamed structure TaskQueue to Task_Queue;
wenzelm
parents: 29118
diff changeset
   262
  while SYNCHRONIZED "scheduler" scheduler_next do ();
28156
5205f7979b4f Functional threads as future values.
wenzelm
parents:
diff changeset
   263
28203
88f18387f1c9 shutdown: global join-and-shutdown operation;
wenzelm
parents: 28202
diff changeset
   264
fun scheduler_active () = (*requires SYNCHRONIZED*)
88f18387f1c9 shutdown: global join-and-shutdown operation;
wenzelm
parents: 28202
diff changeset
   265
  (case ! scheduler of NONE => false | SOME thread => Thread.isActive thread);
88f18387f1c9 shutdown: global join-and-shutdown operation;
wenzelm
parents: 28202
diff changeset
   266
28464
dcc030b52583 tuned SYNCHRONIZED: outermost Exn.release;
wenzelm
parents: 28442
diff changeset
   267
fun scheduler_check name = SYNCHRONIZED name (fn () =>
28206
bcd48c6897d4 eliminated requests, use global state variables uniformly;
wenzelm
parents: 28203
diff changeset
   268
  if not (scheduler_active ()) then
29119
99941fd0cb0e renamed structure TaskQueue to Task_Queue;
wenzelm
parents: 29118
diff changeset
   269
    (do_shutdown := false; scheduler := SOME (SimpleThread.fork false scheduler_loop))
28206
bcd48c6897d4 eliminated requests, use global state variables uniformly;
wenzelm
parents: 28203
diff changeset
   270
  else if ! do_shutdown then error "Scheduler shutdown in progress"
bcd48c6897d4 eliminated requests, use global state variables uniformly;
wenzelm
parents: 28203
diff changeset
   271
  else ());
28156
5205f7979b4f Functional threads as future values.
wenzelm
parents:
diff changeset
   272
5205f7979b4f Functional threads as future values.
wenzelm
parents:
diff changeset
   273
29366
1ffc8cbf39ec tuned map: reduced overhead due to bulk jobs;
wenzelm
parents: 29341
diff changeset
   274
1ffc8cbf39ec tuned map: reduced overhead due to bulk jobs;
wenzelm
parents: 29341
diff changeset
   275
(** futures **)
28156
5205f7979b4f Functional threads as future values.
wenzelm
parents:
diff changeset
   276
29366
1ffc8cbf39ec tuned map: reduced overhead due to bulk jobs;
wenzelm
parents: 29341
diff changeset
   277
(* fork *)
1ffc8cbf39ec tuned map: reduced overhead due to bulk jobs;
wenzelm
parents: 29341
diff changeset
   278
1ffc8cbf39ec tuned map: reduced overhead due to bulk jobs;
wenzelm
parents: 29341
diff changeset
   279
fun fork_future opt_group deps pri e =
1ffc8cbf39ec tuned map: reduced overhead due to bulk jobs;
wenzelm
parents: 29341
diff changeset
   280
  let
1ffc8cbf39ec tuned map: reduced overhead due to bulk jobs;
wenzelm
parents: 29341
diff changeset
   281
    val _ = scheduler_check "future check";
1ffc8cbf39ec tuned map: reduced overhead due to bulk jobs;
wenzelm
parents: 29341
diff changeset
   282
32102
81d03a29980c added worker_group;
wenzelm
parents: 32099
diff changeset
   283
    val group =
81d03a29980c added worker_group;
wenzelm
parents: 32099
diff changeset
   284
      (case opt_group of
81d03a29980c added worker_group;
wenzelm
parents: 32099
diff changeset
   285
        SOME group => group
81d03a29980c added worker_group;
wenzelm
parents: 32099
diff changeset
   286
      | NONE => Task_Queue.new_group (worker_group ()));
29366
1ffc8cbf39ec tuned map: reduced overhead due to bulk jobs;
wenzelm
parents: 29341
diff changeset
   287
    val (result, job) = future_job group e;
28192
6d977729c8fa workers: explicit activity flag;
wenzelm
parents: 28191
diff changeset
   288
    val task = SYNCHRONIZED "future" (fn () =>
29366
1ffc8cbf39ec tuned map: reduced overhead due to bulk jobs;
wenzelm
parents: 29341
diff changeset
   289
      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
   290
  in Future {task = task, group = group, result = result} end;
28162
55772e4e95e0 tuned check_cache;
wenzelm
parents: 28156
diff changeset
   291
29366
1ffc8cbf39ec tuned map: reduced overhead due to bulk jobs;
wenzelm
parents: 29341
diff changeset
   292
fun fork e = fork_future NONE [] 0 e;
1ffc8cbf39ec tuned map: reduced overhead due to bulk jobs;
wenzelm
parents: 29341
diff changeset
   293
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
   294
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
   295
fun fork_pri pri e = fork_future NONE [] pri e;
28186
6a8417f36837 cancel: check_scheduler;
wenzelm
parents: 28177
diff changeset
   296
6a8417f36837 cancel: check_scheduler;
wenzelm
parents: 28177
diff changeset
   297
29366
1ffc8cbf39ec tuned map: reduced overhead due to bulk jobs;
wenzelm
parents: 29341
diff changeset
   298
(* join *)
1ffc8cbf39ec tuned map: reduced overhead due to bulk jobs;
wenzelm
parents: 29341
diff changeset
   299
29551
95e469919c3e join_results: when dependencies are resulved (but not finished yet),
wenzelm
parents: 29431
diff changeset
   300
local
95e469919c3e join_results: when dependencies are resulved (but not finished yet),
wenzelm
parents: 29431
diff changeset
   301
32099
5382c93108db propagate exceptions within future groups;
wenzelm
parents: 32096
diff changeset
   302
fun get_result x =
5382c93108db propagate exceptions within future groups;
wenzelm
parents: 32096
diff changeset
   303
  (case peek x of
32102
81d03a29980c added worker_group;
wenzelm
parents: 32099
diff changeset
   304
    NONE => Exn.Exn (SYS_ERROR "unfinished future")
81d03a29980c added worker_group;
wenzelm
parents: 32099
diff changeset
   305
  | SOME (Exn.Exn Exn.Interrupt) =>
81d03a29980c added worker_group;
wenzelm
parents: 32099
diff changeset
   306
      Exn.Exn (Exn.EXCEPTIONS (Exn.flatten_list (Task_Queue.group_status (group_of x))))
81d03a29980c added worker_group;
wenzelm
parents: 32099
diff changeset
   307
  | SOME res => res);
28186
6a8417f36837 cancel: check_scheduler;
wenzelm
parents: 28177
diff changeset
   308
32095
ad4be204fdfe tuned tracing;
wenzelm
parents: 32058
diff changeset
   309
fun join_next deps = (*requires SYNCHRONIZED*)
ad4be204fdfe tuned tracing;
wenzelm
parents: 32058
diff changeset
   310
  if overloaded () then (worker_wait (); join_next deps)
ad4be204fdfe tuned tracing;
wenzelm
parents: 32058
diff changeset
   311
  else change_result queue (Task_Queue.dequeue_towards deps);
ad4be204fdfe tuned tracing;
wenzelm
parents: 32058
diff changeset
   312
32055
6a46898aa805 recovered a version of dequeue_towards (cf. bb7b5a5942c7);
wenzelm
parents: 32053
diff changeset
   313
fun join_deps deps =
32095
ad4be204fdfe tuned tracing;
wenzelm
parents: 32058
diff changeset
   314
  (case SYNCHRONIZED "join" (fn () => join_next deps) of
29551
95e469919c3e join_results: when dependencies are resulved (but not finished yet),
wenzelm
parents: 29431
diff changeset
   315
    NONE => ()
32055
6a46898aa805 recovered a version of dequeue_towards (cf. bb7b5a5942c7);
wenzelm
parents: 32053
diff changeset
   316
  | SOME (work, deps') => (execute "join" work; join_deps deps'));
29551
95e469919c3e join_results: when dependencies are resulved (but not finished yet),
wenzelm
parents: 29431
diff changeset
   317
95e469919c3e join_results: when dependencies are resulved (but not finished yet),
wenzelm
parents: 29431
diff changeset
   318
in
95e469919c3e join_results: when dependencies are resulved (but not finished yet),
wenzelm
parents: 29431
diff changeset
   319
29366
1ffc8cbf39ec tuned map: reduced overhead due to bulk jobs;
wenzelm
parents: 29341
diff changeset
   320
fun join_results xs =
1ffc8cbf39ec tuned map: reduced overhead due to bulk jobs;
wenzelm
parents: 29341
diff changeset
   321
  if forall is_finished xs then map get_result xs
1ffc8cbf39ec tuned map: reduced overhead due to bulk jobs;
wenzelm
parents: 29341
diff changeset
   322
  else uninterruptible (fn _ => fn () =>
1ffc8cbf39ec tuned map: reduced overhead due to bulk jobs;
wenzelm
parents: 29341
diff changeset
   323
    let
1ffc8cbf39ec tuned map: reduced overhead due to bulk jobs;
wenzelm
parents: 29341
diff changeset
   324
      val _ = scheduler_check "join check";
1ffc8cbf39ec tuned map: reduced overhead due to bulk jobs;
wenzelm
parents: 29341
diff changeset
   325
      val _ = Multithreading.self_critical () andalso
1ffc8cbf39ec tuned map: reduced overhead due to bulk jobs;
wenzelm
parents: 29341
diff changeset
   326
        error "Cannot join future values within critical section";
32055
6a46898aa805 recovered a version of dequeue_towards (cf. bb7b5a5942c7);
wenzelm
parents: 32053
diff changeset
   327
32058
c76fd93b3b99 more abstract Future.is_worker;
wenzelm
parents: 32055
diff changeset
   328
      val worker = is_worker ();
32096
wenzelm
parents: 32095
diff changeset
   329
      val _ = if worker then join_deps (map task_of xs) else ();
wenzelm
parents: 32095
diff changeset
   330
32055
6a46898aa805 recovered a version of dequeue_towards (cf. bb7b5a5942c7);
wenzelm
parents: 32053
diff changeset
   331
      fun join_wait x =
6a46898aa805 recovered a version of dequeue_towards (cf. bb7b5a5942c7);
wenzelm
parents: 32053
diff changeset
   332
        if SYNCHRONIZED "join_wait" (fn () =>
32058
c76fd93b3b99 more abstract Future.is_worker;
wenzelm
parents: 32055
diff changeset
   333
          is_finished x orelse (if worker then worker_wait () else wait (); false))
32055
6a46898aa805 recovered a version of dequeue_towards (cf. bb7b5a5942c7);
wenzelm
parents: 32053
diff changeset
   334
        then () else join_wait x;
6a46898aa805 recovered a version of dequeue_towards (cf. bb7b5a5942c7);
wenzelm
parents: 32053
diff changeset
   335
      val _ = List.app join_wait xs;
6a46898aa805 recovered a version of dequeue_towards (cf. bb7b5a5942c7);
wenzelm
parents: 32053
diff changeset
   336
29366
1ffc8cbf39ec tuned map: reduced overhead due to bulk jobs;
wenzelm
parents: 29341
diff changeset
   337
    in map get_result xs end) ();
28186
6a8417f36837 cancel: check_scheduler;
wenzelm
parents: 28177
diff changeset
   338
29551
95e469919c3e join_results: when dependencies are resulved (but not finished yet),
wenzelm
parents: 29431
diff changeset
   339
end;
95e469919c3e join_results: when dependencies are resulved (but not finished yet),
wenzelm
parents: 29431
diff changeset
   340
28647
8068cdc84e7e join_results: allow CRITICAL join of finished futures;
wenzelm
parents: 28645
diff changeset
   341
fun join_result x = singleton join_results x;
8068cdc84e7e join_results: allow CRITICAL join of finished futures;
wenzelm
parents: 28645
diff changeset
   342
fun join x = Exn.release (join_result x);
28156
5205f7979b4f Functional threads as future values.
wenzelm
parents:
diff changeset
   343
29366
1ffc8cbf39ec tuned map: reduced overhead due to bulk jobs;
wenzelm
parents: 29341
diff changeset
   344
1ffc8cbf39ec tuned map: reduced overhead due to bulk jobs;
wenzelm
parents: 29341
diff changeset
   345
(* map *)
1ffc8cbf39ec tuned map: reduced overhead due to bulk jobs;
wenzelm
parents: 29341
diff changeset
   346
29384
a3c7e9ae9b71 more robust propagation of errors through bulk jobs;
wenzelm
parents: 29366
diff changeset
   347
fun map_future f x =
29366
1ffc8cbf39ec tuned map: reduced overhead due to bulk jobs;
wenzelm
parents: 29341
diff changeset
   348
  let
1ffc8cbf39ec tuned map: reduced overhead due to bulk jobs;
wenzelm
parents: 29341
diff changeset
   349
    val _ = scheduler_check "map_future check";
1ffc8cbf39ec tuned map: reduced overhead due to bulk jobs;
wenzelm
parents: 29341
diff changeset
   350
29384
a3c7e9ae9b71 more robust propagation of errors through bulk jobs;
wenzelm
parents: 29366
diff changeset
   351
    val task = task_of x;
32102
81d03a29980c added worker_group;
wenzelm
parents: 32099
diff changeset
   352
    val group = Task_Queue.new_group (SOME (group_of x));
29384
a3c7e9ae9b71 more robust propagation of errors through bulk jobs;
wenzelm
parents: 29366
diff changeset
   353
    val (result, job) = future_job group (fn () => f (join x));
a3c7e9ae9b71 more robust propagation of errors through bulk jobs;
wenzelm
parents: 29366
diff changeset
   354
29366
1ffc8cbf39ec tuned map: reduced overhead due to bulk jobs;
wenzelm
parents: 29341
diff changeset
   355
    val extended = SYNCHRONIZED "map_future" (fn () =>
1ffc8cbf39ec tuned map: reduced overhead due to bulk jobs;
wenzelm
parents: 29341
diff changeset
   356
      (case Task_Queue.extend task job (! queue) of
1ffc8cbf39ec tuned map: reduced overhead due to bulk jobs;
wenzelm
parents: 29341
diff changeset
   357
        SOME queue' => (queue := queue'; true)
1ffc8cbf39ec tuned map: reduced overhead due to bulk jobs;
wenzelm
parents: 29341
diff changeset
   358
      | NONE => false));
1ffc8cbf39ec tuned map: reduced overhead due to bulk jobs;
wenzelm
parents: 29341
diff changeset
   359
  in
29384
a3c7e9ae9b71 more robust propagation of errors through bulk jobs;
wenzelm
parents: 29366
diff changeset
   360
    if extended then Future {task = task, group = group, result = result}
32099
5382c93108db propagate exceptions within future groups;
wenzelm
parents: 32096
diff changeset
   361
    else fork_future (SOME group) [task] (Task_Queue.pri_of_task task) (fn () => f (join x))
29366
1ffc8cbf39ec tuned map: reduced overhead due to bulk jobs;
wenzelm
parents: 29341
diff changeset
   362
  end;
28979
3ce619d8d432 fork/map: no inheritance of group (structure is nested, not parallel);
wenzelm
parents: 28972
diff changeset
   363
28191
9e5f556409c6 future: allow explicit group;
wenzelm
parents: 28186
diff changeset
   364
29431
0ebe652bfd5a added cancel_group;
wenzelm
parents: 29384
diff changeset
   365
(* cancellation *)
28202
23cb9a974630 added focus, which indicates a particular collection of high-priority tasks;
wenzelm
parents: 28201
diff changeset
   366
30618
046f4f986fb5 restricted interrupts for tasks running as future worker thread -- attempt to prevent interrupt race conditions;
wenzelm
parents: 30612
diff changeset
   367
fun interruptible_task f x =
046f4f986fb5 restricted interrupts for tasks running as future worker thread -- attempt to prevent interrupt race conditions;
wenzelm
parents: 30612
diff changeset
   368
  if Multithreading.available then
046f4f986fb5 restricted interrupts for tasks running as future worker thread -- attempt to prevent interrupt race conditions;
wenzelm
parents: 30612
diff changeset
   369
    Multithreading.with_attributes
32058
c76fd93b3b99 more abstract Future.is_worker;
wenzelm
parents: 32055
diff changeset
   370
      (if is_worker ()
30618
046f4f986fb5 restricted interrupts for tasks running as future worker thread -- attempt to prevent interrupt race conditions;
wenzelm
parents: 30612
diff changeset
   371
       then Multithreading.restricted_interrupts
046f4f986fb5 restricted interrupts for tasks running as future worker thread -- attempt to prevent interrupt race conditions;
wenzelm
parents: 30612
diff changeset
   372
       else Multithreading.regular_interrupts)
046f4f986fb5 restricted interrupts for tasks running as future worker thread -- attempt to prevent interrupt race conditions;
wenzelm
parents: 30612
diff changeset
   373
      (fn _ => f) x
046f4f986fb5 restricted interrupts for tasks running as future worker thread -- attempt to prevent interrupt race conditions;
wenzelm
parents: 30612
diff changeset
   374
  else interruptible f x;
046f4f986fb5 restricted interrupts for tasks running as future worker thread -- attempt to prevent interrupt race conditions;
wenzelm
parents: 30612
diff changeset
   375
28202
23cb9a974630 added focus, which indicates a particular collection of high-priority tasks;
wenzelm
parents: 28201
diff changeset
   376
(*interrupt: permissive signal, may get ignored*)
28197
7053c539ecd8 added interrupt_task (external id);
wenzelm
parents: 28193
diff changeset
   377
fun interrupt_task id = SYNCHRONIZED "interrupt"
29119
99941fd0cb0e renamed structure TaskQueue to Task_Queue;
wenzelm
parents: 29118
diff changeset
   378
  (fn () => Task_Queue.interrupt_external (! queue) id);
28191
9e5f556409c6 future: allow explicit group;
wenzelm
parents: 28186
diff changeset
   379
28206
bcd48c6897d4 eliminated requests, use global state variables uniformly;
wenzelm
parents: 28203
diff changeset
   380
(*cancel: present and future group members will be interrupted eventually*)
29431
0ebe652bfd5a added cancel_group;
wenzelm
parents: 29384
diff changeset
   381
fun cancel_group group =
28464
dcc030b52583 tuned SYNCHRONIZED: outermost Exn.release;
wenzelm
parents: 28442
diff changeset
   382
 (scheduler_check "cancel check";
29431
0ebe652bfd5a added cancel_group;
wenzelm
parents: 29384
diff changeset
   383
  SYNCHRONIZED "cancel" (fn () => (do_cancel group; notify_all ())));
28206
bcd48c6897d4 eliminated requests, use global state variables uniformly;
wenzelm
parents: 28203
diff changeset
   384
29431
0ebe652bfd5a added cancel_group;
wenzelm
parents: 29384
diff changeset
   385
fun cancel x = cancel_group (group_of x);
28206
bcd48c6897d4 eliminated requests, use global state variables uniformly;
wenzelm
parents: 28203
diff changeset
   386
29366
1ffc8cbf39ec tuned map: reduced overhead due to bulk jobs;
wenzelm
parents: 29341
diff changeset
   387
1ffc8cbf39ec tuned map: reduced overhead due to bulk jobs;
wenzelm
parents: 29341
diff changeset
   388
(** global join and shutdown **)
1ffc8cbf39ec tuned map: reduced overhead due to bulk jobs;
wenzelm
parents: 29341
diff changeset
   389
28203
88f18387f1c9 shutdown: global join-and-shutdown operation;
wenzelm
parents: 28202
diff changeset
   390
fun shutdown () =
28276
fbc707811203 shutdown only if Multithreading.available;
wenzelm
parents: 28242
diff changeset
   391
  if Multithreading.available then
28464
dcc030b52583 tuned SYNCHRONIZED: outermost Exn.release;
wenzelm
parents: 28442
diff changeset
   392
   (scheduler_check "shutdown check";
28276
fbc707811203 shutdown only if Multithreading.available;
wenzelm
parents: 28242
diff changeset
   393
    SYNCHRONIZED "shutdown" (fn () =>
29119
99941fd0cb0e renamed structure TaskQueue to Task_Queue;
wenzelm
parents: 29118
diff changeset
   394
     (while not (scheduler_active ()) do wait ();
99941fd0cb0e renamed structure TaskQueue to Task_Queue;
wenzelm
parents: 29118
diff changeset
   395
      while not (Task_Queue.is_empty (! queue)) do wait ();
28276
fbc707811203 shutdown only if Multithreading.available;
wenzelm
parents: 28242
diff changeset
   396
      do_shutdown := true;
fbc707811203 shutdown only if Multithreading.available;
wenzelm
parents: 28242
diff changeset
   397
      notify_all ();
29119
99941fd0cb0e renamed structure TaskQueue to Task_Queue;
wenzelm
parents: 29118
diff changeset
   398
      while not (null (! workers)) do wait ();
99941fd0cb0e renamed structure TaskQueue to Task_Queue;
wenzelm
parents: 29118
diff changeset
   399
      while scheduler_active () do wait ();
28470
409fedeece30 added simple heartbeat thread;
wenzelm
parents: 28468
diff changeset
   400
      OS.Process.sleep (Time.fromMilliseconds 300))))
28276
fbc707811203 shutdown only if Multithreading.available;
wenzelm
parents: 28242
diff changeset
   401
  else ();
28203
88f18387f1c9 shutdown: global join-and-shutdown operation;
wenzelm
parents: 28202
diff changeset
   402
29366
1ffc8cbf39ec tuned map: reduced overhead due to bulk jobs;
wenzelm
parents: 29341
diff changeset
   403
1ffc8cbf39ec tuned map: reduced overhead due to bulk jobs;
wenzelm
parents: 29341
diff changeset
   404
(*final declarations of this structure!*)
1ffc8cbf39ec tuned map: reduced overhead due to bulk jobs;
wenzelm
parents: 29341
diff changeset
   405
val map = map_future;
1ffc8cbf39ec tuned map: reduced overhead due to bulk jobs;
wenzelm
parents: 29341
diff changeset
   406
28156
5205f7979b4f Functional threads as future values.
wenzelm
parents:
diff changeset
   407
end;
28972
cb8a2c3e188f renamed type Future.T to future;
wenzelm
parents: 28647
diff changeset
   408
cb8a2c3e188f renamed type Future.T to future;
wenzelm
parents: 28647
diff changeset
   409
type 'a future = 'a Future.future;
cb8a2c3e188f renamed type Future.T to future;
wenzelm
parents: 28647
diff changeset
   410