+# Task-related bindings to ruby */
+#
+# Copyright 2010. The SimGrid Team. All right reserved. */
+#
+# This program is free software; you can redistribute it and/or modify it
+# under the terms of the license (GNU LGPL) which comes with this package. */
+# # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # #
require 'simgrid_ruby'
-include MSG
require 'thread'
-$DEBUG = true # This is a Global Variable Useful for Debugging
+#######################################
+# Semaphore
+#######################################
-###########################################################################
-# Class Semaphore
-###########################################################################
-class Semaphore
- Thread.abort_on_exception = true
- attr_accessor :permits
-
- def initialize ( permits )
- @permits = permits
+class Semaphore
+ def initialize(initvalue = 0)
+ @counter = initvalue
+ @waiting_list = []
end
-
- def acquire(mutex,cv)
- raise "Interrupted Thread " if (!Thread.current.alive?)
- mutex.synchronize {
- while @permits <= 0
- cv.wait(mutex)
+ def acquire
+ Thread.critical = true
+ if (@counter -= 1) < 0
+ MSG::debug(Thread.current.to_s+" acquires "+self.to_s+". That's blocking.")
+ @waiting_list.push(Thread.current)
+ Thread.stop
+ else
+ MSG::debug(Thread.current.to_s+" acquires "+self.to_s+". It was free.")
+ end
+ self
+ ensure
+ Thread.critical = false
+ end
+
+ def release
+ Thread.critical = true
+ begin
+ if (@counter += 1) <= 0
+ t = @waiting_list.shift
+ t.wakeup if t
+ MSG::debug(Thread.current.to_s+" releases "+self.to_s+". Wakeup "+t.to_s)
+ else
+ MSG::debug(Thread.current.to_s+" releases "+self.to_s+". Nobody to wakeup")
end
-
- @permits = @permits - 1
- cv.signal
- }
- end
-
- def release(mutex,cv)
- mutex.synchronize{
- @permits += 1
- cv.signal
- }
+ rescue ThreadError
+ retry
+ end
+ self
+ ensure
+ Thread.critical = false
end
end
########################################################################
-# Class RbProcess
+# Class Process
########################################################################
-class RbProcess < Thread
+class MSG::Process < Thread
@@nextProcessId = 0
+
# Attributes
- attr_accessor :bind, :id, :properties, :name,
- :pargs, :schedBegin, :schedEnd, :mutex, :cv
+ attr_reader :name, :pargs ,:properties # Read only
-# Initialize : Used from ApplicationHandler to fill it in
- def initialize(*args)
-
- argc = args.size
-
- if argc == 0 # Default initializer
- super() {
- @id = 0
- @bind = 0
- @name = ""
- @pargs = Array.new()
- init_var()
- start()
- if $DEBUG
- puts "Init Default Initializer...Nothing to do...Bye"
- end
- }
-
- # 2 arguments: (HostName,Name) Or (Host , Name)
- elsif argc == 2
- super(){
- type = args[0].type()
- if ( type.to_s == "String")
- host = Host.getByName(args[0])
- elsif ( type.to_s == "MSG::Host")
- host = args[0]
- else
- raise "first argument of type "+args[0].type().to_s+", but expecting either String or MSG::Host"
- end
- if $DEBUG
- puts host
- end
- raise "Process name cannot be null" if args[1].empty?
- @name = args[1]
- if $DEBUG
- puts @name
- end
- @pargs = Array.new() # No Args[] Passed in Arguments
- @@nextProcessId += 1
- @id = @@nextProcessId
- init_var()
- start()
- createProcess(self,host)
- if $DEBUG
- puts "Initilize with 2 args"
- end
- }
-
- # 3 arguments: (hostName,Name,args[]) or (Host,Name,args[])
- elsif argc == 3
+ def initialize(*args)
super(){
- type = args[0].type()
- if ( type.to_s == "String")
- host = Host.getByName(args[0])
- elsif ( type.to_s == "MSG::Host")
- host = args[0]
- else
- raise "first argument of type "+args[0].type().to_s+", but expecting either String or MSG::Host"
- end
- if $DEBUG
- puts host
- end
-
- raise "Process name cannot be null" if args[1].empty?
- @name = args[1]
- type = args[2].type()
- raise "Third argument should be an Array" if type != "Array"
- @pargs = args[3]
- @@nextProcessId +=1
- @id = @@nextProcessId
- init_var()
- createProcess(self,host)
-
- if $DEBUG
- puts "Initilize with 3 args"
- end
-
-# sleep #keep the thread running
- }
- else
- raise "Bad number of argument: Expecting either 1, 2 or 3, but got "+argc
- end
- end
-
- # Init_var Called By Initialize
- def init_var()
- @proprieties = Hash.new()
- @mutex = Mutex.new
- @cv = ConditionVariable.new
- # Process Synchronization Tools
+
+ raise "Bad number of arguments to create a Ruby process. Expected (name,args,prop) " if args.size < 3
+
@schedBegin = Semaphore.new(0)
@schedEnd = Semaphore.new(0)
- end
-
- #main
- def msg_main(args)
- # To Be Implemented within The Process...
- # The Main Code of The Process to be Executed ...
+ @id = @@nextProcessId
+ @@nextProcessId +=1
+ @name = args[0]
+ @pargs = args[1]
+ @properties = args[2]
+ start()
+ }
+ end
+
+ def main(args)
+ # To be overriden by childs
+ raise("You must define a main() function in your process, containing the code of this process")
end
- # Start : To keep the Process Alive and waitin' via semaphore
def start()
- @schedBegin.acquire(@mutex,@cv)
- #execute The Main Code of The Process ( Example Master ; Slave ...)
- msg_main(@pargs)
- processExit(self) #Exite the Native Process
- @schedEnd.release(@mutex,@cv)
+ @schedBegin.acquire
+ MSG::debug("Let's execute the main() of the Ruby process")
+ main(@pargs)
+ @schedEnd.release
+ MSG::debug("Released my schedEnd, bailing out")
+ processExit(self) # Exit the Native Process
+
end
-# NetxId
- def nextId ()
- @@nextProcessId +=1
- return @@nextProcessId
+ def getBind()
+ return @bind
end
-
- if $DEBUG
- #Process List
- def processList()
- Thread.list.each {|t| p t}
- end
+
+ def setBind(bind)
+ @bind = bind
+ end
+
+ def unschedule()
+ @schedEnd.release
+ @schedBegin.acquire
end
- #Get Own ID
- def getID()
- return @id
+ def schedule()
+ @schedBegin.release
+ @schedEnd.acquire
end
- # set Id
- def setID(id)
- @id = id
+ def pause()
+ processSuspend(self)
end
- #Get a Process ID
- def processID(process)
- return process.id
+ def restart()
+ processResume(self)
end
- #Get Own Name
- def getName()
- return @name
+ def isSuspended()
+ processIsSuspended(self)
end
- #Get a Process Name
- def processName(process)
- return process.name
+ def getHost()
+ processGetHost(self)
end
+
+end
+############################################
+# Task Extend from the native Class RbTask
+############################################
+class MSG::Task < MSG::RbTask
- #Get Bind
- def getBind()
- return @bind
+ def initialize(*args)
+ #Nothing todo
end
- #Get Binds
- def setBind(bind)
- @bind = bind
+ def join(value)
+ super(self,value)
end
-
- def unschedule()
-
- @schedEnd.release(@mutex,@cv)
-# info("@schedEnd.release(@mutex,@cv)")
- @schedBegin.acquire(@mutex,@cv)
-# info("@schedBegin.acquire(@mutex,@cv)")
-
+
+ def data()
+ super(self)
end
- def schedule()
- @schedBegin.release(@mutex,@cv)
- @schedEnd.acquire(@mutex,@cv)
+ def name
+ super(self)
end
- #C Simualateur Process Equivalent Management
- # After Binding Ruby Process to C Process
+ def compSize
+ super(self)
+ end
-# pause
- def pause()
- processSuspend(self)
+ def send(mailbox)
+ super(self,mailbox)
end
-# restart
- def restart()
- processResume(self)
+ def receive(mailbox)
+ super(self)
end
-# isSuspended
- def isSuspended()
- processIsSuspended(self)
+ def source
+ super(self)
end
-# getHost
- def getHost()
- processGetHost(self)
+ def sender
+ super(self)
end
-# The Rest of Methods !!! To be Continued ...
-end
-
-########################################################################
-# Class ProcessFactory
-########################################################################
-
-class ProcessFactory
-
-# Attributes
- attr_accessor :args, :proprieties, :hostName, :function
-# Initlialize
- def initialize()
-
- @args = Array.new
- @proprieties = Hash.new
- @hostName = nil
- @function = nil
-
- end
-
-# setProcessIdentity
- def setProcessIdentity(hostName,function)
- @hostName = hostName
- @function = function
-
- if !args.empty?
- args.clear
- end
-
- if !proprieties.empty?
- proprieties.clear
- end
-
- end
-
-# RegisterProcess
- def registerProcessArg(arg)
-
- @args.push(arg)
-
- end
-
-# CreateProcess
- def createProcess()
-
- process = rubyNewInstance(@function) # process = rubyNewInstanceArgs(@function,@args) #
- size = @args.size
- for i in 0..size-1
- process.pargs.push(@args[i])
- end
- process.name = @function
- process.id = process.nextId() # This increment Automaticaly The Static ProcessNextId for The Class RbProces
- host = RbHost.getByName(@hostName)
- processCreate(process,host)
- process.properties = @properties
- @proprieties = Hash.new
-
- end
-
-# SetProperty
- def setProperty(id,value)
- @proprieties[id] = value
- end
-end
-
-#########################################################################
-# Class ApplicationHandler
-#########################################################################
-class ApplicationHandler
- @processFactory
-# Initialize
- def initialize()
- #Nothing todo
+ def listen(t_alias)
+ super(self)
end
- # onStartDocument
- def onStartDocument()
+ def execute
+ super(self)
+ end
- @processFactory = ProcessFactory.new
- if ($DEBUG)
- puts "onStartDocument"
- end
-
+ def listenFromHost(t_alias,host)
+ super(self)
end
-# onBeginProcess
- def onBeginProcess(hostName,function)
-
- @processFactory.setProcessIdentity(hostName,function)
-
- if ($DEBUG)
- puts "onBeginProcess"
- end
-
+ def setPriority(priority)
+ super(self,priority)
end
-
-# onProperty
- def onProperty(id,value)
-
- @processFactory.setProperty(id,value)
-
- if ($DEBUG)
- puts "onProperty"
- end
-
+
+ def cancel
+ super(self)
end
-# RegisterProcessArg
- def onProcessArg(arg)
-
- @processFactory.registerProcessArg(arg)
-
- if ($DEBUG)
- puts "onProcessArg"
- end
-
+ def hasData
+ super(self)
end
-
-# OnEndProcess
- def onEndProcess()
-
- @processFactory.createProcess()
- if ($DEBUG)
- puts "onEndProcess"
- end
-
- end
-
- # onEndDocument
- def onEndDocument()
-# Erm... Actually nothing to do !!
-
- if($DEBUG)
- puts "onEndDocument"
- end
- end
-
- # End Class
- end
-
-#########################
-# Class RbHost
-#########################
+end
-class RbHost < Host
-# Attributes
- attr_accessor :bind, :data
+####################################################
+# Host Extend from the native Class RbHost
+####################################################
+class MSG::Host < MSG::RbHost
-# Initialize
- def initialize()
- super()
- @bind = 0
- @data = nil
+ attr_reader :data
+ def initialize(*ars)
+ @data = 1
+ p "Host Initializer"
end
-end
-
-#########################
-# Class RbTask
-#########################
-class RbTask < Task
- attr_accessor :bind
+ def data()
+ return @data
+ end
- def initialize(name,comp_size,comm_size)
- super(name,comp_size,comm_size)
+ def setData(value)
+ @data = value
end
-
+
+ def getByName(name)
+ super(name)
+ end
+
+ def name
+ super(self)
+ end
+
+ def speed
+ super(self)
+ end
+
+ def getData
+ super(self)
+ end
+
+ def isAvail
+ super(self)
+ end
+
+ def number
+ super()
+ end
+
+ def getHostProcess(process)
+ super(process)
+ end
+
end
-
#########################
# Main chunck
#########################
