borgend: comparison backup.py

-:10bd7e3906d9
+:bc6c3d74e6ea
 import logging
 import time
 import keyring
 import borgend
 import repository
+from enum import IntEnum
 from instance import BorgInstance
 from threading import Thread, Lock, Condition
 from scheduler import TerminableThread
 logger=borgend.logger.getChild(__name__)
-# State
+class State(IntEnum):
-INACTIVE=0
+# State
-SCHEDULED=1
+INACTIVE=0
-QUEUED=2
+SCHEDULED=1
-ACTIVE=3
+QUEUED=2
-BUSY=4
+ACTIVE=3
-OFFLINE=5
-ERRORS=6
+class Errors(IntEnum):
-def combine_state(state1, state2):
+OK=0
-return max(state1, state2)
+BUSY=1
+OFFLINE=2
+ERRORS=3
+def combine(self, other):
+return max(self, other)
+def ok(self):
+return self==self.OK
+def __str__(self):
+return _errorstring[self]
+_errorstring={
+Errors.OK: 'ok',
+Errors.BUSY: 'busy',
+Errors.OFFLINE: 'offline',
+Errors.ERRORS: 'errors'
+}
+def translate_loglevel(x):
+if x in loglevel_translation:
+return loglevel_translation[x]
+else:
+return logging.ERROR
+def safe_get_int(t, x):
+if x in t:
+tmp=t[x]
+if isinstance(tmp, int):
+return tmp
+return None
 loglevel_translation={
 'CRITICAL': logging.CRITICAL,
 'ERROR': logging.ERROR,
 'WARNING': logging.WARNING,
 'DEBUG': logging.DEBUG,
 'INFO': logging.INFO
 }
-def translate_loglevel(x):
-if x in loglevel_translation:
-return loglevel_translation[x]
-else:
-return logging.ERROR
-def safe_get_int(t, x):
-if x in t:
-tmp=t[x]
-if isinstance(tmp, int):
-return tmp
-return None
 class Backup(TerminableThread):
 def __decode_config(self, cfg):
 loc0='backup target %d' % self.identifier
 self.thread_res=None
 self.current_operation=None
 self.scheduled_operation=None
 self.lastrun_when=None
-self.state=INACTIVE
+self.state=State.INACTIVE
+self.errors=Errors.OK
 self.__decode_config(cfg)
 super().__init__(target = self.__main_thread, name = self._name)
 self.daemon=True
 assert(not running)
 def __log_listener(self):
 self.logger.debug('Log listener thread waiting for entries')
 success=True
-for status in iter(self.borg_instance.read_log, None):
+for msg in iter(self.borg_instance.read_log, None):
-self.logger.debug(str(status))
+self.logger.debug(str(msg))
-t=status['type']
+t=msg['type']
-errors_this_message=None
+errormsg=None
 callback=None
 if t=='progress_percent':
-current=safe_get_int(status, 'current')
+current=safe_get_int(msg, 'current')
-total=safe_get_int(status, 'total')
+total=safe_get_int(msg, 'total')
 if current is not None and total is not None:
 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')
+original_size=safe_get_int(msg, 'original_size')
-compressed_size=safe_get_int(status, 'compressed_size')
+compressed_size=safe_get_int(msg, 'compressed_size')
-deduplicated_size=safe_get_int(status, 'deduplicated_size')
+deduplicated_size=safe_get_int(msg, 'deduplicated_size')
 if original_size is not None and original_size is not None and deduplicated_size is not None:
 with self._cond:
 self.current_operation['original_size']=original_size
 self.current_operation['compressed_size']=compressed_size
 self.current_operation['deduplicated_size']=deduplicated_size
 elif t=='file_status':
 pass
 elif t=='log_message':
-if 'levelname' not in status:
+if 'levelname' not in msg:
-status['levelname']='ERROR'
+msg['levelname']='ERROR'
-if 'message' not in status:
+if 'message' not in msg:
-status['message']='UNKNOWN'
+msg['message']='UNKNOWN'
-if 'name' not in status:
+if 'name' not in msg:
-status['name']='borg'
+msg['name']='borg'
-lvl=translate_loglevel(status['levelname'])
+lvl=translate_loglevel(msg['levelname'])
-self.logger.log(lvl, status['name'] + ': ' + status['message'])
+self.logger.log(lvl, msg['name'] + ': ' + msg['message'])
 if lvl>=logging.WARNING:
-errors_this_message=status
+errormsg=msg
-state=ERRORS
+errors=Errors.ERRORS
-if ('msgid' in status and
+if ('msgid' in msg and
-(status['msgid']=='LockTimeout' or # observed in reality
+(msg['msgid']=='LockTimeout' or # observed in reality
-status['msgid']=='LockErrorT' or # in docs
+msg['msgid']=='LockErrorT' or # in docs
-status['msgid']=='LockErrorT')): # in docs
+msg['msgid']=='LockErrorT')): # in docs
-state=BUSY
+errors=Errors.BUSY
 with self._cond:
-self.state=combine_state(self.state, state)
+self.errors=self.errors.combine(errors)
 status, callback=self.__status_unlocked()
 else:
 self.logger.debug('Unrecognised log entry %s' % str(status))
 if callback:
