NOT_STARTED, IN_PROGRESS, FINISHED, FAILED = range(4)
def __init__(self, name):
self.tasks = [ ]
+ self.resident_tasks = [ ]
self.workspace = "/tmp"
self.current_task = 0
self.callback = None
progress = property(getProgress)
+ def getStatustext(self):
+ return { self.NOT_STARTED: _("Waiting"), self.IN_PROGRESS: _("In Progress"), self.FINISHED: _("Finished"), self.FAILED: _("Failed") }[self.status]
+
def task_progress_changed_CB(self):
self.state_changed()
def start(self, callback):
assert self.callback is None
self.callback = callback
+ self.restart()
+
+ def restart(self):
self.status = self.IN_PROGRESS
self.state_changed()
self.runNext()
- sumTaskWeightings = sum([t.weighting for t in self.tasks])
- self.weightScale = (self.end+1) / float(sumTaskWeightings)
+ sumTaskWeightings = sum([t.weighting for t in self.tasks]) or 1
+ self.weightScale = self.end / float(sumTaskWeightings)
def runNext(self):
if self.current_task == len(self.tasks):
- self.callback(self, [])
- self.status = self.FINISHED
- self.state_changed()
+ if len(self.resident_tasks) == 0:
+ cb = self.callback
+ self.callback = None
+ self.status = self.FINISHED
+ self.state_changed()
+ cb(self, None, [])
+ else:
+ print "still waiting for %d resident task(s) %s to finish" % (len(self.resident_tasks), str(self.resident_tasks))
else:
- self.tasks[self.current_task].run(self.taskCallback,self.task_progress_changed_CB)
+ self.tasks[self.current_task].run(self.taskCallback, self.task_progress_changed_CB)
self.state_changed()
- def taskCallback(self, res):
+ def taskCallback(self, task, res, stay_resident = False):
+ cb_idx = self.tasks.index(task)
+ if stay_resident:
+ if cb_idx not in self.resident_tasks:
+ self.resident_tasks.append(self.current_task)
+ print "task going resident:", task
+ else:
+ print "task keeps staying resident:", task
+ return
if len(res):
print ">>> Error:", res
self.status = self.FAILED
self.state_changed()
- self.callback(self, res)
- else:
- self.state_changed();
+ self.callback(self, task, res)
+ if cb_idx != self.current_task:
+ if cb_idx in self.resident_tasks:
+ print "resident task finished:", task
+ self.resident_tasks.remove(cb_idx)
+ if res == []:
+ self.state_changed()
self.current_task += 1
self.runNext()
+ def retry(self):
+ assert self.status == self.FAILED
+ self.restart()
+
def abort(self):
if self.current_task < len(self.tasks):
self.tasks[self.current_task].abort()
+ for i in self.resident_tasks:
+ self.tasks[i].abort()
def cancel(self):
# some Jobs might have a better idea of how to cancel a job
self.abort()
-class Task(object) :
+class Task(object):
def __init__(self, job, name):
self.name = name
self.immediate_preconditions = [ ]
def setTool(self, tool):
self.cmd = tool
- self.args = [tool]
self.global_preconditions.append(ToolExistsPrecondition())
self.postconditions.append(ReturncodePostcondition())
def run(self, callback, task_progress_changed):
failed_preconditions = self.checkPreconditions(True) + self.checkPreconditions(False)
if len(failed_preconditions):
- callback(failed_preconditions)
+ callback(self, failed_preconditions)
return
self.prepare()
self.task_progress_changed = task_progress_changed
from enigma import eConsoleAppContainer
self.container = eConsoleAppContainer()
- self.container.appClosed.get().append(self.processFinished)
- self.container.dataAvail.get().append(self.processOutput)
+ self.container.appClosed.append(self.processFinished)
+ self.container.stdoutAvail.append(self.processStdout)
+ self.container.stderrAvail.append(self.processStderr)
assert self.cmd is not None
assert len(self.args) >= 1
if self.cwd is not None:
self.container.setCWD(self.cwd)
- print "execute:", self.container.execute(self.cmd, self.args), self.cmd, self.args
+ execstr = " ".join([self.cmd]+self.args)
+ print "execute:", self.container.execute(execstr), execstr
if self.initial_input:
self.writeInput(self.initial_input)
def cleanup(self, failed):
pass
+
+ def processStdout(self, data):
+ self.processOutput(data)
+
+ def processStderr(self, data):
+ self.processOutput(data)
def processOutput(self, data):
self.output_line += data
for postcondition in self.postconditions:
if not postcondition.check(self):
not_met.append(postcondition)
-
self.cleanup(not_met)
- self.callback(not_met)
+ self.callback(self, not_met)
def afterRun(self):
pass
progress = self.end
if progress < 0:
progress = 0
- print "progress now", progress
self.__progress = progress
self.task_progress_changed()
progress = property(getProgress, setProgress)
+# The jobmanager will execute multiple jobs, each after another.
+# later, it will also support suspending jobs (and continuing them after reboot etc)
+# It also supports a notification when some error occured, and possibly a retry.
class JobManager:
def __init__(self):
self.active_jobs = [ ]
self.failed_jobs = [ ]
self.job_classes = [ ]
+ self.in_background = False
self.active_job = None
def AddJob(self, job):
self.active_job = self.active_jobs.pop(0)
self.active_job.start(self.jobDone)
- def jobDone(self, job, problems):
- print "job", job, "completed with", problems
+ def jobDone(self, job, task, problems):
+ print "job", job, "completed with", problems, "in", task
+ from Tools import Notifications
+ if self.in_background:
+ from Screens.TaskView import JobView
+ Notifications.AddNotification(JobView, self.active_job)
if problems:
- self.failed_jobs.append(self.active_job)
+ from Screens.MessageBox import MessageBox
+ if problems[0].RECOVERABLE:
+ Notifications.AddNotificationWithCallback(self.errorCB, MessageBox, _("Error: %s\nRetry?") % (problems[0].getErrorMessage(task)))
+ else:
+ Notifications.AddNotification(MessageBox, _("Error") + (': %s') % (problems[0].getErrorMessage(task)), type = MessageBox.TYPE_ERROR )
+ self.errorCB(False)
+ return
+ #self.failed_jobs.append(self.active_job)
self.active_job = None
self.kick()
+ def errorCB(self, answer):
+ if answer:
+ print "retrying job"
+ self.active_job.retry()
+ else:
+ print "not retrying job."
+ self.failed_jobs.append(self.active_job)
+ self.active_job = None
+ self.kick()
+
+ def getPendingJobs(self):
+ list = [ ]
+ if self.active_job:
+ list.append(self.active_job)
+ list += self.active_jobs
+ return list
# some examples:
#class PartitionExistsPostcondition:
# def __init__(self, device):
# self.args += ["-t", filesystem]
# self.args.append(device + "part%d" % partition)
-class WorkspaceExistsPrecondition:
+class Condition:
+ RECOVERABLE = False
+
+ def getErrorMessage(self, task):
+ return _("An unknown error occured!") + " (%s @ task %s)" % (self.__class__.__name__, task.__class__.__name__)
+
+class WorkspaceExistsPrecondition(Condition):
def check(self, task):
return os.access(task.job.workspace, os.W_OK)
-class DiskspacePrecondition:
+class DiskspacePrecondition(Condition):
def __init__(self, diskspace_required):
self.diskspace_required = diskspace_required
+ self.diskspace_available = 0
def check(self, task):
import os
try:
s = os.statvfs(task.job.workspace)
- return s.f_bsize * s.f_bavail >= self.diskspace_required
+ self.diskspace_available = s.f_bsize * s.f_bavail
+ return self.diskspace_available >= self.diskspace_required
except OSError:
return False
-class ToolExistsPrecondition:
+ def getErrorMessage(self, task):
+ return _("Not enough diskspace. Please free up some diskspace and try again. (%d MB required, %d MB available)") % (self.diskspace_required / 1024 / 1024, self.diskspace_available / 1024 / 1024)
+
+class ToolExistsPrecondition(Condition):
def check(self, task):
import os
if task.cmd[0]=='/':
realpath = task.cmd
else:
- realpath = self.cwd + '/' + self.cmd
+ realpath = task.cwd + '/' + task.cmd
+ self.realpath = realpath
return os.access(realpath, os.X_OK)
-class AbortedPostcondition:
+ def getErrorMessage(self, task):
+ return _("A required tool (%s) was not found.") % (self.realpath)
+
+class AbortedPostcondition(Condition):
pass
-class ReturncodePostcondition:
+class ReturncodePostcondition(Condition):
def check(self, task):
return task.returncode == 0
git.cweiske.de / enigma2.git / blobdiff