##################  schedule.rb -- brent@mbari.org  #####################
# $Source: /home/cvs/ESP/gen2/software/esp/lib/schedule.rb,v $
#    Copyright (C) 2003 MBARI
#    MBARI Proprietary Information. All rights reserved.
# $Id: schedule.rb,v 1.59 2005/10/25 21:20:02 brent Exp $
#
#  The Schedule class defines operations on a PriorityQ of Schedule::Events
#  to synchronize a set of ScheduleThreads.  
#  Schedule also maintains the count of busyThreads to aid in  
#  synchronization.  When busyThreads drops to zero and none of the
#  scheduler's child threads are ready to run, all processing
#  for the current (simulation) time is complete and the next deferred
#  event may be processed.  The Scheduler insures that all its child
#  threads have a higher priority than it.  
#
#  Use Thread (:name) {} to create ScheduleThreads for the scheduler to manage.
#  DO NOT have more than one active Schedule in any given Thread.
#  DO NOT create a "private" Schedule in any ScheduleThread.
#
#  Schedule Operations include:
#     #new (initialTime, logger)
#     #thread       => the thread in which this Schedule was initialized
#     #busyThreads  => set of threads not ready for scheduler to advance in time
#     #time         => current (simulation) time being processed
#     #defer (*Schedule::Event)  => schedule events for future processing
#     #cancel (*Schedule::Event) => remove events from schedule
#     #nextEvent    => returns next deferred event to be processed
#     #step             #process deferred events until time must be advanced
#     #process          #process deferred events until terminated or stopped
#     #stop         => gracefully exit process (may be resumed)
#     #terminate    => immediately kill all this schedule's ScheduleThreads
#     #sync         => wait until none of this schedule's Threads are busy
#
#  Schedule::RealTime is a subclass of Schedule that synchronizes event
#  processing with the system real-time clock (i.e. Time class)
#  Schedule::RealTime#process does not support the {assersion} parameter
#
############################################################################

require 'priorityq'
require 'lock'
require 'log'


class Schedule < RMutex

  class Stop < Exception; end
  
  class Error < StandardError; end
  
  class Event
  #subclasses must implement a :process method
  
    def initialize time=nil
      @time = time
    end
    attr_accessor :time

    def cancel sch=Thread.current.schedule
      sch.cancel self
    end
    def defer sch=Thread.current.schedule
      sch.defer self
    end
    def self.defer (*initArgs)
      (ev=new(*initArgs)).defer
      ev
    end
    def reschedule newTime, sch=Thread.current.schedule
      cancel if @time      
      defer if @time = newTime
    end
  end

  def initialize (initialTime=0.0, logger = Log.default)
  #initialize this deferred event schedule
    @time = initialTime
    @log = logger
    @busyThreads = {}
    super ()
    @q = PriorityQ.new {|x,y| y.time<=>x.time}
    (@thread=Thread.current).sCh = self
  end

  attr_reader :time, :thread, :busyThreads
  attr_accessor :log
  alias_method :now, :time
  
  def nextEvent
  # return the next event to be processed
    @q.first
  end
  
  def defer (*events)
  # store event on the priority queue
    lock {
#STDERR.puts "#{Thread.current.name} @#{@time.asLogString @time} deferring #{events}"
      @q.concat events
    }
    self
  end
  
  def cancel (*events)
  # remove any of the specified events found on priority queue
  # returns number of events actually removed
    count = 0
    lock {
#STDERR.puts "#{Thread.current.name} @#{@time.asLogString @time} cancelling #{events}"
      events.each {|ev| count += (@q.deletePriority (ev) {|v| v==ev})}
    }
    count
  end
  
  def sync
  # wait until none of this schedule's threads are busy
  # returns with Thread.critical!!
    Thread.wait until (Thread.critical=true; @busyThreads.empty?)
  end
  
  def asyncStep
  # remove next event from the (locked) queue
  # unlocks queue and returns event or nil if no events remain
    if base=@q.pop
      nxtTime = base.time
      if nxtTime < @time  #time must be monotonically increasing!
        unlock
        raise Error, "#{base.type} for @#{nxtTime.asLogString} is in the past"
      end
      @time = nxtTime
