The Scheduler object implements algorithms of tasks scheduling. More...

Inheritance diagram for dtm.Scheduler.Scheduler:

Collaboration diagram for dtm.Scheduler.Scheduler:

Public Member Functions
def	__init__ (self, configParser, connectBuilderLight, pollerManager=None)
	constructor initialise all connections and event handlers More...

def	createDBIDict (self, configParser)
	create dict config (dict object) More...

def	onNewTask (self, event)
	onNewTask event handler More...

def	planTaskTimeToRun (self, task)

def	onUpdateTask (self, event)
	onUpdateTask event handler More...

def	onDeleteTask (self, event)
	onDeleteTask event handler More...

def	onGetSheduledTasks (self, event)
	onGetSheduledTasks event handler More...

def	onAVGResourcesResponse (self, event)
	onAVGResourcesResponse event handler More...

def	onUpdateTaskFieldsResponse (self, event)
	method handler for DeleteTskResponse event More...

def	taskUpdateProcess (self)
	method start task delete process More...

def	rTimeCalc (self, schedulerTask, task_strategy, replanned)
	method recalculates new value of rTime field More...

def	addPendingEvent (self, event)
	add pending event in all auxiliary structures More...

def	deletePendingEvent (self, event)
	delete pending event from all auxiliary structures More...

def	stateRecalculate (self, task_strategy, schedulerTask)
	recalcutes state field More...

def	sendOperationProcessingError (self, event)
	send operation processing error More...

def	deleteTaskFromSchedule (self, deleteTask, schedulerTask=None)
	delete task from schedule More...

def	reschedulingTasks (self)
	rescheduling schedule after all changes in schedule More...

def	modifyTaskInSchedule (self, task)
	add task in schedule More...

def	checkCorrectTaskType (self, task)
	check that logic of requred operation corresponds with schedule state prevent wrong operation(update absent task, insert the same task) More...

def	isPossibleToRun (self, resourcesAVG, task_strategy)
	Check is task possible to run by comparison of required limits and actual resources. More...

def	getTimeSinceEpoch (self, date=None)
	get time since epoch in millisec More...

def	getPlannedRunTime (self, date_string)
	get planned run time More...

Public Member Functions inherited from app.BaseServerManager.BaseServerManager
def	__init__ (self, poller_manager=None, admin_connection=None, conectionLightBuilder=None, exceptionForward=False, dumpStatVars=True)
	constructor More...

def	addConnection (self, name, connection)

def	setEventHandler (self, eventType, handler)
	set event handler rewrite the current handler for eventType More...

def	send (self, connect_name, event)
	send event More...

def	reply (self, event, reply_event)
	wrapper for sending event in reply for event More...

def	poll (self)
	poll function polling connections receive as multipart msg, the second argument is pickled pyobj More...

def	process (self, event)
	process event call the event handler method that was set by user or on_unhandled_event method if not set More...

def	run (self)

def	is_connection_registered (self, name)
	check is a connection was registered in a instance of BaseServerManager i object More...

def	on_poll_timeout (self)
	function will call every time when ConnectionTimeout exception arrive More...

def	on_unhandled_event (self, event)
	function will call every time when arrive doesn't set handler for event type of event.evenType More...

def	build_poller_list (self)

def	clear_poller (self)

def	onAdminState (self, event)
	onAdminState event handler process admin SHUTDOWN command More...

def	onAdminFetchStatData (self, event)
	onAdminState event handler process admin command More...

def	onAdminSuspend (self, event)
	onAdminState event handler process admin command More...

def	getStatDataFields (self, fields)
	getStatDataFields returns stat data from storage More...

def	getSystemStat (self)
	getSystemStat returns stat data for system indicators: RAMV, RAMR and CPU More...

def	getConfigVarsFields (self, fields)
	getConfigVarsFields returns config vars from storage More...

def	onAdminGetConfigVars (self, event)
	onAdminGetConfigVars event handler process getConfigVars admin command, fill and return config vars array from internal storage More...

def	onAdminSetConfigVars (self, event)
	onAdminSetConfigVars event handler process setConfigVars admin command More...

def	setConfigVars (self, setConfigVars)
	processSetConfigVars sets config vars in storage More...

def	sendAdminReadyEvent (self)
	send ready event to notify adminInterfaceService More...

def	createLogMsg (self, event)
	from string message from event object More...

def	initStatFields (self, connect_name)
	add record in statFields More...

def	updateStatField (self, field_name, value, operation=STAT_FIELDS_OPERATION_ADD)
	update values of stat field - default sum More...

