src/Pure/Concurrent/future.ML
author wenzelm
Wed, 04 Nov 2009 11:37:06 +0100
changeset 33409 0a1c0c1209ec
parent 33408 a69ddd7dce95
child 33410 e351f4c1f18c
permissions -rw-r--r--
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
32246
d4f7934e9555 misc tuning;
wenzelm
parents: 32230
diff changeset
     4
Future values, see also
d4f7934e9555 misc tuning;
wenzelm
parents: 32230
diff changeset
     5
http://www4.in.tum.de/~wenzelm/papers/parallel-isabelle.pdf
28201
7ae5cdb7b122 some general notes on future values;
wenzelm
parents: 28197
diff changeset
     6
7ae5cdb7b122 some general notes on future values;
wenzelm
parents: 28197
diff changeset
     7
Notes:
7ae5cdb7b122 some general notes on future values;
wenzelm
parents: 28197
diff changeset
     8
7ae5cdb7b122 some general notes on future values;
wenzelm
parents: 28197
diff changeset
     9
  * Futures are similar to delayed evaluation, i.e. delay/force is
7ae5cdb7b122 some general notes on future values;
wenzelm
parents: 28197
diff changeset
    10
    generalized to fork/join (and variants).  The idea is to model
7ae5cdb7b122 some general notes on future values;
wenzelm
parents: 28197
diff changeset
    11
    parallel value-oriented computations, but *not* communicating
7ae5cdb7b122 some general notes on future values;
wenzelm
parents: 28197
diff changeset
    12
    processes.
7ae5cdb7b122 some general notes on future values;
wenzelm
parents: 28197
diff changeset
    13
7ae5cdb7b122 some general notes on future values;
wenzelm
parents: 28197
diff changeset
    14
  * Futures are grouped; failure of one group member causes the whole
32220
wenzelm
parents: 32219
diff changeset
    15
    group to be interrupted eventually.  Groups are block-structured.
28201
7ae5cdb7b122 some general notes on future values;
wenzelm
parents: 28197
diff changeset
    16
7ae5cdb7b122 some general notes on future values;
wenzelm
parents: 28197
diff changeset
    17
  * Forked futures are evaluated spontaneously by a farm of worker
7ae5cdb7b122 some general notes on future values;
wenzelm
parents: 28197
diff changeset
    18
    threads in the background; join resynchronizes the computation and
7ae5cdb7b122 some general notes on future values;
wenzelm
parents: 28197
diff changeset
    19
    delivers results (values or exceptions).
7ae5cdb7b122 some general notes on future values;
wenzelm
parents: 28197
diff changeset
    20
7ae5cdb7b122 some general notes on future values;
wenzelm
parents: 28197
diff changeset
    21
  * The pool of worker threads is limited, usually in correlation with
7ae5cdb7b122 some general notes on future values;
wenzelm
parents: 28197
diff changeset
    22
    the number of physical cores on the machine.  Note that allocation
7ae5cdb7b122 some general notes on future values;
wenzelm
parents: 28197
diff changeset
    23
    of runtime resources is distorted either if workers yield CPU time
7ae5cdb7b122 some general notes on future values;
wenzelm
parents: 28197
diff changeset
    24
    (e.g. via system sleep or wait operations), or if non-worker
7ae5cdb7b122 some general notes on future values;
wenzelm
parents: 28197
diff changeset
    25
    threads contend for significant runtime resources independently.
28156
5205f7979b4f Functional threads as future values.
wenzelm
parents:
diff changeset
    26
*)
5205f7979b4f Functional threads as future values.
wenzelm
parents:
diff changeset
    27
5205f7979b4f Functional threads as future values.
wenzelm
parents:
diff changeset
    28
signature FUTURE =
5205f7979b4f Functional threads as future values.
wenzelm
parents:
diff changeset
    29
sig
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
32814
81897d30b97f added Task_Queue.depend (again) -- light-weight version for transitive graph;
wenzelm
parents: 32738
diff changeset
    33
  val worker_task: unit -> Task_Queue.task option
32102
81d03a29980c added worker_group;
wenzelm
parents: 32099
diff changeset
    34
  val worker_group: unit -> Task_Queue.group option
28972
cb8a2c3e188f renamed type Future.T to future;
wenzelm
parents: 28647
diff changeset
    35
  type 'a future
cb8a2c3e188f renamed type Future.T to future;
wenzelm
parents: 28647
diff changeset
    36
  val task_of: 'a future -> task
cb8a2c3e188f renamed type Future.T to future;
wenzelm
parents: 28647
diff changeset
    37
  val group_of: 'a future -> group
cb8a2c3e188f renamed type Future.T to future;
wenzelm
parents: 28647
diff changeset
    38
  val peek: 'a future -> 'a Exn.result option
cb8a2c3e188f renamed type Future.T to future;
wenzelm
parents: 28647
diff changeset
    39
  val is_finished: 'a future -> bool
28997
1b99dcae2156 added constant value;
wenzelm
parents: 28979
diff changeset
    40
  val value: '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
32724
aaeeb0ba2035 added fork_deps_pri;
wenzelm
parents: 32644
diff changeset
    42
  val fork_deps_pri: 'b future list -> int -> (unit -> 'a) -> 'a future
28979
3ce619d8d432 fork/map: no inheritance of group (structure is nested, not parallel);
wenzelm
parents: 28972
diff changeset
    43
  val fork_deps: 'b future list -> (unit -> 'a) -> 'a future
