#######################  lock.rb -- brent@mbari.org  #####################
# $Source: /home/cvs/ESP/gen2/software/esp/lib/lock.rb,v $
# $Id: lock.rb,v 1.35 2005/01/25 01:17:38 brent Exp $
#
#  The Lock class combines an RMutex with a set of EventVariables
#  EventVariables are ConditionVariables with the ability to 
#  rendevous with threads awaiting the event.
#  An RMutex is a recursive analog to Mutex.
#  RMutex#lock can combine the functions of Mutex#lock and Mutex#synchronize
#  Each Lock contains a delegated RMutex
#
#  Selected Methods:
#     Lock.new :stopped, :ready => a new Lock with stopped and ready conditions
#     Lock#add :rolling  => add a rolling condition to an existing Lock
#     Lock#await :stopped => block until some thread signals aLock :stopped
#     Lock#signal :stopped => unblock first thread awaiting aLock :stopped
#     Lock#broadcast :stopped => unblock all threads awaiting "
#     Lock#lock  => synchronize on aLock's mutex
#
############################################################################

require 'mbari'
require 'thread'
require 'threads'


class RMutex
# Recursive Mutex analogous to the standard non-recursive one
# Does not inherit from Mutex because @lock is replaced by @count
# Also note that #lock method has been extended to take a criticalBlock
  
  module Mixin
  
    def initRMutex
      @owner = nil    #owning Thread or nil if RMutex is unlocked
      @count = 0      #times @owner has recursively locked this mutex
      @waiting = []   #threads waiting for this RMutex
      @waiting.taint  # enable tainted comunication?
      taint
    end
      
    attr_reader :owner, :count  
    def waiting; @waiting.dup; end

    def owner= newOwner
    #only allow the thread currently owning the lock to change its owner
      Thread.exclusive {
        if @owner != newOwner
          raise ThreadError, 
            "#{newOwner} is not a Thread" unless newOwner.is_a? Thread
          raise ThreadError, "not #{type} owner" unless @owner == Thread.current
          @owner = newOwner
        end
      }
    end
    
    def lock claims=1
    # lock recursive mutex, block if necessary
    # claims is the number of recursive mutexes to acquire
    # if a critical Block is given, execute it and unlock after
      return self if claims < 1
      me=Thread.current
      if @owner == me
        @count += claims
      else 
        Thread.exclusive {
          if @owner
            @waiting << me
            Thread.stop until @owner == me
          else
            @owner = me
          end
        }
        @count = claims
      end
      if block_given?
        begin
          yield  #otherwise return the results of critical section
        ensure
          unlock
        end
      end
    end
    alias_method :synchronize, :lock  #Why bother with synchronize?

    def locked?
    #true if any thread has the mutex
    #Note:  This is not very useful in practice.  Use try_lock instead
      !@owner.nil?
    end

    def releaseLock
    #common to unlock and exclusive_unlock
    #relase the lock and pass it to the next waiting thread
      @owner = nil
      begin
        if t = @waiting.shift
          t.wakeup
          @owner = t
          @count = 1
        end
      rescue ThreadError
        retry
      end
    end
    private :releaseLock
    
    def unlock
    # unlock recursive mutex
      Thread.exclusive {releaseLock} if 
        Thread.current == @owner and (@count-=1)<=0
      self
    end

    def exclusive_unlock
    # execute block after removing all claims to this RMutex
    # returns the previous number of claims this thread held 
    # (for relocking the RMutex)
      return 0 unless Thread.current == @owner
      claims = @count
      @count = 0
      Thread.exclusive do
        releaseLock
        yield
      end
      claims
    end
    
    def try_lock (claims=1)
    # try to lock recursive mutex without blocking
    # claims = number of unlocks needed to release lock if acquired 
    # returns true if mutexes were acquired, false if mutex is owned by another
      if claims > 0  #attempts to acquire < 1 mutexes always succeed
        Thread.critical = true
        case @owner
          when nil  #become the owner of this RMutex
            @owner = Thread.current
            @count = claims      
          when Thread.current  #just bump count if we already own it
            @count += claims
          else  #return false if some other thread owns it
            return Thread.critical = false
        end
        Thread.critical = false
      end
      true
    end

  end  #Mixin

  include Mixin
  def initialize
    initRMutex
  end 
  
end


class Mutex
#redefine :lock method to ignore RMutex's "claims" argument
  def lock (ignored=nil)
    while (Thread.critical = true; @locked)
      @waiting.push Thread.current
      Thread.stop
    end
    @locked = true
    Thread.critical = false
    self
  end
end

class ConditionVariable
#redefine :wait method to work properly with either Mutex or RMutex
  def wait mutex
    mutex.lock mutex.exclusive_unlock {
      @waiters << Thread.current
      Thread.stop
    }
  end
end
  


