wneuper/isa: comparison src/Pure/Concurrent/future.ML

equal deleted inserted replaced

-:7a44005dc2ec
+:058520fa03a8
 val worker_subgroup: unit -> Task_Queue.group
 type 'a future
 val task_of: 'a future -> Task_Queue.task
 val peek: 'a future -> 'a Exn.result option
 val is_finished: 'a future -> bool
+val cancel_group: Task_Queue.group -> unit
+val cancel: 'a future -> unit
 val forks:
 {name: string, group: Task_Queue.group option, deps: Task_Queue.task list, pri: int} ->
 (unit -> 'a) list -> 'a future list
 val fork_pri: int -> (unit -> 'a) -> 'a future
 val fork: (unit -> 'a) -> 'a future
 val promise_group: Task_Queue.group -> 'a future
 val promise: unit -> 'a future
 val fulfill_result: 'a future -> 'a Exn.result -> unit
 val fulfill: 'a future -> 'a -> unit
 val interruptible_task: ('a -> 'b) -> 'a -> 'b
-val cancel_group: Task_Queue.group -> unit
-val cancel: 'a future -> unit
 val shutdown: unit -> unit
 val status: (unit -> 'a) -> 'a
 end;
 structure Future: FUTURE =
 local
 val tag = Universal.tag () : Task_Queue.task option Universal.tag;
 in
 fun worker_task () = the_default NONE (Thread.getLocal tag);
-fun setmp_worker_task data f x =
+fun setmp_worker_task task f x = setmp_thread_data tag (worker_task ()) (SOME task) f x;
-Library.setmp_thread_data tag (worker_task ()) (SOME data) f x;
 end;
 val worker_group = Option.map Task_Queue.group_of_task o worker_task;
 fun worker_subgroup () = Task_Queue.new_group (worker_group ());
 fun task_of (Future {task, ...}) = task;
 fun result_of (Future {result, ...}) = result;
 fun peek x = Single_Assignment.peek (result_of x);
 fun is_finished x = is_some (peek x);
-fun assign_result group result res =
-let
-val _ = Single_Assignment.assign result res
-handle exn as Fail _ =>
-(case Single_Assignment.peek result of
-SOME (Exn.Exn e) => reraise (if Exn.is_interrupt e then e else exn)
-| _ => reraise exn);
-val ok =
-(case the (Single_Assignment.peek result) of
-Exn.Exn exn => (Task_Queue.cancel_group group exn; false)
-| Exn.Res _ => true);
-in ok end;
 (** scheduling **)
 fun count_workers state = (*requires SYNCHRONIZED*)
 fold (fn (_, state_ref) => fn i => if ! state_ref = state then i + 1 else i) (! workers) 0;
-(* execute future jobs *)
+(* cancellation primitives *)
-fun future_job group (e: unit -> 'a) =
+fun interruptible_task f x =
-let
+if Multithreading.available then
-val result = Single_Assignment.var "future" : 'a result;
+Multithreading.with_attributes
-val pos = Position.thread_data ();
+(if is_some (worker_task ())
-fun job ok =
+then Multithreading.private_interrupts
-let
+else Multithreading.public_interrupts)
-val res =
+(fn _ => f x)
-if ok then
+else interruptible f x;
-Exn.capture (fn () =>
-Multithreading.with_attributes Multithreading.private_interrupts
-(fn _ => Position.setmp_thread_data pos e ()) before
-Multithreading.interrupted ()) ()
-else Exn.interrupt_exn;
-in assign_result group result res end;
-in (result, job) end;
 fun cancel_now group = (*requires SYNCHRONIZED*)
 Task_Queue.cancel (! queue) group;
 fun cancel_later group = (*requires SYNCHRONIZED*)
 (Unsynchronized.change canceled (insert Task_Queue.eq_group group);
 broadcast scheduler_event);
-fun execute (task, jobs) =
+(* worker threads *)
+fun worker_exec (task, jobs) =
 let
 val group = Task_Queue.group_of_task task;
 val valid = not (Task_Queue.is_canceled group);
 val ok =
 Task_Queue.running task (fn () =>
 val _ = broadcast work_finished;
 val _ = if maximal then () else signal work_available;
 in () end);
 in () end;
-(* worker threads *)
 fun worker_wait active cond = (*requires SYNCHRONIZED*)
 let
 val state =
 (case AList.lookup Thread.equal (! workers) (Thread.self ()) of
 SOME state => state
 | some => (signal work_available; some));
 fun worker_loop name =
 (case SYNCHRONIZED name (fn () => worker_next ()) of
 NONE => ()
-| SOME work => (Exn.capture Multithreading.interrupted (); execute work; worker_loop name));
+| SOME work => (Exn.capture Multithreading.interrupted (); worker_exec work; worker_loop name));
 fun worker_start name = (*requires SYNCHRONIZED*)
 Unsynchronized.change workers (cons (Simple_Thread.fork false (fn () => worker_loop name),
 Unsynchronized.ref Working));
 else scheduler := SOME (Simple_Thread.fork false scheduler_loop));
 (** futures **)
+(* cancellation *)
+(*cancel: present and future group members will be interrupted eventually*)
+fun cancel_group group = SYNCHRONIZED "cancel" (fn () =>
+(if cancel_now group then () else cancel_later group;
+signal work_available; scheduler_check ()));
+fun cancel x = cancel_group (Task_Queue.group_of_task (task_of x));
+(* future jobs *)
+fun assign_result group result res =
+let
+val _ = Single_Assignment.assign result res
+handle exn as Fail _ =>
+(case Single_Assignment.peek result of
+SOME (Exn.Exn e) => reraise (if Exn.is_interrupt e then e else exn)
+| _ => reraise exn);
+val ok =
+(case the (Single_Assignment.peek result) of
+Exn.Exn exn => (Task_Queue.cancel_group group exn; false)
+| Exn.Res _ => true);
+in ok end;
+fun future_job group (e: unit -> 'a) =
+let
+val result = Single_Assignment.var "future" : 'a result;
+val pos = Position.thread_data ();
+fun job ok =
+let
+val res =
+if ok then
+Exn.capture (fn () =>
+Multithreading.with_attributes Multithreading.private_interrupts
+(fn _ => Position.setmp_thread_data pos e ()) before
+Multithreading.interrupted ()) ()
+else Exn.interrupt_exn;
+in assign_result group result res end;
+in (result, job) end;
 (* fork *)
 fun forks {name, group, deps, pri} es =
 if null es then []
 | (NONE, deps') =>
 (worker_waiting deps' (fn () => worker_wait true work_finished); join_next deps')
 | (SOME work, deps') => SOME (work, deps'));
 fun execute_work NONE = ()
-| execute_work (SOME (work, deps')) = (worker_joining (fn () => execute work); join_work deps')
+| execute_work (SOME (work, deps')) =
+(worker_joining (fn () => worker_exec work); join_work deps')
 and join_work deps =
 Multithreading.with_attributes Multithreading.no_interrupts
 (fn _ => execute_work (SYNCHRONIZED "join" (fn () => join_next deps)));
 in
 fun join_result x = singleton join_results x;
 fun join x = Exn.release (join_result x);
-(* fast-path versions -- bypassing full task management *)
+(* fast-path versions -- bypassing task queue *)
 fun value (x: 'a) =
 let
 val task = Task_Queue.dummy_task ();
 val group = Task_Queue.group_of_task task;
 let
 val still_passive =
 SYNCHRONIZED "fulfill_result" (fn () =>
 Unsynchronized.change_result queue
 (Task_Queue.dequeue_passive (Thread.self ()) task));
-in if still_passive then execute (task, [job]) else () end);
+in if still_passive then worker_exec (task, [job]) else () end);
 val _ =
 if is_some (Single_Assignment.peek result) then ()
 else worker_waiting [task] (fn () => ignore (Single_Assignment.await result));
 in () end;
 fun fulfill x res = fulfill_result x (Exn.Res res);
-(* cancellation *)
-fun interruptible_task f x =
-if Multithreading.available then
-Multithreading.with_attributes
-(if is_some (worker_task ())
-then Multithreading.private_interrupts
-else Multithreading.public_interrupts)
-(fn _ => f x)
-else interruptible f x;
-(*cancel: present and future group members will be interrupted eventually*)
-fun cancel_group group = SYNCHRONIZED "cancel" (fn () =>
-(if cancel_now group then () else cancel_later group;
-signal work_available; scheduler_check ()));
-fun cancel x = cancel_group (Task_Queue.group_of_task (task_of x));
 (* shutdown *)
 fun shutdown () =

changeset 44993	058520fa03a8
parent 44824	318ca53e3fbb
child 44994	2d16c693d536