def	processSpecialConfigVars (self, name, value)
	send ready event to notify adminInterfaceService More...

def	getLogLevel (self)
	Get log level from first of existing loggers. More...

def	setLogLevel (self, level)
	Set log level for all loggers. More...

def	saveStatVarsDump (self)
	Save stat vars in json file. More...

def	loadStatVarsDump (self)
	Load stat vars in json file. More...

def	getStatVarsDumpFileName (self)
	Get stat vars file name. More...

def	createDBIDict (self, configParser)

Public Attributes
	cfg_section

	tasksToUpdate

	maxTasks

	timeSlot
	value of the timeSlot used in scheduling tasks More...

	waitResourcesEvents

	waitResourcesTasks

	dbi
	db contains schedule table More...

Public Attributes inherited from app.BaseServerManager.BaseServerManager
	dumpStatVars

	poller_manager

	eventBuilder

	exit_flag

	pollTimeout

	connections

	event_handlers

	statFields
	stat fields container More...

	configVars

	exceptionForward

Static Public Attributes
int	OPERATION_ERR = 1024

string	OPERATION_ERR_MSG = "Previous task operations is not finished"

int	RESOURCES_EXCEED_ERR = 1025

string	RESOURCES_EXCEED_ERR_MSG = "Resources are exceed"

string	SERVER = "server"

string	RESOURCES_MANAGER_CLIENT = "clientResourcesManager"

string	CLIENT_INTERFACE_SERVICE_CLIENT = "clientClientInterfaceService"

string	TIME_SLOT_PERIOD = "timeSlotPeriod"

string	MAX_TASKS = "maxTasksPerTimeSlot"

Static Public Attributes inherited from app.BaseServerManager.BaseServerManager
string	ADMIN_CONNECT_ENDPOINT = "Admin"

string	ADMIN_CONNECT_CLIENT = "Admin"

int	POLL_TIMEOUT_DEFAULT = 3000

int	STAT_FIELDS_OPERATION_ADD = 0

int	STAT_FIELDS_OPERATION_SUB = 1

int	STAT_FIELDS_OPERATION_SET = 2

int	STAT_FIELDS_OPERATION_INIT = 3

string	POLL_TIMEOUT_CONFIG_VAR_NAME = "POLL_TIMEOUT"

string	LOG_LEVEL_CONFIG_VAR_NAME = "LOG_LEVEL"

string	STAT_DUMPS_DEFAULT_DIR = "/tmp/"

string	STAT_DUMPS_DEFAULT_NAME = "%APP_NAME%_%CLASS_NAME%_stat_vars.dump"

dictionary	LOGGERS_NAMES = {APP_CONSTS.LOGGER_NAME, "dc", "dtm", "root", ""}

Detailed Description

The Scheduler object implements algorithms of tasks scheduling.

Definition at line 52 of file Scheduler.py.

Constructor & Destructor Documentation

◆ init()

def dtm.Scheduler.Scheduler.__init__	(	self,
		configParser,
		connectBuilderLight,
		pollerManager = `None`
	)

constructor initialise all connections and event handlers

