borgend: comparison backup.py

-:e189d4a6cb8c
+:7b2d2eac6a48
 import config
 import logging
 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,
 'ERROR': logging.ERROR,
 'WARNING': logging.WARNING,
 self.lastrun=None
 self.lastrun_success=None
 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):
 running=self.is_running()
 assert(not running)
 elif t=='exception':
 success=False
 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()
 logging.debug('Borg subprocess terminated; terminating log listener thread')
 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):
 logging.debug('Result listener thread waiting for result')
 if res==None:
 success=False
 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()
 t_log=Thread(target=self.__log_listener)
 t_log.daemon=True
 self.time_started=time.monotonic()
 t_log.start()
 t_res.start()
+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)
+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()
+return True
 def create(self, queue):
-self.__block_when_running()
+with self.lock:
+res=self.__launch('create', queue)
-archive="%s::%s%s" % (self.repository,
+return res
-self.archive_prefix,
-self.archive_template)
-self.__launch(queue, 'create', archive,
-self.common_parameters+self.create_parameters,
-self.paths)
 def prune(self, queue):
-self.__block_when_running()
+with self.lock:
-self.__launch(queue, 'prune', self.repository,
+res=self.__launch('prune', queue)
-([{'prefix': self.archive_prefix}] +
+return res
-self.common_parameters +
-self.prune_parameters))
 # TODO: Decide exact (manual) abort mechanism. Perhaps two stages
 def abort(self):
 with self.lock:
 if self.borg_instance:
 if thread_res:
 thread_res.terminate()
 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
 thread_res=self.thread_res
 if thread_res:
 thread_res.join()
 assert(self.thread_log==None and self.thread_res==None)
-def next_action():
+def __queue_timed_operation(self):
-__block_when_running()
+with self.lock:
-# TODO pruning as well
+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
 elif not self.lastrun_success:
 return 'create', self.lastrun+self.retry_interval
 if self.backup_interval==0:
 return 'none', 0
 else:
 return 'create', self.lastrun+self.backup_interval
+def schedule(self):
+with self.lock:
+return self.__schedule_unlocked()

Mercurial > repos > borgend / file comparison

comparison: backup.py

backup.py