# HG changeset patch # User Tuomo Valkonen # Date 1516460896 0 # Node ID 7b2d2eac6a4857375637e118eeb6b144979cba10 # Parent e189d4a6cb8ccb2aca75e54f17e1fada33ed9832 basic scheduling diff -r e189d4a6cb8c -r 7b2d2eac6a48 backup.py --- a/backup.py Sat Jan 20 14:04:51 2018 +0000 +++ b/backup.py Sat Jan 20 15:08:16 2018 +0000 @@ -7,7 +7,7 @@ import time from instance import BorgInstance from queue import Queue -from threading import Thread, Lock +from threading import Thread, Lock, Timer loglevel_translation={ 'CRITICAL': logging.CRITICAL, @@ -75,12 +75,23 @@ self.borg_instance=None self.current_operation=None self.thread_log=None - self.thread_err=None + self.thread_res=None + self.timer=None + self.timer_operation=None + self.timer_time=None self.lock=Lock() def is_running(self): with self.lock: - running=self.borg_instance or self.thread_log or self.thread_err + running=self.__is_running_unlocked() + return running + + def __is_running_unlocked(self): + running=self.current_operation + if not running: + # Consistency check + assert((not self.borg_instance and not self.thread_log and + not self.thread_res)) return running def __block_when_running(self): @@ -124,11 +135,6 @@ elif t=='unparsed_error': success=False - #if (may_indicate_finished and 'finished' in status and - # status['finished']): - # logging.info('Borg subprocess finished succesfully') - # success=status['finished'] - logging.debug('Waiting for borg subprocess to terminate in log thread') self.borg_instance.wait() @@ -137,7 +143,7 @@ with self.lock: self.thread_log=None - self.__cleanup_if_both_listeners_terminated() + self.__finish_and_reschedule_if_both_listeners_terminated() def __result_listener(self): @@ -154,26 +160,26 @@ logging.debug('Waiting for borg subprocess to terminate in result thread') - self.borg_instance.wait() + success=success and self.borg_instance.wait() - logging.debug('Borg subprocess terminated; terminating result listener thread') + logging.debug('Borg subprocess terminated (success: %s); terminating result listener thread' % str(success)) with self.lock: if self.current_operation=='create': self.lastrun=self.time_started self.lastrun_success=success self.thread_res=None - self.__cleanup_if_both_listeners_terminated() + self.__finish_and_reschedule_if_both_listeners_terminated() - def __cleanup_if_both_listeners_terminated(self): + def __finish_and_reschedule_if_both_listeners_terminated(self): if self.thread_res==None and self.thread_log==None: logging.debug('Both threads terminated') self.borg_instance=None + self.time_started=None self.current_operation=None - self.time_started=None + self.__schedule_unlocked() - def __launch(self, queue, operation, archive_or_repository, *args): - + def __do_launch(self, queue, operation, archive_or_repository, *args): inst=BorgInstance(operation, archive_or_repository, *args) inst.launch() @@ -193,23 +199,48 @@ t_log.start() t_res.start() - def create(self, queue): - self.__block_when_running() + def __launch(self, operation, queue): + if self.__is_running_unlocked(): + logging.info('Cannot start %s: already running %s' + % (operation, self.current_operation)) + return False + else: + if self.timer: + logging.debug('Unscheduling timed operation due to launch of operation') + self.timer=None + self.timer_operation=None + self.timer_time=None + + logging.debug("Launching '%s' on '%s'" % (operation, self.name)) + + if operation=='create': + archive="%s::%s%s" % (self.repository, + self.archive_prefix, + self.archive_template) - archive="%s::%s%s" % (self.repository, - self.archive_prefix, - self.archive_template) + self.__do_launch(queue, operation, archive, + self.common_parameters+self.create_parameters, + self.paths) + elif operation=='prune': + self.__do_launch(queue, 'prune', self.repository, + ([{'prefix': self.archive_prefix}] + + self.common_parameters + + self.prune_parameters)) + else: + logging.error("Invalid operaton '%s'" % operation) + self.__schedule_unlocked() - self.__launch(queue, 'create', archive, - self.common_parameters+self.create_parameters, - self.paths) + return True + + def create(self, queue): + with self.lock: + res=self.__launch('create', queue) + return res def prune(self, queue): - self.__block_when_running() - self.__launch(queue, 'prune', self.repository, - ([{'prefix': self.archive_prefix}] + - self.common_parameters + - self.prune_parameters)) + with self.lock: + res=self.__launch('prune', queue) + return res # TODO: Decide exact (manual) abort mechanism. Perhaps two stages def abort(self): @@ -227,7 +258,7 @@ def join(self): - logging.debug('Waiting for borg listener thread to terminate') + logging.debug('Waiting for borg listener threads to terminate') with self.lock: thread_log=self.thread_log @@ -241,9 +272,42 @@ assert(self.thread_log==None and self.thread_res==None) - def next_action(): - __block_when_running() - # TODO pruning as well + def __queue_timed_operation(self): + with self.lock: + operation=self.timer_operation + self.timer_operation=None + self.timer_time=None + self.timer=None + + if self.__is_running_unlocked(): + logging.info('Aborted queue operation, as an operation is already running') + else: + # TODO: Queue on 'repository' and online status for SSH, etc. + + # TODO: UI comms. queue? + self.__launch(operation, None) + + def __schedule_unlocked(self): + if self.current_operation: + return self.current_operation, None + else: + operation, when=self.__next_operation_unlocked() + + if operation: + now=time.monotonic() + delay=max(0, when-now) + logging.info("Scheduling '%s' of '%s' in %d seconds" % + (operation, self.name, delay)) + tmr=Timer(delay, self.__queue_timed_operation) + self.timer_operation=operation + self.timer_time=when + self.timer=tmr + tmr.start() + + return operation, time + + def __next_operation_unlocked(self): + # TODO: pruning as well now=time.monotonic() if not self.lastrun: return 'create', now+self.retry_interval @@ -255,4 +319,8 @@ else: return 'create', self.lastrun+self.backup_interval + def schedule(self): + with self.lock: + return self.__schedule_unlocked() + diff -r e189d4a6cb8c -r 7b2d2eac6a48 borgend.py --- a/borgend.py Sat Jan 20 14:04:51 2018 +0000 +++ b/borgend.py Sat Jan 20 15:08:16 2018 +0000 @@ -9,12 +9,8 @@ from backup import Backup from config import settings from queue import Queue -import ui +from ui import BorgendTray -if __name__ == "__main__": - #print(settings) - #BorgendTray("Borgend").run() - pass backupconfigs=settings['backups'] backups=[None]*len(backupconfigs); @@ -24,6 +20,12 @@ backups[i]=Backup(i, backupconfigs[i]) queue=Queue() -backups[0].create(queue) +#print(backups[0].create(queue)) +backups[0].schedule() + +#backups[0].join() -backups[0].join() +if __name__ == "__main__": + #print(settings) + BorgendTray("Borgend").run() + pass diff -r e189d4a6cb8c -r 7b2d2eac6a48 scheduler.py --- a/scheduler.py Sat Jan 20 14:04:51 2018 +0000 +++ /dev/null Thu Jan 01 00:00:00 1970 +0000 @@ -1,25 +0,0 @@ -# -# Borgend scheduler -# - -from Queue import Queue -import sched -import ui - - -class Scheduler: - - def __init__(self, backups): - self.backups=backups - self.eventqueue=Queue() - self.t=Thread(target=self.__scheduler) - t.start() - - def __scheduler(sched): - timeout=??? - q=sched.eventqueue - while True: - timeout, timerevent=next_timed_event(sched); - t=Timer(timeout, lambda: q.put(timerevent)); - event=sq.get(True): - # Decide what to do