Definition at line 73 of file Scheduler.py.

   def __init__(self, configParser, connectBuilderLight, pollerManager=None):
     super(Scheduler, self).__init__(pollerManager)
 
     self.cfg_section = self.__class__.__name__
     self.tasksToUpdate = list()
 
     serverAddr = configParser.get(self.cfg_section, self.SERVER)
     resourcesManagerAddr = configParser.get(self.cfg_section, self.RESOURCES_MANAGER_CLIENT)
     clientInterfaceServiceAddr = configParser.get(self.cfg_section, self.CLIENT_INTERFACE_SERVICE_CLIENT)
     self.maxTasks = int(configParser.get(self.cfg_section, self.MAX_TASKS))
 
     
     self.timeSlot = int(configParser.get(self.cfg_section, self.TIME_SLOT_PERIOD))
 
     serverConnection = connectBuilderLight.build(consts.SERVER_CONNECT, serverAddr)
     resourcesManagerConnection = connectBuilderLight.build(consts.CLIENT_CONNECT, resourcesManagerAddr)
     clientInterfaceServiceConnection = connectBuilderLight.build(consts.CLIENT_CONNECT, clientInterfaceServiceAddr,
                                                                  real_connect=False)
 
     self.addConnection(self.SERVER, serverConnection)
     self.addConnection(self.RESOURCES_MANAGER_CLIENT, resourcesManagerConnection)
     self.addConnection(self.CLIENT_INTERFACE_SERVICE_CLIENT, clientInterfaceServiceConnection)
 
     #TasksManager events
     self.setEventHandler(EVENT_TYPES.SCHEDULE_TASK, self.onNewTask)
     self.setEventHandler(EVENT_TYPES.UPDATE_TASK, self.onUpdateTask)
     self.setEventHandler(EVENT_TYPES.DELETE_TASK, self.onDeleteTask)
     self.setEventHandler(EVENT_TYPES.GET_SCHEDULED_TASKS, self.onGetSheduledTasks)
     self.setEventHandler(EVENT_TYPES.UPDATE_TASK_FIELDS_RESPONSE, self.onUpdateTaskFieldsResponse)
 
     #ResourcesManager
     self.setEventHandler(EVENT_TYPES.GET_AVG_RESOURCES_RESPONSE, self.onAVGResourcesResponse)
 
     
     self.waitResourcesEvents = dict()
 
     
     self.waitResourcesTasks = dict()
 
     
     self.dbi = DBI(self.createDBIDict(configParser))
 
     isClearOnStart = configParser.get(self.cfg_section, DTM_CONSTS.CLEAR_ON_START)
     if isClearOnStart == "True":
       schedulerTask = SchedulerTask()
       schedulerTask.state = ScheduledTask.STATE_CLOSED
       try:
         self.dbi.update(SchedulerTaskScheme(schedulerTask), "id=id")
       except DBIErr as err:
         logger.error(">>> Some DBI error in Scheduler.__init__ [" + str(err.message) + "]")
 
 

Member Function Documentation

◆ addPendingEvent()

def dtm.Scheduler.Scheduler.addPendingEvent	(	self,
		event
	)

add pending event in all auxiliary structures

Parameters

event instance of Event object

Definition at line 343 of file Scheduler.py.

   def addPendingEvent(self, event):
     #task = event.eventObj
     #self.waitResourcesTasks[task.id] = True
     self.waitResourcesEvents[event.uid] = event
 
 

Here is the caller graph for this function:

◆ checkCorrectTaskType()

def dtm.Scheduler.Scheduler.checkCorrectTaskType	(	self,
		task
	)

check that logic of requred operation corresponds with schedule state prevent wrong operation(update absent task, insert the same task)

Parameters

task	instance of Task inheritor

Definition at line 423 of file Scheduler.py.

   def checkCorrectTaskType(self, task):
     result = self.dbi.fetch(SchedulerTaskScheme(SchedulerTask()), "id = %s" % task.id)
     if isinstance(task, (NewTask, DeleteTask)):
       if len(result) > 0:
         raise LogicErr(LogicErr.ERR_CODE, "Task is already in schedule")
     else:
       if len(result) == 0:
         raise LogicErr(LogicErr.ERR_CODE, "Task is wrong type:" + str(type(task)))
 
 

Here is the caller graph for this function:

◆ createDBIDict()

def dtm.Scheduler.Scheduler.createDBIDict	(	self,
		configParser
	)

create dict config (dict object)

Definition at line 133 of file Scheduler.py.

   def createDBIDict(self, configParser):
     #get section
     return dict(configParser.items(DTM_CONSTS.DB_CONFIG_SECTION))
 
 

◆ deletePendingEvent()

def dtm.Scheduler.Scheduler.deletePendingEvent	(	self,
		event
	)

delete pending event from all auxiliary structures

Parameters

event instance of Event object

Definition at line 352 of file Scheduler.py.

   def deletePendingEvent(self, event):
     #taskId = self.waitResourcesEvents[event.uid].eventObj.id
     #del self.waitResourcesTasks[taskId]
     del self.waitResourcesEvents[event.uid]
 
 

Here is the caller graph for this function:

◆ deleteTaskFromSchedule()

def dtm.Scheduler.Scheduler.deleteTaskFromSchedule	(	self,
		deleteTask,
		schedulerTask = `None`
	)

delete task from schedule

Parameters

deleteTask instance of DeleteTask object

Definition at line 381 of file Scheduler.py.

   def deleteTaskFromSchedule(self, deleteTask, schedulerTask=None):
     if schedulerTask == None:
       schedulerTask = SchedulerTask()
       schedulerTask.id = deleteTask.id
     self.dbi.delete(SchedulerTaskScheme(schedulerTask), "id=%s" % deleteTask.id)
 
 

Here is the caller graph for this function:

◆ getPlannedRunTime()

def dtm.Scheduler.Scheduler.getPlannedRunTime	(	self,
		date_string
	)