#STDERR.puts "\n*** @#{@time.asLogString} processing #{base} ***"
      unlock
      begin
        base.process
      rescue Exception => err
        raise if err.is_a? Stop
        @log.recordException err, @thread
      end
    else
      unlock
    end
    base
  end
  protected :sync, :asyncStep

  def step
    sync
    lock  
    asyncStep
  end
  
  def process
  # step until some thread raises Schedule::Stop or no more events to process
  #   may be invoked later to resume
    begin
      while step; end
    rescue Stop
    end
    self
  end
  alias_method :start, :process
  
  def stop
  # send schedule thread a Stop exception
    @thread.raise Stop
    self
  end
  
  def terminateOthers
  #abort all other threads of this scheduler
  #returns Thread.current if current thread was a thread os this scheduler
  #nil otherwise
    me=Thread.current
    myself = nil
    Thread.exclusive do
      @thread.eachChild { |kid|  
        kid == me ? myself = me : (puts kid; kid.abort) if kid.schedule == self
    }
    end
    myself
  end
  
  def terminate
  #abort all of this schedule's threads
    myself = terminateOthers
    @q.clear
    @busyThreads.clear
    stop
    myself.kill if myself
  end
 
  def threadBusy thread=Thread.current
  #flag that thread is busy and scheduler time should not advance
  #if Hash#[]= is atomic, need not be invoked with Thread.critical==true
    @busyThreads[thread] = :busy  #any value other than nil or false will do
#STDERR.puts "\nBUSY  -- #{thread.name}"
  end
  
  def threadReady
  #flag that current thread is ready for time to advance
  #if Hash#delete is atomic, need not be invoked with Thread.critical==true
  #returns nil if current thread was not busy 
#STDERR.puts "\nREADY -- #{Thread.current.name}"
    @thread.wakeup if  #this was the last busyThread...
      (wasBusy = @busyThreads.delete Thread.current) and @busyThreads.empty?
    wasBusy
  end
  
  def unsync(*args)
  # unsynchronize current thread from this scheduler to allow time to advance
  # if block is given, resync after executing it
    wasBusy = threadReady
    if block_given?
      return yield (*args) unless wasBusy
      begin
        yield (*args)
      ensure
        threadBusy
      end
    end
  end
    
    
  class << self  #operate on this thread's scheduler by default...
    def log  #return the logger for the current thread
      Thread.current.schedule.log
    end
    def time
      Thread.current.schedule.time
    end
    def now
      Thread.current.schedule.now
    end
    def defer (*events)
      Thread.current.schedule.defer (*events)
    end
    def cancel (*events)
      Thread.current.schedule.cancel (*events)
    end
    def nextEvent
      Thread.current.schedule.nextEvent
    end
    def start
      Thread.current.schedule.start
    end
    def stop
      Thread.current.schedule.stop
    end
    def process
      Thread.current.schedule.process
    end
    def terminate
      Thread.current.schedule.terminate
    end
  end
  

  class RealTime < Schedule  #schedule events based on real system time

    MaxTime = Time.at 2**31-1  #farthest in the future we can represent
    
    def initialize (startTime = Time.now, logger = Log.default)
    #initialize this real-time deferred event schedule
      super startTime, logger
    end

    def now
      Time.now
    end
    
    def defer(*events)
    #see superclass
      reschedule super
    end
    
    def cancel(*events)
    #see superclass
      reschedule super
    end

    def step
    # process next event in real-time
    # block until next event is ready to process
      loop do
        sync  #Thread.critical is set
        if (nxtEv=nextEvent)
          break if (@wakeTime=nxtEv.time)<=(present=Time.now)
