blob: 3685a70a74e4e70508a4888f136cd35c62a33421 [file] [log] [blame]
"""
This file contains the event scheduler class for CHOTestMonkey
Author: you@onlab.us
"""
from threading import Lock, Condition
from tests.CHOTestMonkey.dependencies.events.Event import EventType, EventStates, Event
from tests.CHOTestMonkey.dependencies.events.TestEvent import *
from tests.CHOTestMonkey.dependencies.events.CheckEvent import *
from tests.CHOTestMonkey.dependencies.events.NetworkEvent import *
from tests.CHOTestMonkey.dependencies.events.AppEvent import *
from tests.CHOTestMonkey.dependencies.events.ONOSEvent import *
class EventScheduleMethod:
def __init__( self ):
self.map = {}
self.RUN_NON_BLOCK = 1
self.map[ 1 ] = 'RUN_NON_BLOCK'
self.RUN_BLOCK = -1
self.map[ -1 ] = 'RUN_BLOCK'
class EventTuple:
def __init__( self, id, className, typeString, typeIndex, scheduleMethod, args, rerunInterval, maxRerunNum ):
self.default = ''
self.id = 0
self.className = className
self.typeString = typeString
self.typeIndex = typeIndex
self.scheduleMethod = scheduleMethod
self.args = args
self.rerunInterval = rerunInterval
self.maxRerunNum = maxRerunNum
def startEvent( self ):
assert self.className in globals().keys()
event = globals()[ self.className ]
return event().startEvent( self.args )
class EventScheduler:
def __init__( self ):
self.default = ''
self.pendingEvents = []
self.pendingEventsCondition = Condition()
self.runningEvents = []
self.runningEventsCondition = Condition()
self.isRunning = True
self.idleCondition = Condition()
self.pendingEventsCapacity = int( main.params[ 'SCHEDULER' ][ 'pendingEventsCapacity' ] )
self.runningEventsCapacity = int( main.params[ 'SCHEDULER' ][ 'runningEventsCapacity' ] )
self.scheduleLoopSleep = float( main.params[ 'SCHEDULER' ][ 'scheduleLoopSleep' ] )
def scheduleEvent( self, typeIndex, scheduleMethod, args=None, index=-1 ):
"""
Insert an event to pendingEvents
param:
index: the position to insert into pendingEvents, default value -1 implies the tail of pendingEvents
"""
if typeIndex not in main.enabledEvents.keys():
main.log.warn( "Event Scheduler - event type %s not enabled" % ( typeIndex ) )
return
if main.enabledEvents[ typeIndex ] in main.params[ 'EVENT' ].keys():
if 'rerunInterval' in main.params[ 'EVENT' ][ main.enabledEvents[ typeIndex ] ].keys():
rerunInterval = int( main.params[ 'EVENT' ][ main.enabledEvents[ typeIndex ] ][ 'rerunInterval' ] )
maxRerunNum = int( main.params[ 'EVENT' ][ main.enabledEvents[ typeIndex ] ][ 'maxRerunNum' ] )
else:
rerunInterval = int( main.params[ 'EVENT' ][ 'Event' ][ 'rerunInterval' ] )
maxRerunNum = int( main.params[ 'EVENT' ][ 'Event' ][ 'maxRerunNum' ] )
eventTuple = EventTuple( main.eventID, main.enabledEvents[ typeIndex ], EventType().map[ typeIndex ], typeIndex, scheduleMethod, args, rerunInterval, maxRerunNum )
with main.variableLock:
main.eventID += 1
main.log.debug( "Event Scheduler - Event added: %s, %s, %s" % ( typeIndex,
scheduleMethod,
args ) )
with self.pendingEventsCondition:
if index == -1:
self.pendingEvents.append( eventTuple )
elif index > -1 and index <= len( self.pendingEvents ):
self.pendingEvents.insert( index, eventTuple )
else:
main.log.warn( "Event Scheduler - invalid index when isnerting event: %s" % ( index ) )
self.pendingEventsCondition.notify()
# self.printEvents()
def startScheduler( self ):
"""
Start the loop which schedules the events in pendingEvents
"""
import time
while True:
with self.pendingEventsCondition:
while len( self.pendingEvents ) == 0:
self.pendingEventsCondition.wait()
eventTuple = self.pendingEvents[ 0 ]
main.log.debug( "Event Scheduler - Scheduling event: %s, %s, %s" % ( eventTuple.typeIndex,
eventTuple.scheduleMethod,
eventTuple.args ) )
if eventTuple.scheduleMethod == EventScheduleMethod().RUN_NON_BLOCK:
# Run NON_BLOCK events using threads
with self.pendingEventsCondition:
self.pendingEvents.remove( eventTuple )
t = main.Thread( target=self.startEvent,
threadID=main.threadID,
name="startEvent",
args=[ eventTuple ] )
t.start()
with main.variableLock:
main.threadID += 1
elif eventTuple.scheduleMethod == EventScheduleMethod().RUN_BLOCK:
# Wait for all other events before start
with self.runningEventsCondition:
while not len( self.runningEvents ) == 0:
self.runningEventsCondition.wait()
# BLOCK events will temporarily block the following events until finish running
with self.pendingEventsCondition:
self.pendingEvents.remove( eventTuple )
self.startEvent( eventTuple )
else:
with self.pendingEventsCondition:
self.pendingEvents.remove( eventTuple )
time.sleep( self.scheduleLoopSleep )
def startEvent( self, eventTuple ):
"""
Start a network/ONOS/application event
"""
import time
with self.runningEventsCondition:
self.runningEvents.append( eventTuple )
# self.printEvents()
rerunNum = 0
result = eventTuple.startEvent()
while result == EventStates().FAIL and rerunNum < eventTuple.maxRerunNum:
time.sleep( eventTuple.rerunInterval )
rerunNum += 1
main.log.debug( eventTuple.typeString + ": retry number " + str( rerunNum ) )
result = eventTuple.startEvent()
if result == EventStates().FAIL:
main.log.error( eventTuple.typeString + " failed" )
main.caseResult = main.FALSE
if main.params[ 'TEST' ][ 'pauseTest' ] == 'on':
self.isRunning = False
main.log.error( "Event Scheduler - Test paused. To resume test, run \'resume-test\' command in CLI debugging mode" )
with self.runningEventsCondition:
self.runningEvents.remove( eventTuple )
if len( self.runningEvents ) == 0:
self.runningEventsCondition.notify()
with self.pendingEventsCondition:
if len( self.pendingEvents ) == 0:
with self.idleCondition:
self.idleCondition.notify()
# self.printEvents()
def printEvents( self ):
"""
Print all the events in pendingEvents and runningEvents
"""
events = " ["
with self.runningEventsCondition:
for index in range( 0, len( self.runningEvents ) - 1 ):
events += str( self.runningEvents[ index ].typeIndex )
events += ", "
if len( self.runningEvents ) > 0:
events += str( self.runningEvents[ -1 ].typeIndex )
events += "]"
events += " ["
with self.pendingEventsCondition:
for index in range( 0, len( self.pendingEvents ) - 1 ):
events += str( self.pendingEvents[ index ].typeIndex )
events += ", "
if len( self.pendingEvents ) > 0:
events += str( self.pendingEvents[ -1 ].typeIndex )
events += "]"
main.log.debug( "Event Scheduler - Events: " + events )
def isAvailable( self ):
with self.runningEventsCondition:
with self.pendingEventsCondition:
return len( self.pendingEvents ) < self.pendingEventsCapacity and\
len( self.runningEvents ) < self.runningEventsCapacity and\
self.isRunning
def isIdle( self ):
with self.runningEventsCondition:
with self.pendingEventsCondition:
return len( self.pendingEvents ) == 0 and\
len( self.runningEvents ) == 0 and\
self.isRunning
def setPendingEventsCapacity( self, capacity ):
self.pendingEventsCapacity = capacity
def setRunningState( self, state ):
assert state or state is False
self.isRunning = state