get planned run time

Parameters

date_string sting contains data in strict format

Returns: numeric

Definition at line 482 of file Scheduler.py.

482 def getPlannedRunTime(self, date_string):

483 planed_time = datetime.datetime.strptime(date_string, "%Y-%m-%d %H:%M:%S,%f")

484 return self.getTimeSinceEpoch(planed_time)

485

486

Here is the call graph for this function:

Here is the caller graph for this function:

◆ getTimeSinceEpoch()

def dtm.Scheduler.Scheduler.getTimeSinceEpoch	(	self,
		date = `None`
	)

get time since epoch in millisec

Returns: numeric

Definition at line 470 of file Scheduler.py.

   def getTimeSinceEpoch(self, date=None):
     if date is None:
       date = datetime.datetime.now()
     epoch = datetime.datetime.utcfromtimestamp(0)  #@todo move to init
     delta = date - epoch  #datetime.datetime.now() - epoch
     return int(((delta.days * 24 * 60 * 60 + delta.seconds) * 1000 + delta.microseconds / 1000.0))
 
 

Here is the caller graph for this function:

◆ isPossibleToRun()

def dtm.Scheduler.Scheduler.isPossibleToRun	(	self,
		resourcesAVG,
		task_strategy
	)

Check is task possible to run by comparison of required limits and actual resources.

Parameters

resourcesAVG	instance of ResourcesAVG object
task	instance of Task object

Returns: True if actual resource >= required

Definition at line 438 of file Scheduler.py.

   def isPossibleToRun(self, resourcesAVG, task_strategy):
     ret = True
     if ret and "CPU" in task_strategy and resourcesAVG.cpuCores > 0:
       if (resourcesAVG.threads / resourcesAVG.cpuCores) <= task_strategy["CPU"]:
         logger.debug("CPU limit %s %s", str(resourcesAVG.threads / resourcesAVG.cpuCores),
                      str(task_strategy["CPU"]))
         ret = False
 
     if ret and "CPU_LOAD_MAX" in task_strategy:
       if resourcesAVG.cpu >= task_strategy["CPU_LOAD_MAX"]:
         logger.debug("CPU_LOAD_MAX limit %s %s", str(resourcesAVG.cpu), str(task_strategy["CPU_LOAD_MAX"]))
         ret = False
 
     if ret and "IO_WAIT_MAX" in task_strategy:
       if resourcesAVG.io > task_strategy["IO_WAIT_MAX"]:
         logger.debug("IO_WAIT_MAX limit %s %s", str(resourcesAVG.io), str(task_strategy["IO_WAIT_MAX"]))
         ret = False
 
     if ret and "RAM_FREE" in task_strategy:
       #logger.debug("FREE RAM: %s, LIMIT %s", str(resourcesAVG.ramR - resourcesAVG.ramRU), str(task_strategy["RAM_FREE"]))
       if resourcesAVG.ramR > 0 and resourcesAVG.ramRU > 0 and \
       (resourcesAVG.ramR - resourcesAVG.ramRU) < task_strategy["RAM_FREE"]:
         logger.debug("RAM_FREE limit %s < %s", str(resourcesAVG.ramR - resourcesAVG.ramRU),
                      str(task_strategy["RAM_FREE"]))
         ret = False
 
     return ret
 
 

Here is the caller graph for this function:

◆ modifyTaskInSchedule()

def dtm.Scheduler.Scheduler.modifyTaskInSchedule	(	self,
		task
	)

add task in schedule

Parameters

task	instance of Task object

Definition at line 398 of file Scheduler.py.

   def modifyTaskInSchedule(self, task):
     schedulerTask = SchedulerTask()
     schedulerTask.id = task.id
     schedulerTask.state = ScheduledTask.STATE_PLANNED
     schedulerTask.strategy = pickle.dumps(task.strategy)
 
     if hasattr(task, "rtime"):  #@todo only for tests
       schedulerTask.rTime = task.rtime
     else:
       schedulerTask.rTime = self.rTimeCalc(schedulerTask, task.strategy, False)
 
     priority = 0
     if Task.STRATEGY_PRIORITY in task.strategy:
       priority = task.strategy[Task.STRATEGY_PRIORITY]
     schedulerTask.priority = priority
     if isinstance(task, (NewTask, DeleteTask)):
       self.dbi.insert(SchedulerTaskScheme(schedulerTask))
     else:
       self.dbi.update(SchedulerTaskScheme(schedulerTask), "id = %s" % schedulerTask.id)
 
 

Here is the call graph for this function:

Here is the caller graph for this function:

◆ onAVGResourcesResponse()

def dtm.Scheduler.Scheduler.onAVGResourcesResponse	(	self,
		event
	)

onAVGResourcesResponse event handler

Parameters

event instance of Event object

Definition at line 230 of file Scheduler.py.

   def onAVGResourcesResponse(self, event):
     #get response on event sending in onGetSheduledTasks
     resourcesAVG = event.eventObj
     if event.uid in self.waitResourcesEvents:
       getScheduledTasks = self.waitResourcesEvents[event.uid].eventObj
       ids = list()
       try:
         #@todo add correct select condition
         scheduledTimeSlot = getScheduledTasks.timeSlot
         #get right border of current time slot
         curTime = self.getTimeSinceEpoch(datetime.datetime.now())
         rightBorderMs = int(math.floor(curTime / scheduledTimeSlot)) * scheduledTimeSlot + scheduledTimeSlot
         clause = "SELECT * FROM scheduler_task_scheme WHERE rTime < " + str(rightBorderMs) + " AND state=" + \
                  str(ScheduledTask.STATE_PLANNED) + " ORDER BY rTime ASC, priority DESC, tries DESC"
         if self.maxTasks > 0:
           clause = clause + " LIMIT " + str(self.maxTasks)
         scheme = SchedulerTaskScheme(SchedulerTask())
         taskSchemes = self.dbi.sql(scheme, clause)
         if hasattr(taskSchemes, '__iter__'):
           for taskScheme in taskSchemes:
             # pylint: disable-msg=W0212
             schedulerTask = taskScheme._getSchedulerTask()
             if schedulerTask.rTime > 0:
               rt = datetime.datetime.utcfromtimestamp(schedulerTask.rTime / 1000)
               rb = datetime.datetime.utcfromtimestamp(rightBorderMs / 1000)
               ct = datetime.datetime.utcfromtimestamp(curTime / 1000)
               logger.debug("Task " + str(schedulerTask.id) + " selected by rTime=" + str(rt) + \
                            ", rborder=" + str(rb) + ", now=" + str(ct))
             task_strategy = pickle.loads(str(schedulerTask.strategy))
             if self.isPossibleToRun(resourcesAVG, task_strategy):
               ids.append(schedulerTask.id)
               schedulerTask.state = ScheduledTask.STATE_INPROGRESS
               self.dbi.update(SchedulerTaskScheme(schedulerTask), "id=%s" % str(schedulerTask.id))
             else:
               logger.debug("Run isn't possible id=%s tries=%s", str(schedulerTask.id), str(schedulerTask.tries))
               schedulerTask.tries += 1
               schedulerTask.state = self.stateRecalculate(task_strategy, schedulerTask)
               if schedulerTask.state == ScheduledTask.STATE_CLOSED or \
                 (Task.STRATEGY_SDELAY in task_strategy and task_strategy[Task.STRATEGY_RDELAY] == 0):
                 self.tasksToUpdate.append(schedulerTask)
               else:
                 schedulerTask.rTime = self.rTimeCalc(schedulerTask, task_strategy, True)
               self.dbi.update(SchedulerTaskScheme(schedulerTask), "id=%s" % str(schedulerTask.id))
               #@todo maybe need some state for task isn't selected on execution
           if self.maxTasks > 0 and len(taskSchemes) == self.maxTasks:
             logger.debug(">>> Scheduled tasks limit reached LIMIT = " + str(self.maxTasks))
       except SQLAlchemyError as err:
         logger.critical(str(err.message))
         row_dbi.db.session.rollback()  # pylint: disable-all
       except DBIErr as err:
         logger.critical(">>> Some DBI error in Scheduler.onAVGResourcesResponse [" + str(err.message) + "]")
 
       getScheduledTasksResponse = GetScheduledTasksResponse(ids)
       responseEvent = self.eventBuilder.build(EVENT_TYPES.GET_SCHEDULED_TASKS_RESPONSE , getScheduledTasksResponse)
       self.reply(self.waitResourcesEvents[event.uid], responseEvent)
       self.deletePendingEvent(event)
       self.taskUpdateProcess()
     else:
       logger.error("get resourceAVG for non exist event " + str(event.uid))
 
 

Here is the call graph for this function:

◆ onDeleteTask()

def dtm.Scheduler.Scheduler.onDeleteTask	(	self,
		event
	)

onDeleteTask event handler

Parameters

event instance of Event object

