borgend: comparison backup.py

-:be3ed25df789
+:db33dfa64ad6
 import logging
 import time
 import keyring
 import borgend
 from instance import BorgInstance
-from queue import Queue
+from threading import Thread, Lock, Condition
-from threading import Thread, Lock, Timer
+from scheduler import TerminableThread
 logger=borgend.logger.getChild(__name__)
 # State
 INACTIVE=0
 if isinstance(tmp, int):
 return tmp
 return None
-class Backup:
+class Backup(TerminableThread):
 def __decode_config(self, cfg):
 loc0='backup target %d' % self.identifier
-self.name=config.check_string(cfg, 'name', 'Name', loc0)
+self._name=config.check_string(cfg, 'name', 'Name', loc0)
-self.loc='backup target "%s"' % self.name
+self.loc='backup target "%s"' % self._name
 self.repository=config.check_string(cfg, 'repository',
 'Target repository', self.loc)
 self.archive_prefix=config.check_string(cfg, 'archive_prefix',
 self.__passphrase=pw
 else:
 self.__passphrase=None
 return self.__passphrase
-def __init__(self, identifier, cfg):
+def __init__(self, identifier, cfg, scheduler):
 self.identifier=identifier
-self.__decode_config(cfg)
 self.config=config
 self.lastrun_when=None
 self.borg_instance=None
 self.current_operation=None
 self.thread_log=None
 self.thread_res=None
-self.timer=None
 self.scheduled_operation=None
-self.lock=Lock()
 self.__status_update_callback=None
 self.state=INACTIVE
+self.scheduler=scheduler
+self.__decode_config(cfg)
+super().__init__(target = self.__main_thread, name = self._name)
+self.daemon=True
 def is_running(self):
-with self.lock:
+with self._cond:
 running=self.__is_running_unlocked()
 return running
 def __is_running_unlocked(self):
 running=self.current_operation
 if t=='progress_percent':
 current=safe_get_int(status, 'current')
 total=safe_get_int(status, 'total')
 if current is not None and total is not None:
-with self.lock:
+with self._cond:
 self.current_operation['progress_current']=current
 self.current_operation['progress_total']=total
 status, callback=self.__status_unlocked()
 elif t=='archive_progress':
 original_size=safe_get_int(status, 'original_size')
 compressed_size=safe_get_int(status, 'compressed_size')
 deduplicated_size=safe_get_int(status, 'deduplicated_size')
 if original_size is not None and original_size is not None and deduplicated_size is not None:
-with self.lock:
+with self._cond:
 self.current_operation['original_size']=original_size
 self.current_operation['compressed_size']=compressed_size
 self.current_operation['deduplicated_size']=deduplicated_size
 status, callback=self.__status_unlocked()
 if ('msgid' in status and
 (status['msgid']=='LockTimeout' or # observed in reality
 status['msgid']=='LockErrorT' or # in docs
 status['msgid']=='LockErrorT')): # in docs
 state=BUSY
-with self.lock:
+with self._cond:
 self.state=combine_state(self.state, state)
 status, callback=self.__status_unlocked()
 else:
 logger.debug('Unrecognised log entry %s' % str(status))
 self.borg_instance.wait()
 logger.debug('Borg subprocess terminated; terminating log listener thread')
 def __result_listener(self):
-with self.lock:
+with self._cond:
 status, callback=self.__status_unlocked()
 if callback:
 callback(self, status)
 logger.debug('Result listener thread waiting for result')
 res=self.borg_instance.read_result()
 # Finish processing remaining errors
 self.thread_log.join()
-with self.lock:
+with self._cond:
 state=self.state
 # If there were no errors, reset back to INACTIVE state
 if state==ACTIVE:
 state=INACTIVE
 logger.critical('Borg subprocess did not terminate')
 state=combine_state(state, ERRORS)
 logger.debug('Borg subprocess terminated (end state %s); terminating result listener thread' % str(state))
-with self.lock:
+with self._cond:
 if self.current_operation['operation']=='create':
 self.lastrun_when=self.current_operation['when_monotonic']
 self.thread_res=None
 self.thread_log=None
 self.borg_instance=None
 self.current_operation=None
 self.state=state
-self.__schedule_unlocked()
+self._cond.notify()
-status, callback=self.__status_unlocked()
-if callback:
+def __do_launch(self, op, archive_or_repository, *args):
-callback(self, status)
-def __do_launch(self, queue, op, archive_or_repository, *args):
 passphrase=self.extract_passphrase()
 inst=BorgInstance(op['operation'], archive_or_repository, *args)
 inst.launch(passphrase=passphrase)
 t_res.daemon=True
 self.thread_log=t_log
 self.thread_res=t_res
 self.borg_instance=inst
-self.queue=queue
 self.current_operation=op
 self.current_operation['when_monotonic']=time.monotonic()
 self.state=ACTIVE
 t_log.start()
 t_res.start()
-def __launch(self, op, queue):
+def __launch(self, op):
 if self.__is_running_unlocked():
 logging.info('Cannot start %s: already running %s'
 % (operation, self.current_operation))
 return False
 else:
-if self.timer:
-logger.debug('Unscheduling timed operation due to launch of operation')
-self.timer=None
-self.scheduled_operation=None
 try:
-logger.debug("Launching '%s' on '%s'" % (op['operation'], self.name))
+logger.debug("Launching '%s' on '%s'" % (op['operation'], self._name))
 if op['operation']=='create':
 archive="%s::%s%s" % (self.repository,
 self.archive_prefix,
 self.archive_template)
-self.__do_launch(queue, op, archive,
+self.__do_launch(op, archive,
 self.common_parameters+self.create_parameters,
 self.paths)
 elif op['operation']=='prune':
-self.__do_launch(queue, op, self.repository,
+self.__do_launch(op, self.repository,
 ([{'prefix': self.archive_prefix}] +
 self.common_parameters +
 self.prune_parameters))
 else:
 raise NotImplementedError("Invalid operation '%s'" % op['operation'])
 except Exception as err:
 logger.debug('Rescheduling after failure')
 self.lastrun_when=time.monotonic()
 self.state=ERRORS
-self.__schedule_unlocked()
 raise err
 return True
-def create(self, queue):
+def create(self):
 op={'operation': 'create', 'detail': 'manual'}
-with self.lock:
+with self._cond:
-res=self.__launch(op, queue)
+self.scheduled_operation=op
-return res
+self._cond.notify()
-def prune(self, queue):
+def prune(self):
 op={'operation': 'prune', 'detail': 'manual'}
-with self.lock:
+with self._cond:
-res=self.__launch(op, queue)
+self.scheduled_operation=op
-return res
+self._cond.notify()
 # TODO: Decide exact (manual) abort mechanism. Perhaps two stages
 def abort(self):
-with self.lock:
+with self._cond:
 if self.borg_instance:
 self.borg_instance.terminate()
-#thread_log=self.thread_log
-#thread_res=self.thread_res
-#if thread_log:
-#    thread_log.terminate()
-#if thread_res:
-#    thread_res.terminate()
-def join(self):
-logger.debug('Waiting for borg listener threads to terminate')
-with self.lock:
-thread_log=self.thread_log
-thread_res=self.thread_res
-if thread_log:
-thread_log.join()
-if thread_res:
-thread_res.join()
-assert(self.thread_log==None and self.thread_res==None)
-def __queue_timed_operation(self):
-with self.lock:
-op=self.scheduled_operation
-self.scheduled_operation=None
-self.timer=None
-if self.__is_running_unlocked():
-logger.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(op, None)
 def __next_operation_unlocked(self):
 # TODO: pruning as well
 now=time.monotonic()
 if not self.lastrun_when:
 else:
 return {'operation': 'create',
 'detail': 'normal',
 'when_monotonic': self.lastrun_when+self.backup_interval}
-def __schedule_unlocked(self):
+def __main_thread(self):
-if self.current_operation:
+with self._cond:
-return self.current_operation
+while not self._terminate:
-else:
+op=None
-op=self.__next_operation_unlocked()
+if not self.current_operation:
+op=self.__next_operation_unlocked()
-if op:
+if not op:
-now=time.monotonic()
+self.__update_status()
-delay=max(0, op['when_monotonic']-now)
+self._cond.wait()
-logger.info("Scheduling '%s' (detail: %s) of '%s' in %d seconds" %
+else:
-(op['operation'], op['detail'], self.name, delay))
+now=time.monotonic()
-tmr=Timer(delay, self.__queue_timed_operation)
+delay=max(0, op['when_monotonic']-now)
-self.scheduled_operation=op
+logger.info("Scheduling '%s' (detail: %s) of '%s' in %d seconds" %
-self.timer=tmr
+(op['operation'], op['detail'], self._name, delay))
-self.state=combine_state(self.state, SCHEDULED)
-tmr.start()
+self.scheduled_operation=op
+self.state=combine_state(self.state, SCHEDULED)
-return op
+self.__update_status()
-def schedule(self):
+self.scheduler.wait_until(now+delay, self._cond, self._name)
-with self.lock:
-return self.__schedule_unlocked()
+if self.scheduled_operation:
+op=self.scheduled_operation
-def status(self):
+self.scheduled_operation=None
-with self.lock:
+self.__launch(op)
-res=self.__status_unlocked()
-return res[0]
+# Kill a running borg to cause log and result threads to terminate
+if self.borg_instance:
+logger.debug("Terminating a borg instance")
+self.borg_instance.terminate()
+# Store threads to use outside lock
+thread_log=self.thread_log
+thread_err=self.thread_err
+logger.debug("Waiting for log and result threads to terminate")
+if thread_log:
+thread_log.join()
+if thread_res:
+thread_res.join()
 def __status_unlocked(self):
 callback=self.__status_update_callback
 if self.current_operation:
 status=self.scheduled_operation
 status['type']='scheduled'
 else:
 status={'type': 'nothing'}
-status['name']=self.name
+status['name']=self._name
 status['state']=self.state
 if 'detail' not in status:
 status['detail']='NONE'
 status['when']=(status['when_monotonic']
 -time.monotonic()+time.time())
 return status, callback
+def __update_status(self):
+status, callback = self.__status_unlocked()
+if callback:
+self._cond.release()
+try:
+callback(self, status)
+finally:
+self._cond.acquire()
 def set_status_update_callback(self, callback):
-with self.lock:
+with self._cond:
 self.__status_update_callback=callback
+def status(self):
+with self._cond:
+res=self.__status_unlocked()
+return res[0]

Mercurial > repos > borgend / file comparison

comparison: backup.py

backup.py