(*  Title:      Pure/ML-Systems/multithreading_polyml.ML

     ID:         $Id$

     Author:     Makarius

 Multithreading in Poly/ML (version 5.1).

 *)

 open Thread;

 structure Multithreading: MULTITHREADING =

 struct

 val trace = ref false;

 fun tracing msg =

   if ! trace

   then (TextIO.output (TextIO.stdErr, (">>> " ^ msg () ^ "\n")); TextIO.flushOut TextIO.stdErr)

   else ();

 val available = true;

 val max_threads = ref 1;

 (* critical section -- may be nested within the same thread *)

 local

 fun add_name "" = ""

   | add_name name = " " ^ name;

 fun add_name' "" = ""

   | add_name' name = " [" ^ name ^ "]";

 val critical_lock = Mutex.mutex ();

 val critical_thread = ref (NONE: Thread.thread option);

 val critical_name = ref "";

 in

 fun self_critical () =

   (case ! critical_thread of

     NONE => false

   | SOME id => Thread.equal (id, Thread.self ()));

 fun NAMED_CRITICAL name e =

   if self_critical () then e ()

   else

     let

       val _ =

         if Mutex.trylock critical_lock then ()

         else

           let

             val timer = Timer.startRealTimer ();

             val _ = tracing (fn () =>

               "CRITICAL" ^ add_name name ^ add_name' (! critical_name) ^ ": waiting for lock");

             val _ = Mutex.lock critical_lock;

             val _ = tracing (fn () =>

               "CRITICAL" ^ add_name name ^ add_name' (! critical_name) ^ ": obtained lock after " ^

               Time.toString (Timer.checkRealTimer timer));

           in () end;

       val _ = critical_thread := SOME (Thread.self ());

       val _ = critical_name := name;

       val result = Exn.capture e ();

       val _ = critical_name := "";

       val _ = critical_thread := NONE;

       val _ = Mutex.unlock critical_lock;

     in Exn.release result end;

 fun CRITICAL e = NAMED_CRITICAL "" e;

 end;

 (* scheduling -- non-interruptible threads working on a queue of tasks *)

 fun inc i = (i := ! i + 1; ! i);

 fun dec i = (i := ! i - 1; ! i);

 fun schedule n next_task tasks =

   let

     (*protected execution*)

     val lock = Mutex.mutex ();

     fun PROTECTED name e =

       let

         val _ =

           if Mutex.trylock lock then ()

           else

            (tracing (fn () => "PROTECTED " ^ name ^ ": waiting for lock");

             Mutex.lock lock;

             tracing (fn () => "PROTECTED " ^ name ^ ": obtained lock"));

         val res = Exn.capture e ();

         val _ = Mutex.unlock lock;

       in Exn.release res end;

     (*the queue of tasks*)

     val queue = ref tasks;

     fun dequeue () = PROTECTED "dequeue" (fn () =>

       let

         val (next, tasks') = next_task (! queue);

         val _ = queue := tasks';

       in next end);

     (*worker threads*)

     val running = ref 0;

     val active = ref 0;

     val status = ref ([]: exn list);

     val wakeup = ConditionVar.conditionVar ();

     fun wait () = ConditionVar.wait (wakeup, lock);

     fun continue cont k =

       (PROTECTED "cont" (fn () => queue := cont (! queue));

        ConditionVar.broadcast wakeup; work k ())

     and work k () =

       (case dequeue () of

         (Task.Task f, cont) =>

           (case Exn.capture f () of

             Exn.Result () => continue cont k

           | Exn.Exn exn =>

               (PROTECTED "status" (fn () => status := exn :: ! status); continue cont k))

       | (Task.Running, _) =>

           (PROTECTED "wait" (fn () => (dec active; wait (); inc active)); work k ())

       | (Task.Finished, _) =>

          (PROTECTED "running" (fn () => (dec active; dec running));

           ConditionVar.broadcast wakeup));

     (*main control: fork and wait*)

     fun fork 0 = ()

       | fork k =

          (inc running; inc active;

           Thread.fork (work k, [Thread.InterruptState Thread.InterruptDefer]);

           fork (k - 1));

     val _ = PROTECTED "main" (fn () =>

      (fork (Int.max (n, 1));

       while ! running <> 0 do

         (tracing (fn () => "MAIN: " ^ Int.toString (! active) ^ " active");

          wait ())));

   in ! status end;

 end;

 val NAMED_CRITICAL = Multithreading.NAMED_CRITICAL;

 val CRITICAL = Multithreading.CRITICAL;