Definition at line 198 of file Scheduler.py.

   def onDeleteTask(self, event):
     deleteTask = event.eventObj
     if deleteTask.id not in self.waitResourcesTasks:
       try:
         response = GeneralResponse()
         self.deleteTaskFromSchedule(deleteTask)
         self.reschedulingTasks()
       except DBIErr as err:
         logger.error(">>> Some DBI error in Scheduler.__init__ [" + str(err.message) + "]")
         response = GeneralResponse(err.errCode, err.message)
       finally:
         responseEvent = self.eventBuilder.build(EVENT_TYPES.SCHEDULE_TASK_RESPONSE , response)
         self.reply(event, responseEvent)
     else:
       self.sendOperationProcessingError(event)
 
 

Here is the call graph for this function:

Here is the caller graph for this function:

◆ onGetSheduledTasks()

def dtm.Scheduler.Scheduler.onGetSheduledTasks	(	self,
		event
	)

onGetSheduledTasks event handler

Parameters

event instance of Event object

Definition at line 218 of file Scheduler.py.

   def onGetSheduledTasks(self, event):
     #get the current resources state
     self.addPendingEvent(event)
     getAVGResourcesEvent = self.eventBuilder.build(EVENT_TYPES.GET_AVG_RESOURCES, None)
 
     getAVGResourcesEvent.uid = event.uid
     self.send(self.RESOURCES_MANAGER_CLIENT, getAVGResourcesEvent)
 
 

Here is the call graph for this function:

◆ onNewTask()

def dtm.Scheduler.Scheduler.onNewTask	(	self,
		event
	)

onNewTask event handler

Parameters

event instance of Event object

Definition at line 141 of file Scheduler.py.

   def onNewTask(self, event):
     task = event.eventObj
     try:
       response = GeneralResponse()
       if "DATE" in task.session:
         response = self.planTaskTimeToRun(task)
       else:
         self.checkCorrectTaskType(task)
         self.modifyTaskInSchedule(task)
         self.reschedulingTasks()
 
     except (DBIErr, LogicErr) as err:
       logger.error(">>> Some DBI error in Scheduler.onNewTask [" + str(err.message) + "]")
       response = GeneralResponse(err.errCode, err.message)
     finally:
       responseEvent = self.eventBuilder.build(EVENT_TYPES.SCHEDULE_TASK_RESPONSE , response)
       self.reply(event, responseEvent)
 
 
 

Here is the call graph for this function:

Here is the caller graph for this function:

◆ onUpdateTask()

def dtm.Scheduler.Scheduler.onUpdateTask	(	self,
		event
	)

onUpdateTask event handler

Parameters

event instance of Event object

Definition at line 187 of file Scheduler.py.

   def onUpdateTask(self, event):
     updateTask = event.eventObj
     if updateTask.id not in self.waitResourcesTasks:
       self.onNewTask(event)
     else:
       self.sendOperationProcessingError(event)
 
 

Here is the call graph for this function:

◆ onUpdateTaskFieldsResponse()

def dtm.Scheduler.Scheduler.onUpdateTaskFieldsResponse	(	self,
		event
	)

method handler for DeleteTskResponse event

Definition at line 293 of file Scheduler.py.

   def onUpdateTaskFieldsResponse(self, event):  # pylint: disable=W0613
     logger.debug("Task send update response, tasks len = " + str(len(self.tasksToUpdate)))
     if len(self.tasksToUpdate) > 0:
       del self.tasksToUpdate[0]
       self.taskUpdateProcess()
 
 

Here is the call graph for this function:

◆ planTaskTimeToRun()

def dtm.Scheduler.Scheduler.planTaskTimeToRun	(	self,
		task
	)

Definition at line 162 of file Scheduler.py.

   def planTaskTimeToRun(self, task):
     if "DATE" in task.session:  #@todo only for test
       date_string = task.session["DATE"]
       plannedTime = self.getPlannedRunTime(date_string)
       leftBorderMs = int(math.floor(plannedTime / self.timeSlot)) * self.timeSlot
       rightBorderMs = leftBorderMs + self.timeSlot
       try:
         scheme = SchedulerTaskScheme(SchedulerTask())
         taskInSlotNumber = len(row_dbi.db.session.query(type(scheme)).filter(type(scheme).rTime <= rightBorderMs).\
         filter(type(scheme).rTime >= leftBorderMs).filter_by(state=ScheduledTask.STATE_PLANNED).all())
         logger.debug("taskInSlotNumber=" + str(taskInSlotNumber))
 
         #@todo add more complex condition for planning
         task.rtime = plannedTime
         self.modifyTaskInSchedule(task)
 
         return GeneralResponse()
 
       except DBIErr as err:
         logger.error(">>> Some DBI error in Scheduler.planTaskTimeToRun [" + str(err.message) + "]")
 
 

