src/Pure/Concurrent/task_queue.ML
author wenzelm
Sat Apr 02 23:29:05 2016 +0200 (2016-04-02 ago)
changeset 62826 eb94e570c1a4
parent 62819 d3ff367a16a0
child 62891 7a11ea5c9626
permissions -rw-r--r--
prefer infix operations;
wenzelm@28165
     1
(*  Title:      Pure/Concurrent/task_queue.ML
wenzelm@28165
     2
    Author:     Makarius
wenzelm@28165
     3
wenzelm@28165
     4
Ordered queue of grouped tasks.
wenzelm@28165
     5
*)
wenzelm@28165
     6
wenzelm@28165
     7
signature TASK_QUEUE =
wenzelm@28165
     8
sig
wenzelm@29340
     9
  type group
wenzelm@32221
    10
  val new_group: group option -> group
wenzelm@32052
    11
  val group_id: group -> int
wenzelm@29340
    12
  val eq_group: group * group -> bool
wenzelm@32221
    13
  val cancel_group: group -> exn -> unit
wenzelm@32221
    14
  val is_canceled: group -> bool
wenzelm@53190
    15
  val group_status: group -> exn list
wenzelm@28179
    16
  val str_of_group: group -> string
wenzelm@43951
    17
  val str_of_groups: group -> string
wenzelm@60610
    18
  val urgent_pri: int
wenzelm@41683
    19
  type task
wenzelm@45136
    20
  val dummy_task: task
wenzelm@41683
    21
  val group_of_task: task -> group
wenzelm@41683
    22
  val name_of_task: task -> string
wenzelm@41683
    23
  val pri_of_task: task -> int
wenzelm@41683
    24
  val str_of_task: task -> string
wenzelm@43951
    25
  val str_of_task_groups: task -> string
wenzelm@50975
    26
  val task_statistics: task -> Properties.T
