(* Copyright (C) 1989, Digital Equipment Corporation *)
(* All rights reserved.                              *)
(* See the file COPYRIGHT for a full description.    *)

(* Last Modified On Fri Jul 20 09:45:30 1990 by muller *)

UNSAFE MODULE Thread EXPORTS Thread, ThreadF;

IMPORT Word, ThreadSup, Time;
IMPORT CoreOS, Cstdlib;

FROM ThreadSup IMPORT self, inCritical, contextSize;

TYPE
  State = {
        alive	 (* The thread is running and does not wait *),
        waiting	 (* The thread is waiting for a condition via Wait *),
	locking	 (* The thread is waiting for a mutex to be unlocked *),
        pausing	 (* The thread is waiting until some time is arrived *),
	dying	 (* The thread has finished its task, but has not
			      been joined so far *),
	dead 	 (* The thread has finished its task and has been joined *)
	};

  Stack = ARRAY OF Word.T;
  Context = ARRAY OF Word.T;

REVEAL
  (* Remember, the report (p 43-44) says that MUTEX is predeclared and <: ROOT;
     just pretend that we have
	  TYPE
	    MUTEX <: ROOT
     in our interface.
     The sem field is where we store the semaphore that implements the mutual
     exclusion and waitingForMe is the head of the list of threads that are
     waiting for the mutex to be released so that they can acquire it (the list
     is continued in the nextWaitingForMutex field of the threads) *)

  MUTEX = BRANDED "Mutex 1.0" OBJECT
            sem : INTEGER;
            waitingForMe: T; END; 


  (* Threads that wait on a condition are inserted in the waitingForMe list,
     which is continued in the nextWaitingForCondition field of the waiting
     threads. *)

  Condition = BRANDED "Thread.Condition v1.0" OBJECT
                waitingForMe: T; END;


  (* Finally, here is the thread *)
  T = Mutex BRANDED "Thread.T v1.3" OBJECT
        state: State;

        (* our work and its result *)
        closure : Closure;
        result : REFANY;

        (* where we carry our work. The first thread runs on the
           original stack and stack is NIL *)
        stack : UNTRACED REF Stack;
        context : UNTRACED REF Context;

        (* the threads are organized in a circular list *)
        previous, next: T;

        (* the running threads are organized in a doubly linked. If
           self is not running, then self.nextRunning is one of the
           running threads is there are any; otherwise, it is a thread
           with state # alive (maybe itself) *)
	nextRunning, previousRunning: T;

        (* CASE state OF
           | waiting => the next thread that waits for the same condition;
           | locking => the next thread that waits for the same mutex;
           | pausing => the next thread that waits for specified time; *)
        nextWaiting: T;

        (* if state = waiting, the condition on which we wait *)
        waitingForCondition: Condition;

	(* if state = pausing, the time at which we can restart *)
        waitingForTime : Time.T;

        (* true if we are waiting during an AlertWait or AlertJoin *)
        alertable: BOOLEAN ;

        (* true if somebody alerted us and we did not TestAlert *)
        alertPending : BOOLEAN;

        (* This condition is signaled then the thread terminates;
           other threads that want to join can just wait for it *)
        endCondition: Condition; END;

VAR
  (* we start the heavy machinery only when we have more than one thread *)
  multipleThreads: BOOLEAN := FALSE;

  topThread: T;
  pausedThreads : T;
  selected_interval:= CoreOS.struct_timeval {0, 100 * 1000};

  stackPool := ARRAY [0..2000] OF UNTRACED REF Stack {NIL, ..};
  stackPoolSize := 0;
  contextPool:= ARRAY [0..2000] OF UNTRACED REF Context {NIL, ..};
  contextPoolSize := 0;

EXCEPTION
  DeadThread; InternalError;

PROCEDURE NewMutex (): Mutex =
BEGIN
  RETURN (NEW (Mutex));
END NewMutex;

PROCEDURE NewCondition (): Condition =
BEGIN
  RETURN (NEW (Condition));
END NewCondition;


PROCEDURE CanRun (t: T) = 
BEGIN
  IF t = self THEN
    IF self.nextRunning.state = State.alive THEN
      self.previousRunning := self.nextRunning.previousRunning;
      self.nextRunning.previousRunning.nextRunning := self;
      self.nextRunning.previousRunning := self;
    ELSE
      self.nextRunning := self;
      self.previousRunning := self; END;
  ELSIF self.state = State.alive THEN
    self.nextRunning.previousRunning := t;
    t.nextRunning := self.nextRunning;
    self.nextRunning := t;
    t.previousRunning := self;
  ELSIF self.nextRunning.state = State.alive THEN
    self.nextRunning.previousRunning.nextRunning := t;
    t.previousRunning := self.nextRunning.previousRunning;
    t.nextRunning := self.nextRunning;
    self.nextRunning.previousRunning := t;
  ELSE
    self.nextRunning := t;
    t.previousRunning := t;
   t.nextRunning := t; END;
  t.state := State.alive; 
END CanRun;

PROCEDURE ICannotRun (newState: State) = 
BEGIN
  self.state := newState;
  self.nextRunning.previousRunning := self.previousRunning;
  self.previousRunning.nextRunning := self.nextRunning;
END ICannotRun;

PROCEDURE Fork (cl: Closure): T =
VAR
  t: T; 
  size: CARDINAL;
BEGIN
  INC (inCritical);
  IF NOT multipleThreads THEN
    (* this is the first time we have more than one thread; we can start to
       consider switching *)
    multipleThreads := TRUE;
    StartSwitching (); END;
  t := NEW (T, closure := cl, result := NIL, alertPending := FALSE);
  IF contextPoolSize > 0 THEN
    DEC (contextPoolSize);
    t.context := contextPool [contextPoolSize];
  ELSE 
    t.context := NEW (UNTRACED REF Context, contextSize); END;
  TYPECASE cl OF
    | SizedClosure (scl) => size := scl.stackSize;
    ELSE                    size := DefaultStackSize; END;
  IF size = DefaultStackSize AND stackPoolSize > 0 THEN
    DEC (stackPoolSize);
    t.stack := stackPool [stackPoolSize];
  ELSE
    t.stack := NEW (UNTRACED REF Stack, size); END;
  t.endCondition := NewCondition ();
  t.next := self.next;
  t.previous := self;
  self.next.previous := t;
  self.next := t;
  ThreadSup.InitContext (t, ADR (t.context [0]),
                             ADR (t.stack [0]), NUMBER (t.stack^));
  CanRun (t);
  DEC (inCritical);
  RETURN t;
END Fork;

PROCEDURE Join (t: T): REFANY =
BEGIN
  INC (inCritical);
  TRY
    RETURN XJoin (t, FALSE);
  FINALLY
    DEC (inCritical); END;
END Join;

PROCEDURE AlertJoin (t: T): REFANY RAISES {DeadThread, Alerted} =
BEGIN
  INC (inCritical);
  TRY
    RETURN XJoin (t, TRUE);
  FINALLY
    DEC (inCritical); END;
END AlertJoin;

PROCEDURE XJoin (t: T; alertable: BOOLEAN): REFANY
     RAISES {DeadThread, Alerted} =
BEGIN
  LOCK t DO
    WHILE (t.state # State.dying) AND (t.state # State.dead) DO
      XWait (t, t.endCondition, alertable); END;
    CASE t.state OF
    | State.dying => t.state := State.dead; RETURN t.result;
    | State.dead  => RAISE DeadThread; 
    ELSE             <* ASSERT FALSE *> END; END;
END XJoin;

PROCEDURE Wait (m: Mutex; c: Condition) =
BEGIN
  INC (inCritical);
  TRY
    XWait (m, c, FALSE);
  FINALLY
    DEC (inCritical); END;
END Wait;

PROCEDURE AlertWait (m: Mutex; c: Condition) RAISES {Alerted} =
BEGIN
  INC (inCritical);
  TRY
    XWait (m, c, TRUE);
  FINALLY
    DEC (inCritical); END;
END AlertWait;

<*INLINE*>
PROCEDURE XWait (m: Mutex; c: Condition; alertable: BOOLEAN) RAISES {Alerted} =
BEGIN
  IF alertable AND self.alertPending THEN
    self.alertPending := FALSE;
    RAISE Alerted; END;
  Release (m);
  ICannotRun (State.waiting);
  self.nextWaiting := c.waitingForMe;
  c.waitingForMe := self;
  self.waitingForCondition := c;
  self.alertable := alertable;
  Yield ();
  INC (inCritical);
  Acquire (m);
  IF alertable AND self.alertPending THEN
    self.alertPending := FALSE;
    RAISE Alerted; END;
END XWait;

PROCEDURE Signal (c: Condition) =
BEGIN
  INC (inCritical);
  TRY
    XSignal (c, 1);
  FINALLY
    DEC (inCritical); END;
END Signal;

PROCEDURE Broadcast (c: Condition) =
BEGIN
  INC (inCritical);
  TRY
    XSignal (c, -1);
  FINALLY
    DEC (inCritical); END;
END Broadcast;

<*INLINE*>
PROCEDURE XSignal (c: Condition; limit: INTEGER) =
VAR
  t: T;
BEGIN
  WHILE (c.waitingForMe # NIL) DO
    t := c.waitingForMe;
    c.waitingForMe := t.nextWaiting;
    t.nextWaiting := NIL;
    t.waitingForCondition := NIL;
    CanRun (t);
    DEC (limit);
    IF limit = 0 THEN
      RETURN ; END; END;
END XSignal;

PROCEDURE Alert (t: T) =
BEGIN
  INC (inCritical);
  t.alertPending := TRUE;
  IF t.state = State.waiting AND t.alertable THEN
    (* Remove the thread from the waiting list and make it runnable *)
    WITH c = t.waitingForCondition DO
      IF c.waitingForMe = t THEN
        c.waitingForMe := t.nextWaiting;
      ELSE
        VAR tt: T := c.waitingForMe; BEGIN
          WHILE tt.nextWaiting # t DO tt := tt.nextWaiting; END;
          tt.nextWaiting := t.nextWaiting; END; END;
        t.nextWaiting := NIL;
        t.waitingForCondition := NIL; END;
    CanRun (t); END;
  DEC (inCritical);
END Alert;

PROCEDURE TestAlert (): BOOLEAN =
VAR
  result: BOOLEAN;
BEGIN
  INC (inCritical);
  result := self.alertPending;
  self.alertPending := FALSE;
  DEC (inCritical);
  RETURN result; 
END TestAlert;

PROCEDURE Shell () =
BEGIN
  DEC (inCritical);
  allow_sigprof ();
  self.result := self.closure.apply ();
  INC (inCritical);
  self.next.previous := self.previous;
  self.previous.next := self.next;
  self.previous := NIL;
  self.next := NIL;
  IF stackPoolSize < NUMBER (stackPool) AND
     NUMBER (self.stack^) = DefaultStackSize THEN
    stackPool [stackPoolSize] := self.stack;
    INC (stackPoolSize);
  ELSE
    DISPOSE (self.stack); END;
  IF contextPoolSize < NUMBER (contextPool) THEN
    contextPool [contextPoolSize] := self.context;
    INC (contextPoolSize);
  ELSE
    DISPOSE (self.context); END;
  Broadcast (self.endCondition);
  ICannotRun (State.dying);
  Yield ();
END Shell;

PROCEDURE Yield () =
VAR
  t: T;
BEGIN
  LOOP
    t := self;
    REPEAT
      t := t.nextRunning;
      UNTIL t.state = State.alive OR t = self;
    IF t.state = State.alive THEN
      (* At least one thread wants to run; transfer to it *)
      IF t # self THEN
        disallow_sigprof ();
        ThreadSup.Transfer (ADR (t.context [0])); 
        allow_sigprof (); END;
      DEC (inCritical);
      RETURN;

    ELSIF pausedThreads # NIL THEN
      (* no thread is ready yet, but some are paused *)
      (* figure how long it will be before one of them should be awakened *)

      DelayProfTimer (Time.Subtract (pausedThreads.waitingForTime,
			             Time.Now ()));

    ELSE
      (* looks like everybody has finished, go back to the original context *)
      (* we may have a deadlock if some threads are waiting for conditions or
         mutexes *)
      Cstdlib.exit (0); END; END;

END Yield;

<*INLINE*>
PROCEDURE Acquire (m: Mutex) =
BEGIN
  LOOP
    INC (inCritical);
    IF m.sem = 0 THEN
      m.sem := 1;
      DEC (inCritical);
      RETURN ; END;
    ICannotRun (State.locking);
    self.nextWaiting := m.waitingForMe;
    m.waitingForMe := self;
    Yield (); END;
END Acquire;

<*INLINE*>
PROCEDURE Release (m: Mutex) =
VAR
  t: T;
BEGIN
  INC (inCritical);
  m.sem := 0;
  WHILE (m.waitingForMe # NIL) DO
    t := m.waitingForMe;
    m.waitingForMe := t.nextWaiting;
    t.nextWaiting := NIL;
    CanRun (t); END;
  DEC (inCritical);
END Release;

PROCEDURE Self (): T =
BEGIN
  RETURN self;
END Self;

PROCEDURE ProcessStacks (p: PROCEDURE (start, stop: ADDRESS)) =
VAR 
  t: T;
  start, stop: ADDRESS;
BEGIN
  ThreadSup.SelfTos (start, stop);
  p (start, stop);
  t := self.next;
  WHILE t # self DO
    ThreadSup.Tos (ADR (t.context[0]), start, stop);
    p (start, stop);
    t := t.next; END;
END ProcessStacks;

PROCEDURE Pause (until: Time.T) =
VAR
  t, u: T;
BEGIN
  INC (inCritical);
  self.waitingForTime := until;
  t := pausedThreads;
  u := NIL;
  WHILE (t # NIL) AND Time.Compare (t.waitingForTime, until) < 0 DO
    u := t;
    t := t.nextWaiting; END;
  IF u = NIL THEN
    self.nextWaiting := pausedThreads;
    pausedThreads := self;
  ELSE
    self.nextWaiting := t;
    u.nextWaiting := self; END;
  ICannotRun (State.pausing);
  Yield ();
END Pause;


PROCEDURE ResumeScheduling (<*UNUSED*> sig, code: INTEGER; 
                   <*UNUSED*> scp: UNTRACED REF CoreOS.struct_sigcontext) =

VAR
  it, oit: CoreOS.struct_itimerval;
BEGIN
  it.it_interval.tv_sec  := 0;
  it.it_interval.tv_usec := 0;
  it.it_value.tv_sec  := 0;
  it.it_value.tv_usec := 0;

  IF CoreOS.setitimer (CoreOS.ITIMER_REAL, it, oit) # 0 THEN
    RAISE InternalError; END;

END ResumeScheduling;
  
 
PROCEDURE StartSwitching () =
(* set the SIGPROF timer and handler; can be called to change the 
   switching interval *)
VAR
  it, oit: CoreOS.struct_itimerval;
  sv, osv: CoreOS.struct_sigvec;
BEGIN
  sv.sv_handler := switch_thread;
  sv.sv_mask    := 0;
  sv.sv_flags   := 0;
  IF CoreOS.sigvec (CoreOS.SIGVTALRM, sv, osv) #  0 THEN
    RAISE InternalError; END; 
 
  it.it_interval.tv_sec  := selected_interval.tv_sec;
  it.it_interval.tv_usec := selected_interval.tv_usec;
  it.it_value.tv_sec     := selected_interval.tv_sec;
  it.it_value.tv_usec    := selected_interval.tv_usec;
  IF CoreOS.setitimer (CoreOS.ITIMER_VIRTUAL, it, oit) # 0 THEN
    RAISE InternalError; END;

  allow_sigprof ();
END StartSwitching;


PROCEDURE switch_thread (<*UNUSED*> sig, code: INTEGER; 
                     <*UNUSED*> scp: UNTRACED REF CoreOS.struct_sigcontext) =
BEGIN
  allow_sigprof ();
  IF inCritical = 0 THEN
    INC (inCritical);
    Yield ();  END;
END switch_thread;


PROCEDURE allow_sigprof () =
VAR i : Word.T;
BEGIN
  i := CoreOS.sigsetmask (0);
  i := Word.And (i, Word.Not (CoreOS.sigmask (CoreOS.SIGVTALRM)));
  EVAL CoreOS.sigsetmask (i); 
END allow_sigprof;

PROCEDURE disallow_sigprof () =
VAR i : Word.T;
BEGIN
  i := CoreOS.sigsetmask (0);
  i := Word.Or (i, CoreOS.sigmask (CoreOS.SIGVTALRM));
  EVAL CoreOS.sigsetmask (i); 
END disallow_sigprof;



PROCEDURE DelayProfTimer (t: Time.T) =
VAR
  it, oit: CoreOS.struct_itimerval;
  sv, osv: CoreOS.struct_sigvec;
  i: INTEGER;
BEGIN
  IF t.seconds > 0 OR (t.seconds = 0 AND t.microseconds > 0) THEN
    sv.sv_handler := ResumeScheduling;
    sv.sv_mask    := 0;
    EVAL CoreOS.sigvec (CoreOS.SIGALRM, sv, osv);

    it.it_value.tv_sec := t.seconds;
    it.it_value.tv_usec := t.microseconds;
    it.it_interval.tv_sec := 0;
    it.it_interval.tv_usec := 10000; 

    i := CoreOS.sigsetmask (0);
    i := Word.Or (i, CoreOS.sigmask (CoreOS.SIGALRM));
    IF CoreOS.sigsetmask (i) # 0 THEN
      RAISE InternalError; END;

    IF (CoreOS.setitimer (CoreOS.ITIMER_REAL, it, oit) # 0) THEN
      RAISE InternalError; END;

    EVAL CoreOS.sigpause (0); END;

  IF pausedThreads # NIL THEN
    VAR
      t := pausedThreads;
      tv: CoreOS.struct_timeval;
      tz: CoreOS.struct_timezone; BEGIN

      IF CoreOS.gettimeofday (tv, tz) # 0 THEN
        RAISE InternalError; END;
      WHILE t # NIL AND
            Time.Compare (t.waitingForTime, LOOPHOLE (tv, Time.T)) <= 0 DO
        CanRun (t);
        pausedThreads := t.nextWaiting;
        t.nextWaiting := NIL;
        t := pausedThreads; END; END; END;

END DelayProfTimer;


PROCEDURE SetSwitchingInterval (i: Time.T) =
BEGIN
  selected_interval.tv_sec  := i.seconds;
  selected_interval.tv_usec := i.microseconds;

  IF multipleThreads THEN
    StartSwitching (); END;
END SetSwitchingInterval;


BEGIN
  topThread  := NEW (T, state := State.alive, alertPending := FALSE);
  topThread.context := NEW (UNTRACED REF Context, contextSize);
  ThreadSup.InitTopContext (topThread, ADR (topThread.context [0]));
  self := topThread;

  pausedThreads := NIL;

  topThread.nextRunning := topThread;
  topThread.previousRunning := topThread;
  topThread.next := topThread;
  topThread.previous := topThread;
END Thread.