29119
99941fd0cb0e renamed structure TaskQueue to Task_Queue;
wenzelm
parents: 29118
diff changeset
    44
  val fork_pri: int -> (unit -> 'a) -> 'a future
32724
aaeeb0ba2035 added fork_deps_pri;
wenzelm
parents: 32644
diff changeset
    45
  val fork: (unit -> 'a) -> 'a future
28972
cb8a2c3e188f renamed type Future.T to future;
wenzelm
parents: 28647
diff changeset
    46
  val join_results: 'a future list -> 'a Exn.result list
cb8a2c3e188f renamed type Future.T to future;
wenzelm
parents: 28647
diff changeset
    47
  val join_result: 'a future -> 'a Exn.result
cb8a2c3e188f renamed type Future.T to future;
wenzelm
parents: 28647
diff changeset
    48
  val join: 'a future -> 'a
cb8a2c3e188f renamed type Future.T to future;
wenzelm
parents: 28647
diff changeset
    49
  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
    50
  val interruptible_task: ('a -> 'b) -> 'a -> 'b
29431
0ebe652bfd5a added cancel_group;
wenzelm
parents: 29384
diff changeset
    51
  val cancel_group: group -> unit
28972
cb8a2c3e188f renamed type Future.T to future;
wenzelm
parents: 28647
diff changeset
    52
  val cancel: 'a future -> unit
28203
88f18387f1c9 shutdown: global join-and-shutdown operation;
wenzelm
parents: 28202
diff changeset
    53
  val shutdown: unit -> unit
28156
5205f7979b4f Functional threads as future values.
wenzelm
parents:
diff changeset
    54
end;
5205f7979b4f Functional threads as future values.
wenzelm
parents:
diff changeset
    55
5205f7979b4f Functional threads as future values.
wenzelm
parents:
diff changeset
    56
structure Future: FUTURE =
5205f7979b4f Functional threads as future values.
wenzelm
parents:
diff changeset
    57
struct
5205f7979b4f Functional threads as future values.
wenzelm
parents:
diff changeset
    58
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
    59
(** 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
    60
28167
27e2ca41b58c more interrupt operations;
wenzelm
parents: 28166
diff changeset
    61
(* identifiers *)
27e2ca41b58c more interrupt operations;
wenzelm
parents: 28166
diff changeset
    62
29119
99941fd0cb0e renamed structure TaskQueue to Task_Queue;
wenzelm
parents: 29118
diff changeset
    63
type task = Task_Queue.task;
99941fd0cb0e renamed structure TaskQueue to Task_Queue;
wenzelm
parents: 29118
diff changeset
    64
type group = Task_Queue.group;
28167
27e2ca41b58c more interrupt operations;
wenzelm
parents: 28166
diff changeset
    65
32058
c76fd93b3b99 more abstract Future.is_worker;
wenzelm
parents: 32055
diff changeset
    66
local
33408
a69ddd7dce95 tuned thread data;
wenzelm
parents: 33407
diff changeset
    67
  val tag = Universal.tag () : (task * group) option Universal.tag;
32058
c76fd93b3b99 more abstract Future.is_worker;
wenzelm
parents: 32055
diff changeset
    68
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
    69
  fun thread_data () = the_default NONE (Thread.getLocal tag);
32058
c76fd93b3b99 more abstract Future.is_worker;
wenzelm
parents: 32055
diff changeset
    70
  fun setmp_thread_data data f x =
c76fd93b3b99 more abstract Future.is_worker;
wenzelm
parents: 32055
diff changeset
    71
    Library.setmp_thread_data tag (thread_data ()) (SOME data) f x;
28167
27e2ca41b58c more interrupt operations;
wenzelm
parents: 28166
diff changeset
    72
end;
27e2ca41b58c more interrupt operations;
wenzelm
parents: 28166
diff changeset
    73
32058
c76fd93b3b99 more abstract Future.is_worker;
wenzelm
parents: 32055
diff changeset
    74
val is_worker = is_some o thread_data;
33408
a69ddd7dce95 tuned thread data;
wenzelm
parents: 33407
diff changeset
    75
val worker_task = Option.map #1 o thread_data;
a69ddd7dce95 tuned thread data;
wenzelm
parents: 33407
diff changeset
    76
val worker_group = Option.map #2 o thread_data;
32058
c76fd93b3b99 more abstract Future.is_worker;
wenzelm
parents: 32055
diff changeset
    77
28167
27e2ca41b58c more interrupt operations;
wenzelm
parents: 28166
diff changeset
    78
27e2ca41b58c more interrupt operations;
wenzelm
parents: 28166
diff changeset
    79
(* datatype future *)
27e2ca41b58c more interrupt operations;
wenzelm
parents: 28166
diff changeset
    80
28972
cb8a2c3e188f renamed type Future.T to future;
wenzelm
parents: 28647
diff changeset
    81
datatype 'a future = Future of
28167
27e2ca41b58c more interrupt operations;
wenzelm
parents: 28166
diff changeset
    82
 {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
    83
  group: group,
32253
d9def420c84e future result: Synchronized.var;
wenzelm
parents: 32249
diff changeset
    84
  result: 'a Exn.result option Synchronized.var};
28167
27e2ca41b58c more interrupt operations;
wenzelm
parents: 28166
diff changeset
    85
27e2ca41b58c more interrupt operations;
wenzelm
parents: 28166
diff changeset
    86
fun task_of (Future {task, ...}) = task;
27e2ca41b58c more interrupt operations;
wenzelm
parents: 28166
diff changeset
    87
fun group_of (Future {group, ...}) = group;
32253
d9def420c84e future result: Synchronized.var;
wenzelm
parents: 32249
diff changeset
    88
fun result_of (Future {result, ...}) = result;
28167
27e2ca41b58c more interrupt operations;
wenzelm
parents: 28166
diff changeset
    89
32592
e29c0b7dcf66 Synchronized.value does not require locking, since assigments are atomic;
wenzelm
parents: 32420
diff changeset
    90
fun peek x = Synchronized.value (result_of x);
28558
2a6297b4273c replaced str_of by general peek;
wenzelm
parents: 28548
diff changeset
    91
fun is_finished x = is_some (peek x);
28320
c6aef67f964d added is_finished;
wenzelm
parents: 28304
diff changeset
    92
28997
1b99dcae2156 added constant value;
wenzelm
parents: 28979
diff changeset
    93
fun value x = Future
29119
99941fd0cb0e renamed structure TaskQueue to Task_Queue;
wenzelm
parents: 29118
diff changeset
    94
 {task = Task_Queue.new_task 0,
32102
81d03a29980c added worker_group;
wenzelm
parents: 32099
diff changeset
    95
  group = Task_Queue.new_group NONE,
32253
d9def420c84e future result: Synchronized.var;
wenzelm
parents: 32249
diff changeset
    96
  result = Synchronized.var "future" (SOME (Exn.Result x))};
28997
1b99dcae2156 added constant value;
wenzelm
parents: 28979
diff changeset
    97
28167
27e2ca41b58c more interrupt operations;
wenzelm
parents: 28166
diff changeset
    98
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
    99
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
(** 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
   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
(* 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
   103
32738
15bb09ca0378 explicit indication of Unsynchronized.ref;
wenzelm
parents: 32724
diff changeset
   104
val queue = Unsynchronized.ref Task_Queue.empty;
15bb09ca0378 explicit indication of Unsynchronized.ref;
wenzelm
parents: 32724
diff changeset
   105
val next = Unsynchronized.ref 0;
33406
1ddcb8472bd2 slightly leaner and more direct control of worker activity etc.;
wenzelm
parents: 33061
diff changeset
   106
val max_workers = Unsynchronized.ref 0;
1ddcb8472bd2 slightly leaner and more direct control of worker activity etc.;
wenzelm
parents: 33061
diff changeset
   107
val max_active = Unsynchronized.ref 0;
1ddcb8472bd2 slightly leaner and more direct control of worker activity etc.;
wenzelm
parents: 33061
diff changeset
   108
val workers = Unsynchronized.ref ([]: (Thread.thread * bool Unsynchronized.ref) list);
32738
15bb09ca0378 explicit indication of Unsynchronized.ref;
wenzelm
parents: 32724
diff changeset
   109
val scheduler = Unsynchronized.ref (NONE: Thread.thread option);
15bb09ca0378 explicit indication of Unsynchronized.ref;
wenzelm
parents: 32724
diff changeset
   110
val canceled = Unsynchronized.ref ([]: Task_Queue.group list);
15bb09ca0378 explicit indication of Unsynchronized.ref;
wenzelm
parents: 32724
diff changeset
   111
val do_shutdown = Unsynchronized.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
   112
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
(* synchronization *)
28156
5205f7979b4f Functional threads as future values.
wenzelm
parents:
diff changeset
   115
32219
9a2566d1fdbd more specific conditions: scheduler_event, work_available, work_finished -- considereably reduces overhead with many threads;
wenzelm
parents: 32186
diff changeset
   116
val scheduler_event = ConditionVar.conditionVar ();
9a2566d1fdbd more specific conditions: scheduler_event, work_available, work_finished -- considereably reduces overhead with many threads;
wenzelm
parents: 32186
diff changeset
   117
val work_available = ConditionVar.conditionVar ();
9a2566d1fdbd more specific conditions: scheduler_event, work_available, work_finished -- considereably reduces overhead with many threads;
wenzelm
parents: 32186
diff changeset
   118
val work_finished = ConditionVar.conditionVar ();
9a2566d1fdbd more specific conditions: scheduler_event, work_available, work_finished -- considereably reduces overhead with many threads;
wenzelm
parents: 32186
diff changeset
   119
28156
5205f7979b4f Functional threads as future values.
wenzelm
parents:
diff changeset
   120
local
5205f7979b4f Functional threads as future values.
wenzelm
parents:
diff changeset
   121
  val lock = Mutex.mutex ();
5205f7979b4f Functional threads as future values.
wenzelm
parents:
diff changeset
   122
in
5205f7979b4f Functional threads as future values.
wenzelm
parents:
diff changeset
   123
28575
ed869f019642 SimpleThread.synchronized;
wenzelm
parents: 28558
diff changeset
   124
fun SYNCHRONIZED name = SimpleThread.synchronized name lock;
28156
5205f7979b4f Functional threads as future values.
wenzelm
parents:
diff changeset
   125
32219
9a2566d1fdbd more specific conditions: scheduler_event, work_available, work_finished -- considereably reduces overhead with many threads;
wenzelm
parents: 32186
diff changeset
   126
fun wait cond = (*requires SYNCHRONIZED*)
32295
400cc493d466 renamed Multithreading.regular_interrupts to Multithreading.public_interrupts;
wenzelm
parents: 32293
diff changeset
   127
  Multithreading.sync_wait NONE NONE cond lock;
28206
bcd48c6897d4 eliminated requests, use global state variables uniformly;
wenzelm
parents: 28203
diff changeset
   128
32295
400cc493d466 renamed Multithreading.regular_interrupts to Multithreading.public_interrupts;
wenzelm
parents: 32293
diff changeset
   129
fun wait_timeout timeout cond = (*requires SYNCHRONIZED*)
400cc493d466 renamed Multithreading.regular_interrupts to Multithreading.public_interrupts;
wenzelm
parents: 32293
diff changeset
   130
  Multithreading.sync_wait NONE (SOME (Time.+ (Time.now (), timeout))) cond lock;
28166
43087721a66e moved task, thread_data, group, queue to task_queue.ML;
wenzelm
parents: 28163
diff changeset
   131
32219
9a2566d1fdbd more specific conditions: scheduler_event, work_available, work_finished -- considereably reduces overhead with many threads;
wenzelm
parents: 32186
diff changeset
   132
fun signal cond = (*requires SYNCHRONIZED*)
9a2566d1fdbd more specific conditions: scheduler_event, work_available, work_finished -- considereably reduces overhead with many threads;
wenzelm
parents: 32186
diff changeset
   133
  ConditionVar.signal cond;
9a2566d1fdbd more specific conditions: scheduler_event, work_available, work_finished -- considereably reduces overhead with many threads;
wenzelm
parents: 32186
diff changeset
   134
9a2566d1fdbd more specific conditions: scheduler_event, work_available, work_finished -- considereably reduces overhead with many threads;
wenzelm
parents: 32186
diff changeset
   135
fun broadcast cond = (*requires SYNCHRONIZED*)
28166
43087721a66e moved task, thread_data, group, queue to task_queue.ML;
wenzelm
parents: 28163
diff changeset
   136
  ConditionVar.broadcast cond;
28156
5205f7979b4f Functional threads as future values.
wenzelm
parents:
diff changeset
   137
32248
0241916a5f06 more precise treatment of scheduler_event: continous pulse (50ms) instead of flooding, which was burning many CPU cycles in spare threads;
wenzelm
parents: 32247
diff changeset
   138
fun broadcast_work () = (*requires SYNCHRONIZED*)
0241916a5f06 more precise treatment of scheduler_event: continous pulse (50ms) instead of flooding, which was burning many CPU cycles in spare threads;
wenzelm
parents: 32247
diff changeset
   139
 (ConditionVar.broadcast work_available;
32225
d5d6f47fb018 cancel: improved reactivity due to more careful broadcasting;
wenzelm
parents: 32224
diff changeset
   140
  ConditionVar.broadcast work_finished);
d5d6f47fb018 cancel: improved reactivity due to more careful broadcasting;
wenzelm
parents: 32224
diff changeset
   141
28156
5205f7979b4f Functional threads as future values.
wenzelm
parents:
diff changeset
   142
end;
5205f7979b4f Functional threads as future values.
wenzelm
parents:
diff changeset
   143
5205f7979b4f Functional threads as future values.
wenzelm
parents:
diff changeset
   144
32099
5382c93108db propagate exceptions within future groups;
wenzelm
parents: 32096
diff changeset
   145
(* execute future jobs *)
5382c93108db propagate exceptions within future groups;
wenzelm
parents: 32096
diff changeset
   146
5382c93108db propagate exceptions within future groups;
wenzelm
parents: 32096
diff changeset
   147
fun future_job group (e: unit -> 'a) =
5382c93108db propagate exceptions within future groups;
wenzelm
parents: 32096
diff changeset
   148
  let
32253
d9def420c84e future result: Synchronized.var;
wenzelm
parents: 32249
diff changeset
   149
    val result = Synchronized.var "future" (NONE: 'a Exn.result option);
32107
47d0da617fcc future_job: tight scope for interrupts, to prevent shooting ourselves in the foot via cancel_group;
wenzelm
parents: 32102
diff changeset
   150
    fun job ok =
47d0da617fcc future_job: tight scope for interrupts, to prevent shooting ourselves in the foot via cancel_group;
wenzelm
parents: 32102
diff changeset
   151
      let
47d0da617fcc future_job: tight scope for interrupts, to prevent shooting ourselves in the foot via cancel_group;
wenzelm
parents: 32102
diff changeset
   152
        val res =
47d0da617fcc future_job: tight scope for interrupts, to prevent shooting ourselves in the foot via cancel_group;
wenzelm
parents: 32102
diff changeset
   153
          if ok then
32230
9f6461b1c9cc interruptible: Thread.testInterrupt before changing thread attributes;
wenzelm
parents: 32229
diff changeset
   154
            Exn.capture (fn () =>
32295
400cc493d466 renamed Multithreading.regular_interrupts to Multithreading.public_interrupts;
wenzelm
parents: 32293
diff changeset
   155
              Multithreading.with_attributes Multithreading.private_interrupts (fn _ => e ())) ()
32107
47d0da617fcc future_job: tight scope for interrupts, to prevent shooting ourselves in the foot via cancel_group;
wenzelm
parents: 32102
diff changeset
   156
          else Exn.Exn Exn.Interrupt;
33061
e3e61133e0fc use Synchronized.assign to achieve actual immutable results;
wenzelm
parents: 32814
diff changeset
   157
        val _ = Synchronized.assign result (K (SOME res));
32107
47d0da617fcc future_job: tight scope for interrupts, to prevent shooting ourselves in the foot via cancel_group;
wenzelm
parents: 32102
diff changeset
   158
      in
47d0da617fcc future_job: tight scope for interrupts, to prevent shooting ourselves in the foot via cancel_group;
wenzelm
parents: 32102
diff changeset
   159
        (case res of
47d0da617fcc future_job: tight scope for interrupts, to prevent shooting ourselves in the foot via cancel_group;
wenzelm
parents: 32102
diff changeset
   160
          Exn.Exn exn => (Task_Queue.cancel_group group exn; false)
47d0da617fcc future_job: tight scope for interrupts, to prevent shooting ourselves in the foot via cancel_group;
wenzelm
parents: 32102
diff changeset
   161
        | Exn.Result _ => true)
47d0da617fcc future_job: tight scope for interrupts, to prevent shooting ourselves in the foot via cancel_group;
wenzelm
parents: 32102
diff changeset
   162
      end;
32099
5382c93108db propagate exceptions within future groups;
wenzelm
parents: 32096
diff changeset
   163
  in (result, job) end;
28156
5205f7979b4f Functional threads as future values.
wenzelm
parents:
diff changeset
   164
29341
6bb007a0f9f2 more reactive scheduler: reduced loop timeout, propagate broadcast interrupt via TaskQueue.cancel_all;
wenzelm
parents: 29119
diff changeset
   165
fun do_cancel group = (*requires SYNCHRONIZED*)
32738
15bb09ca0378 explicit indication of Unsynchronized.ref;
wenzelm
parents: 32724
diff changeset
   166
 (Unsynchronized.change canceled (insert Task_Queue.eq_group group);
15bb09ca0378 explicit indication of Unsynchronized.ref;
wenzelm
parents: 32724
diff changeset
   167
  broadcast scheduler_event);
29341
6bb007a0f9f2 more reactive scheduler: reduced loop timeout, propagate broadcast interrupt via TaskQueue.cancel_all;
wenzelm
parents: 29119
diff changeset
   168
33408
a69ddd7dce95 tuned thread data;
wenzelm
parents: 33407
diff changeset
   169
fun execute (task, group, jobs) =
28167
27e2ca41b58c more interrupt operations;
wenzelm
parents: 28166
diff changeset
   170
  let
32102
81d03a29980c added worker_group;
wenzelm
parents: 32099
diff changeset
   171
    val valid = not (Task_Queue.is_canceled group);
33408
a69ddd7dce95 tuned thread data;
wenzelm
parents: 33407
diff changeset
   172
    val ok = setmp_thread_data (task, group) (fn () =>
29384
a3c7e9ae9b71 more robust propagation of errors through bulk jobs;
wenzelm
parents: 29366
diff changeset
   173
      fold (fn job => fn ok => job valid andalso ok) jobs true) ();
32246
d4f7934e9555 misc tuning;
wenzelm
parents: 32230
diff changeset
   174
    val _ = SYNCHRONIZED "finish" (fn () =>
32219
9a2566d1fdbd more specific conditions: scheduler_event, work_available, work_finished -- considereably reduces overhead with many threads;
wenzelm
parents: 32186
diff changeset
   175
      let
32738
15bb09ca0378 explicit indication of Unsynchronized.ref;
wenzelm
parents: 32724
diff changeset
   176
        val maximal = Unsynchronized.change_result queue (Task_Queue.finish task);
32219
9a2566d1fdbd more specific conditions: scheduler_event, work_available, work_finished -- considereably reduces overhead with many threads;
wenzelm
parents: 32186
diff changeset
   177
        val _ =
9a2566d1fdbd more specific conditions: scheduler_event, work_available, work_finished -- considereably reduces overhead with many threads;
wenzelm
parents: 32186
diff changeset
   178
          if ok then ()
9a2566d1fdbd more specific conditions: scheduler_event, work_available, work_finished -- considereably reduces overhead with many threads;
wenzelm
parents: 32186
diff changeset
   179
          else if Task_Queue.cancel (! queue) group then ()
9a2566d1fdbd more specific conditions: scheduler_event, work_available, work_finished -- considereably reduces overhead with many threads;
wenzelm
parents: 32186
diff changeset
   180
          else do_cancel group;
9a2566d1fdbd more specific conditions: scheduler_event, work_available, work_finished -- considereably reduces overhead with many threads;
wenzelm
parents: 32186
diff changeset
   181
        val _ = broadcast work_finished;
9a2566d1fdbd more specific conditions: scheduler_event, work_available, work_finished -- considereably reduces overhead with many threads;
wenzelm
parents: 32186
diff changeset
   182
        val _ = if maximal then () else broadcast work_available;
9a2566d1fdbd more specific conditions: scheduler_event, work_available, work_finished -- considereably reduces overhead with many threads;
wenzelm
parents: 32186
diff changeset
   183
      in () end);
28167
27e2ca41b58c more interrupt operations;
wenzelm
parents: 28166
diff changeset
   184
  in () end;
27e2ca41b58c more interrupt operations;
wenzelm
parents: 28166
diff changeset
   185
27e2ca41b58c more interrupt operations;
wenzelm
parents: 28166
diff changeset
   186
32246
d4f7934e9555 misc tuning;
wenzelm
parents: 32230
diff changeset
   187
(* worker activity *)
d4f7934e9555 misc tuning;
wenzelm
parents: 32230
diff changeset
   188
d4f7934e9555 misc tuning;
wenzelm
parents: 32230
diff changeset
   189
fun count_active () = (*requires SYNCHRONIZED*)
33406
1ddcb8472bd2 slightly leaner and more direct control of worker activity etc.;
wenzelm
parents: 33061
diff changeset
   190
  fold (fn (_, active) => fn i => if ! active then i + 1 else i) (! workers) 0;
32246
d4f7934e9555 misc tuning;
wenzelm
parents: 32230
diff changeset
   191
33406
1ddcb8472bd2 slightly leaner and more direct control of worker activity etc.;
wenzelm
parents: 33061
diff changeset
   192
fun find_active () = (*requires SYNCHRONIZED*)
1ddcb8472bd2 slightly leaner and more direct control of worker activity etc.;
wenzelm
parents: 33061
diff changeset
   193
  (case AList.lookup Thread.equal (! workers) (Thread.self ()) of
1ddcb8472bd2 slightly leaner and more direct control of worker activity etc.;
wenzelm
parents: 33061
diff changeset
   194
    SOME active => active
1ddcb8472bd2 slightly leaner and more direct control of worker activity etc.;
wenzelm
parents: 33061
diff changeset
   195
  | NONE => raise Fail "Unregistered worker thread");
32246
d4f7934e9555 misc tuning;
wenzelm
parents: 32230
diff changeset
   196
d4f7934e9555 misc tuning;
wenzelm
parents: 32230
diff changeset
   197
28167
27e2ca41b58c more interrupt operations;
wenzelm
parents: 28166
diff changeset
   198
(* worker threads *)
27e2ca41b58c more interrupt operations;
wenzelm
parents: 28166
diff changeset
   199
32219
9a2566d1fdbd more specific conditions: scheduler_event, work_available, work_finished -- considereably reduces overhead with many threads;
wenzelm
parents: 32186
diff changeset
   200
fun worker_wait cond = (*requires SYNCHRONIZED*)
33406
1ddcb8472bd2 slightly leaner and more direct control of worker activity etc.;
wenzelm
parents: 33061
diff changeset
   201
  let
1ddcb8472bd2 slightly leaner and more direct control of worker activity etc.;
wenzelm
parents: 33061
diff changeset
   202
    val active = find_active ();
1ddcb8472bd2 slightly leaner and more direct control of worker activity etc.;
wenzelm
parents: 33061
diff changeset
   203
    val _ = active := false;
1ddcb8472bd2 slightly leaner and more direct control of worker activity etc.;
wenzelm
parents: 33061
diff changeset
   204
    val _ = wait cond;
1ddcb8472bd2 slightly leaner and more direct control of worker activity etc.;
wenzelm
parents: 33061
diff changeset
   205
    val _ = active := true;
1ddcb8472bd2 slightly leaner and more direct control of worker activity etc.;
wenzelm
parents: 33061
diff changeset
   206
  in () end;
28162
55772e4e95e0 tuned check_cache;
wenzelm
parents: 28156
diff changeset
   207
33409
wenzelm
parents: 33408
diff changeset
   208
fun worker_next have_work = (*requires SYNCHRONIZED*)
33406
1ddcb8472bd2 slightly leaner and more direct control of worker activity etc.;
wenzelm
parents: 33061
diff changeset
   209
  if length (! workers) > ! max_workers then
1ddcb8472bd2 slightly leaner and more direct control of worker activity etc.;
wenzelm
parents: 33061
diff changeset
   210
    (Unsynchronized.change workers (AList.delete Thread.equal (Thread.self ()));
33409
wenzelm
parents: 33408
diff changeset
   211
     if have_work then signal work_available else ();
32219
9a2566d1fdbd more specific conditions: scheduler_event, work_available, work_finished -- considereably reduces overhead with many threads;
wenzelm
parents: 32186
diff changeset
   212
     broadcast scheduler_event;
28167
27e2ca41b58c more interrupt operations;
wenzelm
parents: 28166
diff changeset
   213
     NONE)
33406
1ddcb8472bd2 slightly leaner and more direct control of worker activity etc.;
wenzelm
parents: 33061
diff changeset
   214
  else if count_active () > ! max_active then
33409
wenzelm
parents: 33408
diff changeset
   215
    (if have_work then signal work_available else ();
33407
1427333220bc worker_next: ensure that work_available is passed on before sleeping (was occasionally lost when worker configuration changed, causing scheduler deadlock);
wenzelm
parents: 33406
diff changeset
   216
     worker_wait scheduler_event;
1427333220bc worker_next: ensure that work_available is passed on before sleeping (was occasionally lost when worker configuration changed, causing scheduler deadlock);
wenzelm
parents: 33406
diff changeset
   217
     worker_next false)
28166
43087721a66e moved task, thread_data, group, queue to task_queue.ML;
wenzelm
parents: 28163
diff changeset
   218
  else
32738
15bb09ca0378 explicit indication of Unsynchronized.ref;
wenzelm
parents: 32724
diff changeset
   219
    (case Unsynchronized.change_result queue (Task_Queue.dequeue (Thread.self ())) of
33407
1427333220bc worker_next: ensure that work_available is passed on before sleeping (was occasionally lost when worker configuration changed, causing scheduler deadlock);
wenzelm
parents: 33406
diff changeset
   220
      NONE => (worker_wait work_available; worker_next true)
28166
43087721a66e moved task, thread_data, group, queue to task_queue.ML;
wenzelm
parents: 28163
diff changeset
   221
    | some => some);
28156
5205f7979b4f Functional threads as future values.
wenzelm
parents:
diff changeset
   222
28167
27e2ca41b58c more interrupt operations;
wenzelm
parents: 28166
diff changeset
   223
fun worker_loop name =
33407
1427333220bc worker_next: ensure that work_available is passed on before sleeping (was occasionally lost when worker configuration changed, causing scheduler deadlock);
wenzelm
parents: 33406
diff changeset
   224
  (case SYNCHRONIZED name (fn () => worker_next false) of
29119
99941fd0cb0e renamed structure TaskQueue to Task_Queue;
wenzelm
parents: 29118
diff changeset
   225
    NONE => ()
33408
a69ddd7dce95 tuned thread data;
wenzelm
parents: 33407
diff changeset
   226
  | SOME work => (execute work; worker_loop name));
28156
5205f7979b4f Functional threads as future values.
wenzelm
parents:
diff changeset
   227
33407
1427333220bc worker_next: ensure that work_available is passed on before sleeping (was occasionally lost when worker configuration changed, causing scheduler deadlock);
wenzelm
parents: 33406
diff changeset
   228
fun worker_start name = (*requires SYNCHRONIZED*)
1427333220bc worker_next: ensure that work_available is passed on before sleeping (was occasionally lost when worker configuration changed, causing scheduler deadlock);
wenzelm
parents: 33406
diff changeset
   229
  Unsynchronized.change workers (cons (SimpleThread.fork false (fn () => worker_loop name),
1427333220bc worker_next: ensure that work_available is passed on before sleeping (was occasionally lost when worker configuration changed, causing scheduler deadlock);
wenzelm
parents: 33406
diff changeset
   230
    Unsynchronized.ref true));
28156
5205f7979b4f Functional threads as future values.
wenzelm
parents:
diff changeset
   231
5205f7979b4f Functional threads as future values.
wenzelm
parents:
diff changeset
   232
5205f7979b4f Functional threads as future values.
wenzelm
parents:
diff changeset
   233
(* scheduler *)
5205f7979b4f Functional threads as future values.
wenzelm
parents:
diff changeset
   234
33407
1427333220bc worker_next: ensure that work_available is passed on before sleeping (was occasionally lost when worker configuration changed, causing scheduler deadlock);
wenzelm
parents: 33406
diff changeset
   235
val status_ticks = Unsynchronized.ref 0;
1427333220bc worker_next: ensure that work_available is passed on before sleeping (was occasionally lost when worker configuration changed, causing scheduler deadlock);
wenzelm
parents: 33406
diff changeset
   236
1427333220bc worker_next: ensure that work_available is passed on before sleeping (was occasionally lost when worker configuration changed, causing scheduler deadlock);
wenzelm
parents: 33406
diff changeset
   237
val last_round = Unsynchronized.ref Time.zeroTime;
32248
0241916a5f06 more precise treatment of scheduler_event: continous pulse (50ms) instead of flooding, which was burning many CPU cycles in spare threads;
wenzelm
parents: 32247
diff changeset
   238
val next_round = Time.fromMilliseconds 50;
32226
23511e4da055 tuned tracing;
wenzelm
parents: 32225
diff changeset
   239
28206
bcd48c6897d4 eliminated requests, use global state variables uniformly;
wenzelm
parents: 28203
diff changeset
   240
fun scheduler_next () = (*requires SYNCHRONIZED*)
28156
5205f7979b4f Functional threads as future values.
wenzelm
parents:
diff changeset
   241
  let
33407
1427333220bc worker_next: ensure that work_available is passed on before sleeping (was occasionally lost when worker configuration changed, causing scheduler deadlock);
wenzelm
parents: 33406
diff changeset
   242
    val now = Time.now ();
1427333220bc worker_next: ensure that work_available is passed on before sleeping (was occasionally lost when worker configuration changed, causing scheduler deadlock);
wenzelm
parents: 33406
diff changeset
   243
    val tick = Time.<= (Time.+ (! last_round, next_round), now);
1427333220bc worker_next: ensure that work_available is passed on before sleeping (was occasionally lost when worker configuration changed, causing scheduler deadlock);
wenzelm
parents: 33406
diff changeset
   244
    val _ = if tick then last_round := now else ();
1427333220bc worker_next: ensure that work_available is passed on before sleeping (was occasionally lost when worker configuration changed, causing scheduler deadlock);
wenzelm
parents: 33406
diff changeset
   245
32226
23511e4da055 tuned tracing;
wenzelm
parents: 32225
diff changeset
   246
    (*queue and worker status*)
23511e4da055 tuned tracing;
wenzelm
parents: 32225
diff changeset
   247
    val _ =
33407
1427333220bc worker_next: ensure that work_available is passed on before sleeping (was occasionally lost when worker configuration changed, causing scheduler deadlock);
wenzelm
parents: 33406
diff changeset
   248
      if tick then Unsynchronized.change status_ticks (fn i => (i + 1) mod 10) else ();
1427333220bc worker_next: ensure that work_available is passed on before sleeping (was occasionally lost when worker configuration changed, causing scheduler deadlock);
wenzelm
parents: 33406
diff changeset
   249
    val _ =
1427333220bc worker_next: ensure that work_available is passed on before sleeping (was occasionally lost when worker configuration changed, causing scheduler deadlock);
wenzelm
parents: 33406
diff changeset
   250
      if tick andalso ! status_ticks = 0 then
1427333220bc worker_next: ensure that work_available is passed on before sleeping (was occasionally lost when worker configuration changed, causing scheduler deadlock);
wenzelm
parents: 33406
diff changeset
   251
        Multithreading.tracing 1 (fn () =>
1427333220bc worker_next: ensure that work_available is passed on before sleeping (was occasionally lost when worker configuration changed, causing scheduler deadlock);
wenzelm
parents: 33406
diff changeset
   252
          let
1427333220bc worker_next: ensure that work_available is passed on before sleeping (was occasionally lost when worker configuration changed, causing scheduler deadlock);
wenzelm
parents: 33406
diff changeset
   253
            val {ready, pending, running} = Task_Queue.status (! queue);
1427333220bc worker_next: ensure that work_available is passed on before sleeping (was occasionally lost when worker configuration changed, causing scheduler deadlock);
wenzelm
parents: 33406
diff changeset
   254
            val total = length (! workers);
1427333220bc worker_next: ensure that work_available is passed on before sleeping (was occasionally lost when worker configuration changed, causing scheduler deadlock);
wenzelm
parents: 33406
diff changeset
   255
            val active = count_active ();
1427333220bc worker_next: ensure that work_available is passed on before sleeping (was occasionally lost when worker configuration changed, causing scheduler deadlock);
wenzelm
parents: 33406
diff changeset
   256
          in
1427333220bc worker_next: ensure that work_available is passed on before sleeping (was occasionally lost when worker configuration changed, causing scheduler deadlock);
wenzelm
parents: 33406
diff changeset
   257
            "SCHEDULE " ^ Time.toString now ^ ": " ^
1427333220bc worker_next: ensure that work_available is passed on before sleeping (was occasionally lost when worker configuration changed, causing scheduler deadlock);
wenzelm
parents: 33406
diff changeset
   258
              string_of_int ready ^ " ready, " ^
1427333220bc worker_next: ensure that work_available is passed on before sleeping (was occasionally lost when worker configuration changed, causing scheduler deadlock);
wenzelm
parents: 33406
diff changeset
   259
              string_of_int pending ^ " pending, " ^
1427333220bc worker_next: ensure that work_available is passed on before sleeping (was occasionally lost when worker configuration changed, causing scheduler deadlock);
wenzelm
parents: 33406
diff changeset
   260
              string_of_int running ^ " running; " ^
1427333220bc worker_next: ensure that work_available is passed on before sleeping (was occasionally lost when worker configuration changed, causing scheduler deadlock);
wenzelm
parents: 33406
diff changeset
   261
              string_of_int total ^ " workers, " ^
1427333220bc worker_next: ensure that work_available is passed on before sleeping (was occasionally lost when worker configuration changed, causing scheduler deadlock);
wenzelm
parents: 33406
diff changeset
   262
              string_of_int active ^ " active "
1427333220bc worker_next: ensure that work_available is passed on before sleeping (was occasionally lost when worker configuration changed, causing scheduler deadlock);
wenzelm
parents: 33406
diff changeset
   263
          end)
1427333220bc worker_next: ensure that work_available is passed on before sleeping (was occasionally lost when worker configuration changed, causing scheduler deadlock);
wenzelm
parents: 33406
diff changeset
   264
      else ();
32053
257eac3946e9 scheduler: tuned tracing (status);
wenzelm
parents: 31631
diff changeset
   265
28206
bcd48c6897d4 eliminated requests, use global state variables uniformly;
wenzelm
parents: 28203
diff changeset
   266
    (*worker threads*)
28191
9e5f556409c6 future: allow explicit group;
wenzelm
parents: 28186
diff changeset
   267
    val _ =
32219
9a2566d1fdbd more specific conditions: scheduler_event, work_available, work_finished -- considereably reduces overhead with many threads;
wenzelm
parents: 32186
diff changeset
   268
      if forall (Thread.isActive o #1) (! workers) then ()
32095
ad4be204fdfe tuned tracing;
wenzelm
parents: 32058
diff changeset
   269
      else
33409
wenzelm
parents: 33408
diff changeset
   270
        let
wenzelm
parents: 33408
diff changeset
   271
          val  (alive, dead) = List.partition (Thread.isActive o #1) (! workers);
wenzelm
parents: 33408
diff changeset
   272
          val _ = workers := alive;
wenzelm
parents: 33408
diff changeset
   273
        in
wenzelm
parents: 33408
diff changeset
   274
          Multithreading.tracing 0 (fn () =>
wenzelm
parents: 33408
diff changeset
   275
            "SCHEDULE: disposed " ^ string_of_int (length dead) ^ " dead worker threads")
wenzelm
parents: 33408
diff changeset
   276
        end;
28191
9e5f556409c6 future: allow explicit group;
wenzelm
parents: 28186
diff changeset
   277
28206
bcd48c6897d4 eliminated requests, use global state variables uniformly;
wenzelm
parents: 28203
diff changeset
   278
    val m = if ! do_shutdown then 0 else Multithreading.max_threads_value ();
33406
1ddcb8472bd2 slightly leaner and more direct control of worker activity etc.;
wenzelm
parents: 33061
diff changeset
   279
    val _ = max_active := m;
1ddcb8472bd2 slightly leaner and more direct control of worker activity etc.;
wenzelm
parents: 33061
diff changeset
   280
32644
e4511a1b4c1b full reserve of worker threads -- for improved CPU utilization;
wenzelm
parents: 32617
diff changeset
   281
    val mm = if m = 9999 then 1 else m * 2;
33406
1ddcb8472bd2 slightly leaner and more direct control of worker activity etc.;
wenzelm
parents: 33061
diff changeset
   282
    val _ = max_workers := mm;
1ddcb8472bd2 slightly leaner and more direct control of worker activity etc.;
wenzelm
parents: 33061
diff changeset
   283
33407
1427333220bc worker_next: ensure that work_available is passed on before sleeping (was occasionally lost when worker configuration changed, causing scheduler deadlock);
wenzelm
parents: 33406
diff changeset
   284
    val missing = ! max_workers - length (! workers);
28203
88f18387f1c9 shutdown: global join-and-shutdown operation;
wenzelm
parents: 28202
diff changeset
   285
    val _ =
33407
1427333220bc worker_next: ensure that work_available is passed on before sleeping (was occasionally lost when worker configuration changed, causing scheduler deadlock);
wenzelm
parents: 33406
diff changeset
   286
      if missing > 0 then
1427333220bc worker_next: ensure that work_available is passed on before sleeping (was occasionally lost when worker configuration changed, causing scheduler deadlock);
wenzelm
parents: 33406
diff changeset
   287
       (funpow missing (fn () =>
1427333220bc worker_next: ensure that work_available is passed on before sleeping (was occasionally lost when worker configuration changed, causing scheduler deadlock);
wenzelm
parents: 33406
diff changeset
   288
          ignore (worker_start ("worker " ^ string_of_int (Unsynchronized.inc next)))) ();
1427333220bc worker_next: ensure that work_available is passed on before sleeping (was occasionally lost when worker configuration changed, causing scheduler deadlock);
wenzelm
parents: 33406
diff changeset
   289
        broadcast scheduler_event)
28203
88f18387f1c9 shutdown: global join-and-shutdown operation;
wenzelm
parents: 28202
diff changeset
   290
      else ();
28206
bcd48c6897d4 eliminated requests, use global state variables uniformly;
wenzelm
parents: 28203
diff changeset
   291
bcd48c6897d4 eliminated requests, use global state variables uniformly;
wenzelm
parents: 28203
diff changeset
   292
    (*canceled groups*)
32225
d5d6f47fb018 cancel: improved reactivity due to more careful broadcasting;
wenzelm
parents: 32224
diff changeset
   293
    val _ =
d5d6f47fb018 cancel: improved reactivity due to more careful broadcasting;
wenzelm
parents: 32224
diff changeset
   294
      if null (! canceled) then ()
32293
e0b8da3fae4d tuned tracing;
wenzelm
parents: 32286
diff changeset
   295
      else
e0b8da3fae4d tuned tracing;
wenzelm
parents: 32286
diff changeset
   296
       (Multithreading.tracing 1 (fn () =>
e0b8da3fae4d tuned tracing;
wenzelm
parents: 32286
diff changeset
   297
          string_of_int (length (! canceled)) ^ " canceled groups");
32738
15bb09ca0378 explicit indication of Unsynchronized.ref;
wenzelm
parents: 32724
diff changeset
   298
        Unsynchronized.change canceled (filter_out (Task_Queue.cancel (! queue)));
32293
e0b8da3fae4d tuned tracing;
wenzelm
parents: 32286
diff changeset
   299
        broadcast_work ());
28206
bcd48c6897d4 eliminated requests, use global state variables uniformly;
wenzelm
parents: 28203
diff changeset
   300
32228
7622c03141b0 scheduler: shutdown spontaneously (after some delay) if queue is empty;
wenzelm
parents: 32227
diff changeset
   301
    (*delay loop*)
32295
400cc493d466 renamed Multithreading.regular_interrupts to Multithreading.public_interrupts;
wenzelm
parents: 32293
diff changeset
   302
    val _ = Exn.release (wait_timeout next_round scheduler_event);
28167
27e2ca41b58c more interrupt operations;
wenzelm
parents: 28166
diff changeset
   303
32219
9a2566d1fdbd more specific conditions: scheduler_event, work_available, work_finished -- considereably reduces overhead with many threads;
wenzelm
parents: 32186
diff changeset
   304
    (*shutdown*)
32228
7622c03141b0 scheduler: shutdown spontaneously (after some delay) if queue is empty;
wenzelm
parents: 32227
diff changeset
   305
    val _ = if Task_Queue.is_empty (! queue) then do_shutdown := true else ();
32219
9a2566d1fdbd more specific conditions: scheduler_event, work_available, work_finished -- considereably reduces overhead with many threads;
wenzelm
parents: 32186
diff changeset
   306
    val continue = not (! do_shutdown andalso null (! workers));
9a2566d1fdbd more specific conditions: scheduler_event, work_available, work_finished -- considereably reduces overhead with many threads;
wenzelm
parents: 32186
diff changeset
   307
    val _ = if continue then () else scheduler := NONE;
9a2566d1fdbd more specific conditions: scheduler_event, work_available, work_finished -- considereably reduces overhead with many threads;
wenzelm
parents: 32186
diff changeset
   308
    val _ = broadcast scheduler_event;
32295
400cc493d466 renamed Multithreading.regular_interrupts to Multithreading.public_interrupts;
wenzelm
parents: 32293
diff changeset
   309
  in continue end
400cc493d466 renamed Multithreading.regular_interrupts to Multithreading.public_interrupts;
wenzelm
parents: 32293
diff changeset
   310
  handle Exn.Interrupt =>
400cc493d466 renamed Multithreading.regular_interrupts to Multithreading.public_interrupts;
wenzelm
parents: 32293
diff changeset
   311
   (Multithreading.tracing 1 (fn () => "Interrupt");
32296
e6a8ed8aed3a future scheduler: uninterruptible cancelation;
wenzelm
parents: 32295
diff changeset
   312
    uninterruptible (fn _ => fn () => List.app do_cancel (Task_Queue.cancel_all (! queue))) ();
32295
400cc493d466 renamed Multithreading.regular_interrupts to Multithreading.public_interrupts;
wenzelm
parents: 32293
diff changeset
   313
    scheduler_next ());
400cc493d466 renamed Multithreading.regular_interrupts to Multithreading.public_interrupts;
wenzelm
parents: 32293
diff changeset
   314
28206
bcd48c6897d4 eliminated requests, use global state variables uniformly;
wenzelm
parents: 28203
diff changeset
   315
fun scheduler_loop () =
32295
400cc493d466 renamed Multithreading.regular_interrupts to Multithreading.public_interrupts;
wenzelm
parents: 32293
diff changeset
   316
  Multithreading.with_attributes
400cc493d466 renamed Multithreading.regular_interrupts to Multithreading.public_interrupts;
wenzelm
parents: 32293
diff changeset
   317
    (Multithreading.sync_interrupts Multithreading.public_interrupts)
400cc493d466 renamed Multithreading.regular_interrupts to Multithreading.public_interrupts;
wenzelm
parents: 32293
diff changeset
   318
    (fn _ => while SYNCHRONIZED "scheduler" (fn () => scheduler_next ()) do ());
28156
5205f7979b4f Functional threads as future values.
wenzelm
parents:
diff changeset
   319
28203
88f18387f1c9 shutdown: global join-and-shutdown operation;
wenzelm
parents: 28202
diff changeset
   320
fun scheduler_active () = (*requires SYNCHRONIZED*)
88f18387f1c9 shutdown: global join-and-shutdown operation;
wenzelm
parents: 28202
diff changeset
   321
  (case ! scheduler of NONE => false | SOME thread => Thread.isActive thread);
88f18387f1c9 shutdown: global join-and-shutdown operation;
wenzelm
parents: 28202
diff changeset
   322
32228
7622c03141b0 scheduler: shutdown spontaneously (after some delay) if queue is empty;
wenzelm
parents: 32227
diff changeset
   323
fun scheduler_check () = (*requires SYNCHRONIZED*)
7622c03141b0 scheduler: shutdown spontaneously (after some delay) if queue is empty;
wenzelm
parents: 32227
diff changeset
   324
 (do_shutdown := false;
32248
0241916a5f06 more precise treatment of scheduler_event: continous pulse (50ms) instead of flooding, which was burning many CPU cycles in spare threads;
wenzelm
parents: 32247
diff changeset
   325
  if scheduler_active () then ()
0241916a5f06 more precise treatment of scheduler_event: continous pulse (50ms) instead of flooding, which was burning many CPU cycles in spare threads;
wenzelm
parents: 32247
diff changeset
   326
  else scheduler := SOME (SimpleThread.fork false scheduler_loop));
28156
5205f7979b4f Functional threads as future values.
wenzelm
parents:
diff changeset
   327
5205f7979b4f Functional threads as future values.
wenzelm
parents:
diff changeset
   328
29366
1ffc8cbf39ec tuned map: reduced overhead due to bulk jobs;
wenzelm
parents: 29341
diff changeset
   329
1ffc8cbf39ec tuned map: reduced overhead due to bulk jobs;
wenzelm
parents: 29341
diff changeset
   330
(** futures **)
28156
5205f7979b4f Functional threads as future values.
wenzelm
parents:
diff changeset
   331
29366
1ffc8cbf39ec tuned map: reduced overhead due to bulk jobs;
wenzelm
parents: 29341
diff changeset
   332
(* fork *)
1ffc8cbf39ec tuned map: reduced overhead due to bulk jobs;
wenzelm
parents: 29341
diff changeset
   333
1ffc8cbf39ec tuned map: reduced overhead due to bulk jobs;
wenzelm
parents: 29341
diff changeset
   334
fun fork_future opt_group deps pri e =
1ffc8cbf39ec tuned map: reduced overhead due to bulk jobs;
wenzelm
parents: 29341
diff changeset
   335
  let
32102
81d03a29980c added worker_group;
wenzelm
parents: 32099
diff changeset
   336
    val group =
81d03a29980c added worker_group;
wenzelm
parents: 32099
diff changeset
   337
      (case opt_group of
81d03a29980c added worker_group;
wenzelm
parents: 32099
diff changeset
   338
        SOME group => group
81d03a29980c added worker_group;
wenzelm
parents: 32099
diff changeset
   339
      | NONE => Task_Queue.new_group (worker_group ()));
29366
1ffc8cbf39ec tuned map: reduced overhead due to bulk jobs;
wenzelm
parents: 29341
diff changeset
   340
    val (result, job) = future_job group e;
32246
d4f7934e9555 misc tuning;
wenzelm
parents: 32230
diff changeset
   341
    val task = SYNCHRONIZED "enqueue" (fn () =>
32219
9a2566d1fdbd more specific conditions: scheduler_event, work_available, work_finished -- considereably reduces overhead with many threads;
wenzelm
parents: 32186
diff changeset
   342
      let
32738
15bb09ca0378 explicit indication of Unsynchronized.ref;
wenzelm
parents: 32724
diff changeset
   343
        val (task, minimal) =
15bb09ca0378 explicit indication of Unsynchronized.ref;
wenzelm
parents: 32724
diff changeset
   344
          Unsynchronized.change_result queue (Task_Queue.enqueue group deps pri job);
32219
9a2566d1fdbd more specific conditions: scheduler_event, work_available, work_finished -- considereably reduces overhead with many threads;
wenzelm
parents: 32186
diff changeset
   345
        val _ = if minimal then signal work_available else ();
32228
7622c03141b0 scheduler: shutdown spontaneously (after some delay) if queue is empty;
wenzelm
parents: 32227
diff changeset
   346
        val _ = scheduler_check ();
32219
9a2566d1fdbd more specific conditions: scheduler_event, work_available, work_finished -- considereably reduces overhead with many threads;
wenzelm
parents: 32186
diff changeset
   347
      in task end);
28166
43087721a66e moved task, thread_data, group, queue to task_queue.ML;
wenzelm
parents: 28163
diff changeset
   348
  in Future {task = task, group = group, result = result} end;
28162
55772e4e95e0 tuned check_cache;
wenzelm
parents: 28156
diff changeset
   349
29366
1ffc8cbf39ec tuned map: reduced overhead due to bulk jobs;
wenzelm
parents: 29341
diff changeset
   350
fun fork_group group e = fork_future (SOME group) [] 0 e;
32724
aaeeb0ba2035 added fork_deps_pri;
wenzelm
parents: 32644
diff changeset
   351
fun fork_deps_pri deps pri e = fork_future NONE (map task_of deps) pri e;
aaeeb0ba2035 added fork_deps_pri;
wenzelm
parents: 32644
diff changeset
   352
fun fork_deps deps e = fork_deps_pri deps 0 e;
aaeeb0ba2035 added fork_deps_pri;
wenzelm
parents: 32644
diff changeset
   353
fun fork_pri pri e = fork_deps_pri [] pri e;
aaeeb0ba2035 added fork_deps_pri;
wenzelm
parents: 32644
diff changeset
   354
fun fork e = fork_deps [] e;
28186
6a8417f36837 cancel: check_scheduler;
wenzelm
parents: 28177
diff changeset
   355
6a8417f36837 cancel: check_scheduler;
wenzelm
parents: 28177
diff changeset
   356
29366
1ffc8cbf39ec tuned map: reduced overhead due to bulk jobs;
wenzelm
parents: 29341
diff changeset
   357
(* join *)
1ffc8cbf39ec tuned map: reduced overhead due to bulk jobs;
wenzelm
parents: 29341
diff changeset
   358
29551
95e469919c3e join_results: when dependencies are resulved (but not finished yet),
wenzelm
parents: 29431
diff changeset
   359
local
95e469919c3e join_results: when dependencies are resulved (but not finished yet),
wenzelm
parents: 29431
diff changeset
   360
32099
5382c93108db propagate exceptions within future groups;
wenzelm
parents: 32096
diff changeset
   361
fun get_result x =
5382c93108db propagate exceptions within future groups;
wenzelm
parents: 32096
diff changeset
   362
  (case peek x of
32102
81d03a29980c added worker_group;
wenzelm
parents: 32099
diff changeset
   363
    NONE => Exn.Exn (SYS_ERROR "unfinished future")
81d03a29980c added worker_group;
wenzelm
parents: 32099
diff changeset
   364
  | SOME (Exn.Exn Exn.Interrupt) =>
81d03a29980c added worker_group;
wenzelm
parents: 32099
diff changeset
   365
      Exn.Exn (Exn.EXCEPTIONS (Exn.flatten_list (Task_Queue.group_status (group_of x))))
81d03a29980c added worker_group;
wenzelm
parents: 32099
diff changeset
   366
  | SOME res => res);
28186
6a8417f36837 cancel: check_scheduler;
wenzelm
parents: 28177
diff changeset
   367
33409
wenzelm
parents: 33408
diff changeset
   368
fun passive_wait x =
33061
e3e61133e0fc use Synchronized.assign to achieve actual immutable results;
wenzelm
parents: 32814
diff changeset
   369
  Synchronized.readonly_access (result_of x) (fn NONE => NONE | SOME _ => SOME ());
32224
a46f5e9b1718 dequeue_towards: always return active tasks;
wenzelm
parents: 32222
diff changeset
   370
32095
ad4be204fdfe tuned tracing;
wenzelm
parents: 32058
diff changeset
   371
fun join_next deps = (*requires SYNCHRONIZED*)
32224
a46f5e9b1718 dequeue_towards: always return active tasks;
wenzelm
parents: 32222
diff changeset
   372
  if null deps then NONE
a46f5e9b1718 dequeue_towards: always return active tasks;
wenzelm
parents: 32222
diff changeset
   373
  else
32738
15bb09ca0378 explicit indication of Unsynchronized.ref;
wenzelm
parents: 32724
diff changeset
   374
    (case Unsynchronized.change_result queue (Task_Queue.dequeue_towards (Thread.self ()) deps) of
32224
a46f5e9b1718 dequeue_towards: always return active tasks;
wenzelm
parents: 32222
diff changeset
   375
      (NONE, []) => NONE
a46f5e9b1718 dequeue_towards: always return active tasks;
wenzelm
parents: 32222
diff changeset
   376
    | (NONE, deps') => (worker_wait work_finished; join_next deps')
a46f5e9b1718 dequeue_towards: always return active tasks;
wenzelm
parents: 32222
diff changeset
   377
    | (SOME work, deps') => SOME (work, deps'));
32095
ad4be204fdfe tuned tracing;
wenzelm
parents: 32058
diff changeset
   378
32814
81897d30b97f added Task_Queue.depend (again) -- light-weight version for transitive graph;
wenzelm
parents: 32738
diff changeset
   379
fun execute_work NONE = ()
33408
a69ddd7dce95 tuned thread data;
wenzelm
parents: 33407
diff changeset
   380
  | execute_work (SOME (work, deps')) = (execute work; join_work deps')
32814
81897d30b97f added Task_Queue.depend (again) -- light-weight version for transitive graph;
wenzelm
parents: 32738
diff changeset
   381
and join_work deps =
81897d30b97f added Task_Queue.depend (again) -- light-weight version for transitive graph;
wenzelm
parents: 32738
diff changeset
   382
  execute_work (SYNCHRONIZED "join" (fn () => join_next deps));
81897d30b97f added Task_Queue.depend (again) -- light-weight version for transitive graph;
wenzelm
parents: 32738
diff changeset
   383
81897d30b97f added Task_Queue.depend (again) -- light-weight version for transitive graph;
wenzelm
parents: 32738
diff changeset
   384
fun join_depend task deps =
81897d30b97f added Task_Queue.depend (again) -- light-weight version for transitive graph;
wenzelm
parents: 32738
diff changeset
   385
  execute_work (SYNCHRONIZED "join" (fn () =>
81897d30b97f added Task_Queue.depend (again) -- light-weight version for transitive graph;
wenzelm
parents: 32738
diff changeset
   386
    (Unsynchronized.change queue (Task_Queue.depend task deps); join_next deps)));
29551
95e469919c3e join_results: when dependencies are resulved (but not finished yet),
wenzelm
parents: 29431
diff changeset
   387
95e469919c3e join_results: when dependencies are resulved (but not finished yet),
wenzelm
parents: 29431
diff changeset
   388
in
95e469919c3e join_results: when dependencies are resulved (but not finished yet),
wenzelm
parents: 29431
diff changeset
   389
29366
1ffc8cbf39ec tuned map: reduced overhead due to bulk jobs;
wenzelm
parents: 29341
diff changeset
   390
fun join_results xs =
1ffc8cbf39ec tuned map: reduced overhead due to bulk jobs;
wenzelm
parents: 29341
diff changeset
   391
  if forall is_finished xs then map get_result xs
32246
d4f7934e9555 misc tuning;
wenzelm
parents: 32230
diff changeset
   392
  else if Multithreading.self_critical () then
d4f7934e9555 misc tuning;
wenzelm
parents: 32230
diff changeset
   393
    error "Cannot join future values within critical section"
32814
81897d30b97f added Task_Queue.depend (again) -- light-weight version for transitive graph;
wenzelm
parents: 32738
diff changeset
   394
  else
81897d30b97f added Task_Queue.depend (again) -- light-weight version for transitive graph;
wenzelm
parents: 32738
diff changeset
   395
    (case worker_task () of
81897d30b97f added Task_Queue.depend (again) -- light-weight version for transitive graph;
wenzelm
parents: 32738
diff changeset
   396
      SOME task => join_depend task (map task_of xs)
33409
wenzelm
parents: 33408
diff changeset
   397
    | NONE => List.app passive_wait xs;
32814
81897d30b97f added Task_Queue.depend (again) -- light-weight version for transitive graph;
wenzelm
parents: 32738
diff changeset
   398
    map get_result xs);
28186
6a8417f36837 cancel: check_scheduler;
wenzelm
parents: 28177
diff changeset
   399
29551
95e469919c3e join_results: when dependencies are resulved (but not finished yet),
wenzelm
parents: 29431
diff changeset
   400
end;
95e469919c3e join_results: when dependencies are resulved (but not finished yet),
wenzelm
parents: 29431
diff changeset
   401
28647
8068cdc84e7e join_results: allow CRITICAL join of finished futures;
wenzelm
parents: 28645
diff changeset
   402
fun join_result x = singleton join_results x;
8068cdc84e7e join_results: allow CRITICAL join of finished futures;
wenzelm
parents: 28645
diff changeset
   403
fun join x = Exn.release (join_result x);
28156
5205f7979b4f Functional threads as future values.
wenzelm
parents:
diff changeset
   404
29366
1ffc8cbf39ec tuned map: reduced overhead due to bulk jobs;
wenzelm
parents: 29341
diff changeset
   405
1ffc8cbf39ec tuned map: reduced overhead due to bulk jobs;
wenzelm
parents: 29341
diff changeset
   406
(* map *)
1ffc8cbf39ec tuned map: reduced overhead due to bulk jobs;
wenzelm
parents: 29341
diff changeset
   407
29384
a3c7e9ae9b71 more robust propagation of errors through bulk jobs;
wenzelm
parents: 29366
diff changeset
   408
fun map_future f x =
29366
1ffc8cbf39ec tuned map: reduced overhead due to bulk jobs;
wenzelm
parents: 29341
diff changeset
   409
  let
29384
a3c7e9ae9b71 more robust propagation of errors through bulk jobs;
wenzelm
parents: 29366
diff changeset
   410
    val task = task_of x;
32102
81d03a29980c added worker_group;
wenzelm
parents: 32099
diff changeset
   411
    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
   412
    val (result, job) = future_job group (fn () => f (join x));
a3c7e9ae9b71 more robust propagation of errors through bulk jobs;
wenzelm
parents: 29366
diff changeset
   413
32246
d4f7934e9555 misc tuning;
wenzelm
parents: 32230
diff changeset
   414
    val extended = SYNCHRONIZED "extend" (fn () =>
29366
1ffc8cbf39ec tuned map: reduced overhead due to bulk jobs;
wenzelm
parents: 29341
diff changeset
   415
      (case Task_Queue.extend task job (! queue) of
1ffc8cbf39ec tuned map: reduced overhead due to bulk jobs;
wenzelm
parents: 29341
diff changeset
   416
        SOME queue' => (queue := queue'; true)
1ffc8cbf39ec tuned map: reduced overhead due to bulk jobs;
wenzelm
parents: 29341
diff changeset
   417
      | NONE => false));
1ffc8cbf39ec tuned map: reduced overhead due to bulk jobs;
wenzelm
parents: 29341
diff changeset
   418
  in
29384
a3c7e9ae9b71 more robust propagation of errors through bulk jobs;
wenzelm
parents: 29366
diff changeset
   419
    if extended then Future {task = task, group = group, result = result}
32099
5382c93108db propagate exceptions within future groups;
wenzelm
parents: 32096
diff changeset
   420
    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
   421
  end;
28979
3ce619d8d432 fork/map: no inheritance of group (structure is nested, not parallel);
wenzelm
parents: 28972
diff changeset
   422
28191
9e5f556409c6 future: allow explicit group;
wenzelm
parents: 28186
diff changeset
   423
29431
0ebe652bfd5a added cancel_group;
wenzelm
parents: 29384
diff changeset
   424
(* cancellation *)
28202
23cb9a974630 added focus, which indicates a particular collection of high-priority tasks;
wenzelm
parents: 28201
diff changeset
   425
30618
046f4f986fb5 restricted interrupts for tasks running as future worker thread -- attempt to prevent interrupt race conditions;
wenzelm
parents: 30612
diff changeset
   426
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
   427
  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
   428
    Multithreading.with_attributes
32058
c76fd93b3b99 more abstract Future.is_worker;
wenzelm
parents: 32055
diff changeset
   429
      (if is_worker ()
32295
400cc493d466 renamed Multithreading.regular_interrupts to Multithreading.public_interrupts;
wenzelm
parents: 32293
diff changeset
   430
       then Multithreading.private_interrupts
400cc493d466 renamed Multithreading.regular_interrupts to Multithreading.public_interrupts;
wenzelm
parents: 32293
diff changeset
   431
       else Multithreading.public_interrupts)
400cc493d466 renamed Multithreading.regular_interrupts to Multithreading.public_interrupts;
wenzelm
parents: 32293
diff changeset
   432
      (fn _ => f x)
30618
046f4f986fb5 restricted interrupts for tasks running as future worker thread -- attempt to prevent interrupt race conditions;
wenzelm
parents: 30612
diff changeset
   433
  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
   434
32228
7622c03141b0 scheduler: shutdown spontaneously (after some delay) if queue is empty;
wenzelm
parents: 32227
diff changeset
   435
(*cancel: present and future group members will be interrupted eventually*)
7622c03141b0 scheduler: shutdown spontaneously (after some delay) if queue is empty;
wenzelm
parents: 32227
diff changeset
   436
fun cancel_group group = SYNCHRONIZED "cancel" (fn () => do_cancel group);
29431
0ebe652bfd5a added cancel_group;
wenzelm
parents: 29384
diff changeset
   437
fun cancel x = cancel_group (group_of x);
28206
bcd48c6897d4 eliminated requests, use global state variables uniformly;
wenzelm
parents: 28203
diff changeset
   438
29366
1ffc8cbf39ec tuned map: reduced overhead due to bulk jobs;
wenzelm
parents: 29341
diff changeset
   439
32228
7622c03141b0 scheduler: shutdown spontaneously (after some delay) if queue is empty;
wenzelm
parents: 32227
diff changeset
   440
(* shutdown *)
29366
1ffc8cbf39ec tuned map: reduced overhead due to bulk jobs;
wenzelm
parents: 29341
diff changeset
   441
28203
88f18387f1c9 shutdown: global join-and-shutdown operation;
wenzelm
parents: 28202
diff changeset
   442
fun shutdown () =
28276
fbc707811203 shutdown only if Multithreading.available;
wenzelm
parents: 28242
diff changeset
   443
  if Multithreading.available then
fbc707811203 shutdown only if Multithreading.available;
wenzelm
parents: 28242
diff changeset
   444
    SYNCHRONIZED "shutdown" (fn () =>
32228
7622c03141b0 scheduler: shutdown spontaneously (after some delay) if queue is empty;
wenzelm
parents: 32227
diff changeset
   445
     while scheduler_active () do
32248
0241916a5f06 more precise treatment of scheduler_event: continous pulse (50ms) instead of flooding, which was burning many CPU cycles in spare threads;
wenzelm
parents: 32247
diff changeset
   446
      (wait scheduler_event; broadcast_work ()))
28276
fbc707811203 shutdown only if Multithreading.available;
wenzelm
parents: 28242
diff changeset
   447
  else ();
28203
88f18387f1c9 shutdown: global join-and-shutdown operation;
wenzelm
parents: 28202
diff changeset
   448
29366
1ffc8cbf39ec tuned map: reduced overhead due to bulk jobs;
wenzelm
parents: 29341
diff changeset
   449
1ffc8cbf39ec tuned map: reduced overhead due to bulk jobs;
wenzelm
parents: 29341
diff changeset
   450
(*final declarations of this structure!*)
1ffc8cbf39ec tuned map: reduced overhead due to bulk jobs;
wenzelm
parents: 29341
diff changeset
   451
val map = map_future;
1ffc8cbf39ec tuned map: reduced overhead due to bulk jobs;
wenzelm
parents: 29341
diff changeset
   452
28156
5205f7979b4f Functional threads as future values.
wenzelm
parents:
diff changeset
   453
end;
28972
cb8a2c3e188f renamed type Future.T to future;
wenzelm
parents: 28647
diff changeset
   454
cb8a2c3e188f renamed type Future.T to future;
wenzelm
parents: 28647
diff changeset
   455
type 'a future = 'a Future.future;
cb8a2c3e188f renamed type Future.T to future;
wenzelm
parents: 28647
diff changeset
   456