wenzelm@41695
    27
  val running: task -> (unit -> 'a) -> 'a
wenzelm@41695
    28
  val joining: task -> (unit -> 'a) -> 'a
wenzelm@41695
    29
  val waiting: task -> task list -> (unit -> 'a) -> 'a
wenzelm@28165
    30
  type queue
wenzelm@28165
    31
  val empty: queue
wenzelm@47404
    32
  val group_tasks: queue -> group -> task list
wenzelm@41708
    33
  val known_task: queue -> task -> bool
wenzelm@34277
    34
  val all_passive: queue -> bool
wenzelm@60610
    35
  val status: queue -> {ready: int, pending: int, running: int, passive: int, urgent: int}
wenzelm@47404
    36
  val cancel: queue -> group -> Thread.thread list
wenzelm@44341
    37
  val cancel_all: queue -> group list * Thread.thread list
wenzelm@41681
    38
  val finish: task -> queue -> bool * queue
wenzelm@54649
    39
  val enroll: Thread.thread -> string -> group -> queue -> task * queue
wenzelm@37854
    40
  val enqueue_passive: group -> (unit -> bool) -> queue -> task * queue
wenzelm@41708
    41
  val enqueue: string -> group -> task list -> int -> (bool -> bool) -> queue -> task * queue
wenzelm@29365
    42
  val extend: task -> (bool -> bool) -> queue -> queue option
wenzelm@47423
    43
  val dequeue_passive: Thread.thread -> task -> queue -> bool option * queue
wenzelm@60610
    44
  val dequeue: Thread.thread -> bool -> queue -> (task * (bool -> bool) list) option * queue
wenzelm@41695
    45
  val dequeue_deps: Thread.thread -> task list -> queue ->
wenzelm@41695
    46
    (((task * (bool -> bool) list) option * task list) * queue)
wenzelm@28165
    47
end;
wenzelm@28165
    48
wenzelm@35242
    49
structure Task_Queue: TASK_QUEUE =
wenzelm@28165
    50
struct
wenzelm@28165
    51
wenzelm@52537
    52
val new_id = Counter.make ();
wenzelm@40450
    53
wenzelm@40450
    54
wenzelm@41683
    55
(** nested groups of tasks **)
wenzelm@41670
    56
wenzelm@41683
    57
(* groups *)
wenzelm@32101
    58
wenzelm@35242
    59
abstype group = Group of
wenzelm@32101
    60
 {parent: group option,
wenzelm@40450
    61
  id: int,
wenzelm@44247
    62
  status: exn option Synchronized.var}
wenzelm@35242
    63
with
wenzelm@29121
    64
wenzelm@32101
    65
fun make_group (parent, id, status) = Group {parent = parent, id = id, status = status};
wenzelm@32052
    66
wenzelm@44247
    67
fun new_group parent = make_group (parent, new_id (), Synchronized.var "group_status" NONE);
wenzelm@32101
    68
wenzelm@32101
    69
fun group_id (Group {id, ...}) = id;
wenzelm@32101
    70
fun eq_group (group1, group2) = group_id group1 = group_id group2;
wenzelm@28551
    71
wenzelm@43951
    72
fun fold_groups f (g as Group {parent = NONE, ...}) a = f g a
wenzelm@43951
    73
  | fold_groups f (g as Group {parent = SOME group, ...}) a = fold_groups f group (f g a);
wenzelm@32101
    74
wenzelm@32101
    75
wenzelm@32221
    76
(* group status *)
wenzelm@32221
    77
wenzelm@32251
    78
fun cancel_group (Group {status, ...}) exn =
wenzelm@32251
    79
  Synchronized.change status
wenzelm@44247
    80
    (fn exns => SOME (Par_Exn.make (exn :: the_list exns)));
wenzelm@29121
    81
wenzelm@32251
    82
fun is_canceled (Group {parent, status, ...}) =
wenzelm@44247
    83
  is_some (Synchronized.value status) orelse
wenzelm@32251
    84
    (case parent of NONE => false | SOME group => is_canceled group);
wenzelm@32221
    85
wenzelm@53190
    86
fun group_status (Group {parent, status, ...}) =
wenzelm@44247
    87
  the_list (Synchronized.value status) @
wenzelm@53190
    88
    (case parent of NONE => [] | SOME group => group_status group);
wenzelm@28165
    89
wenzelm@32101
    90
fun str_of_group group =
wenzelm@32101
    91
  (is_canceled group ? enclose "(" ")") (string_of_int (group_id group));
wenzelm@28179
    92
wenzelm@43951
    93
fun str_of_groups group =
wenzelm@43951
    94
  space_implode "/" (map str_of_group (rev (fold_groups cons group [])));
wenzelm@43951
    95
wenzelm@35242
    96
end;
wenzelm@35242
    97
wenzelm@28165
    98
wenzelm@41683
    99
(* tasks *)
wenzelm@41683
   100
wenzelm@60610
   101
val urgent_pri = 1000;
wenzelm@60610
   102
wenzelm@41683
   103
type timing = Time.time * Time.time * string list;  (*run, wait, wait dependencies*)
wenzelm@41683
   104
wenzelm@45136
   105
val timing_start = (Time.zeroTime, Time.zeroTime, []): timing;
wenzelm@45136
   106
wenzelm@45354
   107
fun new_timing () =
wenzelm@45354
   108
  if ! Multithreading.trace < 2 then NONE
wenzelm@45354
   109
  else SOME (Synchronized.var "timing" timing_start);
wenzelm@41683
   110
wenzelm@41683
   111
abstype task = Task of
wenzelm@41683
   112
 {group: group,
wenzelm@41683
   113
  name: string,
wenzelm@41683
   114
  id: int,
wenzelm@41683
   115
  pri: int option,
wenzelm@50975
   116
  timing: timing Synchronized.var option,
wenzelm@50975
   117
  pos: Position.T}
wenzelm@41683
   118
with
wenzelm@41683
   119
wenzelm@45136
   120
val dummy_task =
wenzelm@50975
   121
  Task {group = new_group NONE, name = "", id = 0, pri = NONE, timing = NONE,
wenzelm@50975
   122
    pos = Position.none};
wenzelm@41683
   123
wenzelm@41683
   124
fun new_task group name pri =
wenzelm@50975
   125
  Task {group = group, name = name, id = new_id (), pri = pri, timing = new_timing (),
wenzelm@50975
   126
    pos = Position.thread_data ()};
wenzelm@41683
   127
wenzelm@41683
   128
fun group_of_task (Task {group, ...}) = group;
wenzelm@41683
   129
fun name_of_task (Task {name, ...}) = name;
wenzelm@41683
   130
fun pri_of_task (Task {pri, ...}) = the_default 0 pri;
wenzelm@43951
   131
wenzelm@41683
   132
fun str_of_task (Task {name, id, ...}) =
wenzelm@41683
   133
  if name = "" then string_of_int id else string_of_int id ^ " (" ^ name ^ ")";
wenzelm@41683
   134
wenzelm@43951
   135
fun str_of_task_groups task = str_of_task task ^ " in " ^ str_of_groups (group_of_task task);
wenzelm@43951
   136
wenzelm@41683
   137
fun update_timing update (Task {timing, ...}) e =
wenzelm@41702
   138
  uninterruptible (fn restore_attributes => fn () =>
wenzelm@41702
   139
    let
wenzelm@41702
   140
      val start = Time.now ();
wenzelm@41702
   141
      val result = Exn.capture (restore_attributes e) ();
wenzelm@62826
   142
      val t = Time.now () - start;
wenzelm@45136
   143
      val _ = (case timing of NONE => () | SOME var => Synchronized.change var (update t));
wenzelm@41702
   144
    in Exn.release result end) ();
wenzelm@41683
   145
wenzelm@41683
   146
fun task_ord (Task {id = id1, pri = pri1, ...}, Task {id = id2, pri = pri2, ...}) =
wenzelm@41683
   147
  prod_ord (rev_order o option_ord int_ord) int_ord ((pri1, id1), (pri2, id2));
wenzelm@41683
   148
wenzelm@50975
   149
fun task_statistics (Task {name, id, timing, pos, ...}) =
wenzelm@50975
   150
  let
wenzelm@50975
   151
    val (run, wait, wait_deps) =
wenzelm@50975
   152
      (case timing of NONE => timing_start | SOME var => Synchronized.value var);
wenzelm@50975
   153
    fun micros time = string_of_int (Time.toNanoseconds time div 1000);
wenzelm@50975
   154
  in
wenzelm@51990
   155
    [("now", Markup.print_real (Time.toReal (Time.now ()))),
wenzelm@50975
   156
     ("task_name", name), ("task_id", Markup.print_int id),
wenzelm@50975
   157
     ("run", micros run), ("wait", micros wait), ("wait_deps", commas wait_deps)] @
wenzelm@50975
   158
    Position.properties_of pos
wenzelm@50975
   159
  end;
wenzelm@50975
   160
wenzelm@41683
   161
end;
wenzelm@41683
   162
wenzelm@41709
   163
structure Tasks = Table(type key = task val ord = task_ord);
wenzelm@41683
   164
structure Task_Graph = Graph(type key = task val ord = task_ord);
wenzelm@41683
   165
wenzelm@41683
   166
wenzelm@41695
   167
(* timing *)
wenzelm@41695
   168
wenzelm@41695
   169
fun running task =
wenzelm@62826
   170
  update_timing (fn t => fn (a, b, ds) => (a + t, b, ds)) task;
wenzelm@41695
   171
wenzelm@41695
   172
fun joining task =
wenzelm@62826
   173
  update_timing (fn t => fn (a, b, ds) => (a - t, b, ds)) task;
wenzelm@41695
   174
wenzelm@41695
   175
fun waiting task deps =
wenzelm@41695
   176
  update_timing (fn t => fn (a, b, ds) =>
wenzelm@62826
   177
    (a - t, b + t,
wenzelm@41702
   178
      if ! Multithreading.trace > 0
wenzelm@41702
   179
      then fold (insert (op =) o name_of_task) deps ds else ds)) task;
wenzelm@41695
   180
wenzelm@41695
   181
wenzelm@41683
   182
wenzelm@41681
   183
(** queue of jobs and groups **)
wenzelm@41681
   184
wenzelm@41709
   185
(* known group members *)
wenzelm@41709
   186
wenzelm@41709
   187
type groups = unit Tasks.table Inttab.table;
wenzelm@41709
   188
wenzelm@41709
   189
fun get_tasks (groups: groups) gid =
wenzelm@41709
   190
  the_default Tasks.empty (Inttab.lookup groups gid);
wenzelm@41709
   191
wenzelm@41709
   192
fun add_task (gid, task) groups =
wenzelm@41709
   193
  Inttab.update (gid, Tasks.update (task, ()) (get_tasks groups gid)) groups;
wenzelm@41709
   194
wenzelm@41709
   195
fun del_task (gid, task) groups =
wenzelm@41709
   196
  let val tasks = Tasks.delete_safe task (get_tasks groups gid) in
wenzelm@41709
   197
    if Tasks.is_empty tasks then Inttab.delete_safe gid groups
wenzelm@41709
   198
    else Inttab.update (gid, tasks) groups
wenzelm@41709
   199
  end;
wenzelm@41709
   200
wenzelm@41709
   201
wenzelm@41709
   202
(* job dependency graph *)
wenzelm@28165
   203
wenzelm@28165
   204
datatype job =
wenzelm@29365
   205
  Job of (bool -> bool) list |
wenzelm@34277
   206
  Running of Thread.thread |
wenzelm@37854
   207
  Passive of unit -> bool;
wenzelm@28165
   208
wenzelm@41683
   209
type jobs = job Task_Graph.T;
wenzelm@28176
   210
wenzelm@41683
   211
fun get_job (jobs: jobs) task = Task_Graph.get_node jobs task;
wenzelm@41683
   212
fun set_job task job (jobs: jobs) = Task_Graph.map_node task (K job) jobs;
wenzelm@28202
   213
wenzelm@32250
   214
fun add_job task dep (jobs: jobs) =
wenzelm@32250
   215
  Task_Graph.add_edge (dep, task) jobs handle Task_Graph.UNDEF _ => jobs;
wenzelm@32250
   216
wenzelm@28176
   217
wenzelm@41681
   218
(* queue *)
wenzelm@28176
   219
wenzelm@60610
   220
datatype queue = Queue of {groups: groups, jobs: jobs, urgent: int};
wenzelm@28165
   221
wenzelm@60610
   222
fun make_queue groups jobs urgent = Queue {groups = groups, jobs = jobs, urgent = urgent};
wenzelm@60610
   223
val empty = make_queue Inttab.empty Task_Graph.empty 0;
wenzelm@34277
   224
wenzelm@47404
   225
fun group_tasks (Queue {groups, ...}) group = Tasks.keys (get_tasks groups (group_id group));
wenzelm@41708
   226
fun known_task (Queue {jobs, ...}) task = can (Task_Graph.get_entry jobs) task;
wenzelm@41708
   227
wenzelm@37854
   228
wenzelm@37854
   229
(* job status *)
wenzelm@37854
   230
wenzelm@59332
   231
fun ready_job (task, (Job list, (deps, _))) =
wenzelm@44338
   232
      if Task_Graph.Keys.is_empty deps then SOME (task, rev list) else NONE
wenzelm@59332
   233
  | ready_job (task, (Passive abort, (deps, _))) =
wenzelm@44338
   234
      if Task_Graph.Keys.is_empty deps andalso is_canceled (group_of_task task)
wenzelm@44338
   235
      then SOME (task, [fn _ => abort ()])
wenzelm@37854
   236
      else NONE
wenzelm@59332
   237
  | ready_job _ = NONE;
wenzelm@37854
   238
wenzelm@60610
   239
fun ready_job_urgent false = ready_job
wenzelm@60610
   240
  | ready_job_urgent true = (fn entry as (task, _) =>
wenzelm@60610
   241
      if pri_of_task task >= urgent_pri then ready_job entry else NONE);
wenzelm@60610
   242
wenzelm@59333
   243
fun active_job (task, (Running _, _)) = SOME (task, [])
wenzelm@59333
   244
  | active_job arg = ready_job arg;
wenzelm@37854
   245
wenzelm@41683
   246
fun all_passive (Queue {jobs, ...}) = is_none (Task_Graph.get_first active_job jobs);
wenzelm@28165
   247
wenzelm@28165
   248
wenzelm@32101
   249
(* queue status *)
wenzelm@32052
   250
wenzelm@60610
   251
fun status (Queue {jobs, urgent, ...}) =
wenzelm@32052
   252
  let
wenzelm@34277
   253
    val (x, y, z, w) =
wenzelm@41683
   254
      Task_Graph.fold (fn (_, (job, (deps, _))) => fn (x, y, z, w) =>
wenzelm@32052
   255
          (case job of
wenzelm@44338
   256
            Job _ => if Task_Graph.Keys.is_empty deps then (x + 1, y, z, w) else (x, y + 1, z, w)
wenzelm@34277
   257
          | Running _ => (x, y, z + 1, w)
wenzelm@37854
   258
          | Passive _ => (x, y, z, w + 1)))
wenzelm@34277
   259
        jobs (0, 0, 0, 0);
wenzelm@60610
   260
  in {ready = x, pending = y, running = z, passive = w, urgent = urgent} end;
wenzelm@32052
   261
wenzelm@32052
   262
wenzelm@41681
   263
wenzelm@41681
   264
(** task queue operations **)
wenzelm@41681
   265
wenzelm@32101
   266
(* cancel -- peers and sub-groups *)
wenzelm@32101
   267
wenzelm@60610
   268
fun cancel (Queue {groups, jobs, ...}) group =
wenzelm@32101
   269
  let
wenzelm@32101
   270
    val _ = cancel_group group Exn.Interrupt;
wenzelm@44341
   271
    val running =
wenzelm@47404
   272
      Tasks.fold (fn (task, _) =>
wenzelm@47404
   273
          (case get_job jobs task of Running thread => insert Thread.equal thread | _ => I))
wenzelm@47404
   274
        (get_tasks groups (group_id group)) [];
wenzelm@44341
   275
  in running end;
wenzelm@32101
   276
wenzelm@37854
   277
fun cancel_all (Queue {jobs, ...}) =
wenzelm@32101
   278
  let
wenzelm@41683
   279
    fun cancel_job (task, (job, _)) (groups, running) =
wenzelm@41683
   280
      let
wenzelm@41683
   281
        val group = group_of_task task;
wenzelm@41683
   282
        val _ = cancel_group group Exn.Interrupt;
wenzelm@41683
   283
      in
wenzelm@34277
   284
        (case job of
wenzelm@34277
   285
          Running t => (insert eq_group group groups, insert Thread.equal t running)
wenzelm@41683
   286
        | _ => (groups, running))
wenzelm@41683
   287
      end;
wenzelm@44341
   288
    val running = Task_Graph.fold cancel_job jobs ([], []);
wenzelm@44341
   289
  in running end;
wenzelm@32101
   290
wenzelm@32101
   291
wenzelm@41681
   292
(* finish *)
wenzelm@41681
   293
wenzelm@60610
   294
fun finish task (Queue {groups, jobs, urgent}) =
wenzelm@41681
   295
  let
wenzelm@41683
   296
    val group = group_of_task task;
wenzelm@43951
   297
    val groups' = fold_groups (fn g => del_task (group_id g, task)) group groups;
wenzelm@41681
   298
    val jobs' = Task_Graph.del_node task jobs;
wenzelm@44338
   299
    val maximal = Task_Graph.is_maximal jobs task;
wenzelm@60610
   300
  in (maximal, make_queue groups' jobs' urgent) end;
wenzelm@41681
   301
wenzelm@41681
   302
wenzelm@54649
   303
(* enroll *)
wenzelm@54649
   304
wenzelm@60610
   305
fun enroll thread name group (Queue {groups, jobs, urgent}) =
wenzelm@54649
   306
  let
wenzelm@54649
   307
    val task = new_task group name NONE;
wenzelm@54649
   308
    val groups' = fold_groups (fn g => add_task (group_id g, task)) group groups;
wenzelm@54649
   309
    val jobs' = jobs |> Task_Graph.new_node (task, Running thread);
wenzelm@60610
   310
  in (task, make_queue groups' jobs' urgent) end;
wenzelm@54649
   311
wenzelm@54649
   312
wenzelm@28185
   313
(* enqueue *)
wenzelm@28165
   314
wenzelm@60610
   315
fun enqueue_passive group abort (Queue {groups, jobs, urgent}) =
wenzelm@34277
   316
  let
wenzelm@41683
   317
    val task = new_task group "passive" NONE;
wenzelm@43951
   318
    val groups' = fold_groups (fn g => add_task (group_id g, task)) group groups;
wenzelm@41683
   319
    val jobs' = jobs |> Task_Graph.new_node (task, Passive abort);
wenzelm@60610
   320
  in (task, make_queue groups' jobs' urgent) end;
wenzelm@34277
   321
wenzelm@60610
   322
fun enqueue name group deps pri job (Queue {groups, jobs, urgent}) =
wenzelm@28165
   323
  let
wenzelm@41683
   324
    val task = new_task group name (SOME pri);
wenzelm@43951
   325
    val groups' = fold_groups (fn g => add_task (group_id g, task)) group groups;
wenzelm@28185
   326
    val jobs' = jobs
wenzelm@41683
   327
      |> Task_Graph.new_node (task, Job [job])
wenzelm@41684
   328
      |> fold (add_job task) deps;
wenzelm@60610
   329
    val urgent' = if pri >= urgent_pri then urgent + 1 else urgent;
wenzelm@60610
   330
  in (task, make_queue groups' jobs' urgent') end;
wenzelm@28165
   331
wenzelm@60610
   332
fun extend task job (Queue {groups, jobs, urgent}) =
wenzelm@29365
   333
  (case try (get_job jobs) task of
wenzelm@60610
   334
    SOME (Job list) => SOME (make_queue groups (set_job task (Job (job :: list)) jobs) urgent)
wenzelm@29365
   335
  | _ => NONE);
wenzelm@29365
   336
wenzelm@28185
   337
wenzelm@28185
   338
(* dequeue *)
wenzelm@28185
   339
wenzelm@60610
   340
fun dequeue_passive thread task (queue as Queue {groups, jobs, urgent}) =
wenzelm@39243
   341
  (case try (get_job jobs) task of
wenzelm@39243
   342
    SOME (Passive _) =>
wenzelm@39243
   343
      let val jobs' = set_job task (Running thread) jobs
wenzelm@60610
   344
      in (SOME true, make_queue groups jobs' urgent) end
wenzelm@47423
   345
  | SOME _ => (SOME false, queue)
wenzelm@47423
   346
  | NONE => (NONE, queue));
wenzelm@39243
   347
wenzelm@60610
   348
fun dequeue thread urgent_only (queue as Queue {groups, jobs, urgent}) =
wenzelm@60610
   349
  if not urgent_only orelse urgent > 0 then
wenzelm@60610
   350
    (case Task_Graph.get_first (ready_job_urgent urgent_only) jobs of
wenzelm@60610
   351
      SOME (result as (task, _)) =>
wenzelm@60610
   352
        let
wenzelm@60610
   353
          val jobs' = set_job task (Running thread) jobs;
wenzelm@60610
   354
          val urgent' = if pri_of_task task >= urgent_pri then urgent - 1 else urgent;
wenzelm@60610
   355
        in (SOME result, make_queue groups jobs' urgent') end
wenzelm@60610
   356
    | NONE => (NONE, queue))
wenzelm@60610
   357
  else (NONE, queue);
wenzelm@28202
   358
wenzelm@28165
   359
wenzelm@41681
   360
(* dequeue wrt. dynamic dependencies *)
wenzelm@41681
   361
wenzelm@60610
   362
fun dequeue_deps thread deps (queue as Queue {groups, jobs, urgent}) =
wenzelm@32055
   363
  let
wenzelm@41684
   364
    fun ready [] rest = (NONE, rev rest)
wenzelm@41684
   365
      | ready (task :: tasks) rest =
wenzelm@41684
   366
          (case try (Task_Graph.get_entry jobs) task of
wenzelm@41684
   367
            NONE => ready tasks rest
wenzelm@43792
   368
          | SOME (_, entry) =>
wenzelm@59332
   369
              (case ready_job (task, entry) of
wenzelm@41684
   370
                NONE => ready tasks (task :: rest)
wenzelm@45642
   371
              | some => (some, fold cons rest tasks)));
wenzelm@41684
   372
wenzelm@41684
   373
    fun ready_dep _ [] = NONE
wenzelm@41684
   374
      | ready_dep seen (task :: tasks) =
wenzelm@41709
   375
          if Tasks.defined seen task then ready_dep seen tasks
wenzelm@41684
   376
          else
wenzelm@43792
   377
            let val entry as (_, (ds, _)) = #2 (Task_Graph.get_entry jobs task) in
wenzelm@59332
   378
              (case ready_job (task, entry) of
wenzelm@44338
   379
                NONE => ready_dep (Tasks.update (task, ()) seen) (Task_Graph.Keys.dest ds @ tasks)
wenzelm@41684
   380
              | some => some)
wenzelm@41684
   381
            end;
wenzelm@41684
   382
wenzelm@41684
   383
    fun result (res as (task, _)) deps' =
wenzelm@60610
   384
      let
wenzelm@60610
   385
        val jobs' = set_job task (Running thread) jobs;
wenzelm@60610
   386
        val urgent' = if pri_of_task task >= urgent_pri then urgent - 1 else urgent;
wenzelm@60610
   387
      in ((SOME res, deps'), make_queue groups jobs' urgent') end;
wenzelm@32055
   388
  in
wenzelm@41684
   389
    (case ready deps [] of
wenzelm@41684
   390
      (SOME res, deps') => result res deps'
wenzelm@41684
   391
    | (NONE, deps') =>
wenzelm@41709
   392
        (case ready_dep Tasks.empty deps' of
wenzelm@41684
   393
          SOME res => result res deps'
wenzelm@41695
   394
        | NONE => ((NONE, deps'), queue)))
wenzelm@32055
   395
  end;
wenzelm@32055
   396
wenzelm@62663
   397
wenzelm@62663
   398
(* toplevel pretty printing *)
wenzelm@62663
   399
wenzelm@62819
   400
val _ = ML_system_pp (fn _ => fn _ => Pretty.to_polyml o Pretty.str o str_of_task);
wenzelm@62819
   401
val _ = ML_system_pp (fn _ => fn _ => Pretty.to_polyml o Pretty.str o str_of_group);
wenzelm@62663
   402
wenzelm@41681
   403
end;