Here is the call graph for this function:

Here is the caller graph for this function:

◆ reschedulingTasks()

def dtm.Scheduler.Scheduler.reschedulingTasks ( self )

rescheduling schedule after all changes in schedule

Definition at line 390 of file Scheduler.py.

   def reschedulingTasks(self):
     #@todo implement strategy
     pass
 
 

Here is the caller graph for this function:

◆ rTimeCalc()

def dtm.Scheduler.Scheduler.rTimeCalc	(	self,
		schedulerTask,
		task_strategy,
		replanned
	)

method recalculates new value of rTime field

Parameters

schedulerTask	incoming schedulerTask
task_strategy	incoming strategy dict

Definition at line 316 of file Scheduler.py.

   def rTimeCalc(self, schedulerTask, task_strategy, replanned):
     logger.debug("Old rTime=" + str(datetime.datetime.utcfromtimestamp(schedulerTask.rTime / 1000)))
 
     # Get NOW time
     ret = self.getTimeSinceEpoch()
     #logger.debug("Now time=" + str(datetime.datetime.utcfromtimestamp(ret / 1000)))
     if replanned:
       if Task.STRATEGY_RDELAY in task_strategy and int(task_strategy[Task.STRATEGY_RDELAY]) > 0:
         ret += int(task_strategy[Task.STRATEGY_RDELAY])
         logger.debug("New (RDELAY) rTime=" + str(datetime.datetime.utcfromtimestamp(ret / 1000)))
     else:
       if Task.STRATEGY_SDELAY in task_strategy and int(task_strategy[Task.STRATEGY_SDELAY]) > 0:
         ret += int(task_strategy[Task.STRATEGY_SDELAY])
         logger.debug("New (SDELAY) rTime=" + str(datetime.datetime.utcfromtimestamp(ret / 1000)))
 
     '''
     if Task.STRATEGY_RDELAY in task_strategy and int(task_strategy[Task.STRATEGY_RDELAY]) > 0:
       ret += int(task_strategy[Task.STRATEGY_RDELAY])
       logger.debug("New rTime=" + str(datetime.datetime.utcfromtimestamp(ret / 1000)))
     '''
 
     return ret
 
 

Here is the call graph for this function:

Here is the caller graph for this function:

◆ sendOperationProcessingError()

def dtm.Scheduler.Scheduler.sendOperationProcessingError	(	self,
		event
	)

send operation processing error

Parameters

event instance of Event object

Definition at line 372 of file Scheduler.py.

   def sendOperationProcessingError(self, event):
     response = GeneralResponse(self.OPERATION_ERR, self.OPERATION_ERR_MSG)
     responseEvent = self.eventBuilder.build(EVENT_TYPES.SCHEDULE_TASK_RESPONSE , response)
     self.reply(event, responseEvent)
 
 

Here is the call graph for this function:

Here is the caller graph for this function:

◆ stateRecalculate()

def dtm.Scheduler.Scheduler.stateRecalculate	(	self,
		task_strategy,
		schedulerTask
	)

recalcutes state field

Parameters

task_strategy	task's strategies
schedulerTask	task object

Definition at line 362 of file Scheduler.py.

   def stateRecalculate(self, task_strategy, schedulerTask):
     ret = schedulerTask.state
     if Task.STRATEGY_RETRY in task_strategy and task_strategy[Task.STRATEGY_RETRY] <= schedulerTask.tries:
       ret = ScheduledTask.STATE_CLOSED
     return ret
 
 

Here is the caller graph for this function:

◆ taskUpdateProcess()

def dtm.Scheduler.Scheduler.taskUpdateProcess ( self )

method start task delete process

Parameters

idsToDelete list tasks for deleting

Definition at line 303 of file Scheduler.py.

   def taskUpdateProcess(self):
     if len(self.tasksToUpdate) > 0:
       updateTaskFields = UpdateTaskFields(self.tasksToUpdate[0].id)
       updateTaskFields.fields["state"] = EEResponseData.TASK_STATE_SCHEDULE_TRIES_EXCEEDED
       event = self.eventBuilder.build(EVENT_TYPES.UPDATE_TASK_FIELDS, updateTaskFields)
       self.send(self.CLIENT_INTERFACE_SERVICE_CLIENT, event)
       logger.debug("Task send to update id = " + str(self.tasksToUpdate[0].id))
 
 