#STDERR.puts "dozing #{@wakeTime-present}"
          doze @wakeTime-present  #clears Thread.critical
        else
#STDERR.puts "awaiting External Event"
          @wakeTime=MaxTime           
          doze                    #clears Thread.critical
        end
      end
      lock
      @wakeTime=nil
      asyncStep
    end
  
    private
    def reschedule result
    # private method to wake the schedule thread if it's sleeping and 
    # the time it will awaken is is after the (new) first event's time
      lock {
        thread.run if 
          @wakeTime && (!(nxtEv=nextEvent) || nxtEv.time < @wakeTime)
      }
      result
    end

  end #class Schedule::RealTime
  
end #class Schedule



class ScheduleThread < Thread
#subclass Thread to do necessary bookkeeping for a schedulers

  def initialize (name=nil, sch=Thread.schedule, *args, &block)    
    schThread = sch.thread
    type.critical=true  #don't allow parent to continue until child is running
    super(type.current) do |parent|
      begin
        parent.runChild(block, args, sch.log) do #1st do this initialization...
          (child=type.current).install parent, name
          child.sCh = schThread
          sch.threadBusy
          #note that changing our own priority may clear Thread.critical
          child.priority=schThread.priority+1 if parent == schThread
          type.critical=false
        end  
      ensure
#puts "#{name} terminated"
        sch.threadReady
      end
    end
  end

  class << self  #class methods    
    def create (name=nil, *args, &block)
      new (name, sch=Thread.schedule, *args, &block)
    end
  end
  
  def raise(*args)
#puts "raise #{inspect}"
    rebusy
    super
  end    
  def wakeup
#puts "wakeup #{inspect}"
    rebusy
    super
  end    
  def run
#puts "run #{inspect}"
    rebusy
    super
  end
    
  def stop
  # Stop current model thread and resume the scheduler if it was the last
    type.critical=true
#puts "stop #{inspect}"
    @wasBusy = schedule.threadReady
    type.wait #wakeup or run will make thread busy only if it was when stopped
  end

  def join
  # join (this model thread) with the current model thread
#puts "join #{inspect}"
    schedule.unsync {super}
  end

  def value
  # join (this model thread) with the current model thread
#puts "value #{inspect}"
    schedule.unsync {super}
  end

  def priority= newPriority
  #must keep scheduler's priority less than any of its children
    if sCh.kind_of? Thread
      sCh.setPriority newPriority-1 if sCh.priority >= newPriority
    end
    super
  end

  protected
  def rebusy
  #return thread to busy state only if became ready when last stopped
    type.exclusive {(@wasBusy=nil; schedule.threadBusy self) if @wasBusy}
  end
    
end


class Thread
#Thread.stop must be implemented via an instance method...
  def stop
    type.wait
  end
    
  rename_method :setPriority, :priority=
  protected :setPriority
  
  class << self  #class methods
    rename_method :wait, :stop
    def stop
      current.stop  #...to invoke ScheduleThread#stop when appropriate
    end
    def schedule
      current.schedule
    end
    def log
      schedule.log
    end
  end
  
  attr_accessor :sCh  #reference to ScheduleThread or Schedule object
  
  def schedule
  #return schedule object for this thread
    @sCh.kind_of?(Thread) ? @sCh.sCh : @sCh
  end
  
end


module Kernel
  def Thread (*args, &block)
    (Thread.schedule ? ScheduleThread : Thread).create(*args, &block)
  end
end


class IO
  rename_method :sysreadRaw, :sysread  
  def sysread (*args)
  #we need to unsync any real-time thread doing tty input from its scheduler
  #because it will otherwise stall the scheduler while awaiting input
    return sysreadRaw(*args) unless tty? and
      (sch=Thread.schedule).is_a? Schedule::RealTime
    sch.unsync {sysreadRaw(*args)}
  end
end


#IRB needs to create ScheduleThreads instead of normal ones
IRB.conf[:IRB_THREAD]=ScheduleThread if defined? IRB