class Queue

  def push obj
  #This push will remain critical if invoked within a critical section
    Thread.exclusive do
      @que.push obj
      begin
        t = @waiting.shift
        t.wakeup if t
      rescue ThreadError
        retry
      end
    end
  end
  alias_method :<<, :push
  alias_method :enq, :push
  
  class PopResult  #result type from Queue.pop below
    def initialize queue, item
      @source = queue
      @item = item
    end
    attr_reader :source, :item
  end
  
  def self.pop sources
  #pop from any of array of sources (searching from first to last)
  #block only if all the sources are empty
  #returns a Queue::PopResult
    Thread.exclusive do
      currentThread = Thread.current
      loop {
        sources.each {|q| return PopResult.new(q, q.shIfT) unless q.empty?}
        sources.each {|q| q.addWaiTer currentThread}
        Thread.stop
        sources.each {|q| q.removeWaiTer currentThread}
      } 
    end
  end
  
  def shIfT  #"private" helper methods for Queue.pop
    @que.shift
  end
  def addWaiTer thread
    @waiting.push thread
  end
  def removeWaiTer thread
    @waiting.delete thread
  end
  
  def first
  #return first element without removing it
    @que.first
  end
  def last
  #return last element
    @que.last
  end
  
end


class EventVariable < ConditionVariable
#A "synchronous" version of ConditionVariable
  def initialize
    @loiterers = []  #threads awaiting a rendevous with waiters
    super
  end

  def rendevous (mutex, minWaiters=1)
  #block until at least minWaiters threads are awaiting event under (r)mutex
  #this thread should already have locked the mutex or rmutex!
    while @waiters.length < minWaiters
      mutex.lock mutex.exclusive_unlock {
        @loiterers << Thread.current
        Thread.stop
      }
    end
  end
  
  def wait mutex
  #this thread should already have locked the mutex (or rmutex)
  #first, awaken all threads that were awaiting rendevous
    Thread.critical = true
    for t in @loiterers
      begin
	t.wakeup
      rescue ThreadError
      end
    end
    @loiterers.clear
    #if we own the mutex, stop this thread 
    #                     until one of the others (@loiterers?) resume it
    mutex.lock mutex.exclusive_unlock {
      @waiters << Thread.current
      Thread.stop
    }
  end
  
  def to_s
  #convert to a pithy string for debugging
   a=[]
   a<<@waiters.collect{|t|t.name}.inspect+" awaiting" unless @waiters.empty?
   a<<@loiterers.collect{|t|t.name}.inspect+" signaling"unless @loiterers.empty?
   a.join(" & ")
  end
  
  def refs? (thread)
  #true iff this has some reference to thread
    @waiters.include?(thread) || @loiterers.include?(thread)
  end
end


class Lock < RMutex
  
  @@serialNumber = 0 unless defined? @@serialNumber  #never reset serialNumber
  
  def initialize (name, *events)
    @name = name
    @serialNumber = (@@serialNumber += 1)  #monotonically increasing
    super ()
    @eventHash = {}   #hash of EventVariables
    add *events
  end

  attr_reader :serialNumber
  protected :serialNumber
  
  def name
    return @name unless @name.nil?
    id
  end
    
  def add (*events)
    lock {events.each {|c| 
      @eventHash[c] = EventVariable.new unless @eventHash.has_key? c 
    }}
    self
  end
  
  def await (event)
    lock {@eventHash.fetch(event).wait self}
    self
  end
  
  def rendevous (event, minWaiters=1)
  #block until at least minWaiters threads are awaiting event
    lock {@eventHash.fetch(event).rendevous (self, minWaiters)}
    self
  end

  def syncSignal (event, minWaiters=1)
  #signal after establishing that at least minWaiters threads will receive it
    meshPoint=@eventHash.fetch event
    meshPoint.rendevous (self, minWaiters)
    meshPoint.signal
    self
  end
  alias_method :signal, :syncSignal
  
  def asyncSignal (event)
  #signal without waiting for rendevous
    @eventHash.fetch(event).signal
    self
  end
  
  def broadcast (event)
  #signal all waiting threads (without blocking)
    @eventHash.fetch(event).broadcast
    self
  end

  rename_method :raw, :inspect
  def inspect
    "#<%s:%s %s>" % [type, name.inspect, owner ? 
"held#{"("+self.count+")" if self.count!=1} by "+owner.name.inspect : "open"
    ] 
  end
  alias_method :to_s, :inspect
  
  def self.unsortedList
  #list all Locks in existence
    list = []
    ObjectSpace.each_object (Lock) {|lock| list << lock }
    list
  end
  
  def <=> other
    @serialNumber <=> other.serialNumber
  end
  
  def self.list (key=nil)
  #list all Locks named key sorted by birth time
  #if key is omitted, list all Locks sorted by serial number
    return unsortedList.sort unless key
    unsortedList.find_all {|t| t.name == key}.sort
  end
  
  def self.[] (key=nil)
  # the matching lock with the latest serial number
  # if no key supplied, the most recently created lock
    list(key).last
  end

  def [] (event)
  # the specified EventVariable
    @eventHash[event]
  end 
   
  #the remaining Lock definitions are for debugging only
  def list
  # all EventVariables
    @eventHash
  end 
   
  def refs? (thread)
  #true iff lock contains any reference to thread
    return true if owner == thread || waiting.include?(thread)
    @eventHash.each_value {|event| return true if event.refs?(thread)}
    false
  end
  
  def details
  #report on all aspects of this lock
    puts inspect+
     "#{' with '+waiting.collect{|t|t.name}.inspect+' waiting for it' unless
      waiting.empty?}"
    @eventHash.each {|key, event|
      s = event.to_s
      puts (s+" event "+key.inspect) unless s.empty?
    }
    nil
  end
end


class Thread
  def locks
  #array of locks in which thread is referenced
    Lock.list.find_all {|lock| lock.refs? self}
  end
end


module Kernel
  def Lock (name, *events)
    Lock.new (name, *events)
  end
end