Here is the call graph for this function:

Here is the caller graph for this function:

Member Data Documentation

◆ cfg_section

dtm.Scheduler.Scheduler.cfg_section

Definition at line 76 of file Scheduler.py.

◆ CLIENT_INTERFACE_SERVICE_CLIENT

string dtm.Scheduler.Scheduler.CLIENT_INTERFACE_SERVICE_CLIENT = "clientClientInterfaceService"

static

Definition at line 65 of file Scheduler.py.

◆ dbi

dtm.Scheduler.Scheduler.dbi

db contains schedule table

Definition at line 119 of file Scheduler.py.

◆ MAX_TASKS

string dtm.Scheduler.Scheduler.MAX_TASKS = "maxTasksPerTimeSlot"

static

Definition at line 67 of file Scheduler.py.

◆ maxTasks

dtm.Scheduler.Scheduler.maxTasks

Definition at line 82 of file Scheduler.py.

◆ OPERATION_ERR

int dtm.Scheduler.Scheduler.OPERATION_ERR = 1024

static

Definition at line 56 of file Scheduler.py.

◆ OPERATION_ERR_MSG

string dtm.Scheduler.Scheduler.OPERATION_ERR_MSG = "Previous task operations is not finished"

static

Definition at line 57 of file Scheduler.py.

◆ RESOURCES_EXCEED_ERR

int dtm.Scheduler.Scheduler.RESOURCES_EXCEED_ERR = 1025

static

Definition at line 59 of file Scheduler.py.

◆ RESOURCES_EXCEED_ERR_MSG

string dtm.Scheduler.Scheduler.RESOURCES_EXCEED_ERR_MSG = "Resources are exceed"

static

Definition at line 60 of file Scheduler.py.

◆ RESOURCES_MANAGER_CLIENT

string dtm.Scheduler.Scheduler.RESOURCES_MANAGER_CLIENT = "clientResourcesManager"

static

Definition at line 64 of file Scheduler.py.

◆ SERVER

string dtm.Scheduler.Scheduler.SERVER = "server"

static

Definition at line 63 of file Scheduler.py.

◆ tasksToUpdate

dtm.Scheduler.Scheduler.tasksToUpdate

Definition at line 77 of file Scheduler.py.

◆ TIME_SLOT_PERIOD

string dtm.Scheduler.Scheduler.TIME_SLOT_PERIOD = "timeSlotPeriod"

static

Definition at line 66 of file Scheduler.py.

◆ timeSlot

dtm.Scheduler.Scheduler.timeSlot

value of the timeSlot used in scheduling tasks

Definition at line 86 of file Scheduler.py.

◆ waitResourcesEvents

dtm.Scheduler.Scheduler.waitResourcesEvents

Definition at line 110 of file Scheduler.py.

◆ waitResourcesTasks

dtm.Scheduler.Scheduler.waitResourcesTasks

Definition at line 115 of file Scheduler.py.

The documentation for this class was generated from the following file:

sources/hce/dtm/Scheduler.py

Public Member Functions

Public Attributes

Static Public Attributes

Detailed Description

Constructor & Destructor Documentation

◆ __init__()

Member Function Documentation

◆ addPendingEvent()

◆ checkCorrectTaskType()

◆ createDBIDict()

◆ deletePendingEvent()

◆ deleteTaskFromSchedule()

◆ getPlannedRunTime()

◆ getTimeSinceEpoch()

◆ isPossibleToRun()

◆ modifyTaskInSchedule()

◆ onAVGResourcesResponse()

◆ onDeleteTask()

◆ onGetSheduledTasks()

◆ onNewTask()

◆ onUpdateTask()

◆ onUpdateTaskFieldsResponse()

◆ planTaskTimeToRun()

◆ reschedulingTasks()

◆ rTimeCalc()

◆ sendOperationProcessingError()

◆ stateRecalculate()

◆ taskUpdateProcess()

Member Data Documentation

◆ cfg_section

◆ CLIENT_INTERFACE_SERVICE_CLIENT

◆ dbi

◆ MAX_TASKS

◆ maxTasks

◆ OPERATION_ERR

◆ OPERATION_ERR_MSG

◆ RESOURCES_EXCEED_ERR

◆ RESOURCES_EXCEED_ERR_MSG

◆ RESOURCES_MANAGER_CLIENT

◆ SERVER

◆ tasksToUpdate

◆ TIME_SLOT_PERIOD

◆ timeSlot

◆ waitResourcesEvents

◆ waitResourcesTasks

◆ init()