-callback(status, errors=errors_this_message)
+callback(status, errors=errormsg)
 self.logger.debug('Waiting for borg subprocess to terminate in log thread')
 self.borg_instance.wait()
 self.logger.debug('Borg subprocess terminated; terminating log listener thread')
 def __result_listener(self):
-# self.state=ACTIVE
-# with self._cond:
-#     status, callback=self.__status_unlocked()
-# if callback:
-#     callback(status)
 self.logger.debug('Result listener thread waiting for result')
 res=self.borg_instance.read_result()
 # Finish processing remaining errors
 self.thread_log.join()
 with self._cond:
-state=self.state
+errors=self.errors
-# If there were no errors, reset back to INACTIVE state
-if state==ACTIVE:
-state=INACTIVE
 self.logger.debug('Borg result: %s' % str(res))
-if res is None and state==INACTIVE:
+if res is None:
 self.logger.error('No result from borg despite no error in log')
-state=ERRORS
+if errors.ok():
+errors=Errors.ERRORS
 self.logger.debug('Waiting for borg subprocess to terminate in result thread')
 if not self.borg_instance.wait():
 self.logger.error('Borg subprocess did not terminate')
-state=combine_state(state, ERRORS)
+if errors.ok():
+errors=Errors.ERRORS
-self.logger.debug('Borg subprocess terminated (end state %s); terminating result listener thread' % str(state))
+self.logger.debug('Borg subprocess terminated (errors state: %s); terminating result listener thread' % str(errors))
 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.state=State.INACTIVE
+self.errors=errors
 self.__update_status()
 self._cond.notify()
 def __do_launch(self, op, archive_or_repository, *args):
 passphrase=self.extract_passphrase()
 self.common_parameters +
 self.prune_parameters))
 else:
 raise NotImplementedError("Invalid operation '%s'" % op['operation'])
+# This must be called with self._cond held.
 def __launch_check(self):
 op=self.scheduled_operation
 if not op:
 self.logger.debug("Queued operation aborted")
 else:
 self.__launch(op)
 self.current_operation=op
 self.current_operation['when_monotonic']=time.monotonic()
-self.state=ACTIVE
+self.state=State.ACTIVE
+# Reset error status when starting a new operation
+self.errors=Errors.OK
 self.__update_status()
 def __main_thread(self):
 with self._cond:
 self.__main_thread_wait_schedule()
 if not self._terminate:
 self.__main_thread_queue_and_launch()
 except Exception as err:
 self.logger.exception("Error with backup '%s'" % self._name)
-self.lastrun_when=time.monotonic()
+self.errors=Errors.ERRORS
-self.state=ERRORS
-self.scheduled_operation=None
+self.state=State.INACTIVE
+self.scheduled_operation=None
 # Clean up to terminate: kill borg instance and communication threads
 if self.borg_instance:
 self.logger.debug("Terminating a borg instance")
 self.borg_instance.terminate()
 # Store threads to use outside lock
 thread_log=self.thread_log
 thread_res=self.thread_res
+self.thread_log=None
+self.thread_res=None
 self.logger.debug("Waiting for log and result threads to terminate")
 if thread_log:
 thread_log.join()
 delay=max(0, op['when_monotonic']-now)
 self.logger.info("Scheduling '%s' (detail: %s) in %d seconds" %
 (op['operation'], op['detail'],  delay))
 self.scheduled_operation=op
-self.state=combine_state(self.state, SCHEDULED)
+self.state=State.SCHEDULED
 self.__update_status()
 # Wait under scheduled wait
 self.scheduler.wait_until(now+delay, self._cond, self._name)
 else:
 # Nothing scheduled - just wait
-self.logger.debug("Waiting for manual scheduling")
+self.logger.info("Waiting for manual scheduling")
+self.state=State.INACTIVE
+self.__update_status()
 self._cond.wait()
 # Main thread/3. If there is a scheduled operation (it might have been
 # changed manually from 'op' created in __main_thread_wait_schedule above),
 # queue it on the repository, and launch the operation once repository
 # available
 def __main_thread_queue_and_launch(self):
 if self.scheduled_operation:
 self.logger.debug("Queuing")
-self.state=combine_state(self.state, QUEUED)
+self.state=State.QUEUED
 self.__update_status()
 self.repository.queue_action(self._cond,
 action=self.__launch_check,
 name=self._name)
 return None
 else:
 return {'operation': 'create',
 'detail': 'initial',
 'when_monotonic': now+initial_interval}
-elif self.state>=BUSY:
+elif not self.errors.ok():
 if self.retry_interval==0:
 return None
 else:
 return {'operation': 'create',
 'detail': 'retry',
 def __status_unlocked(self):
 callback=self.__status_update_callback
 if self.current_operation:
 status=self.current_operation
-status['type']='current'
+elif self.scheduled_operation:
-# Errors should be set by listeners
+status=self.scheduled_operation
 else:
-if self.scheduled_operation:
+status={'type': 'nothing'}
-status=self.scheduled_operation
-if self.state==QUEUED:
-status['type']='queued'
-else:
-status['type']='scheduled'
-else:
-status={'type': 'nothing'}
 status['name']=self._name
 status['state']=self.state
+status['errors']=self.errors
 if 'detail' not in status:
 status['detail']='NONE'
 if 'when_monotonic' in status:

Mercurial > repos > borgend / file comparison

comparison: backup.py

backup.py