Mercurial > repos > borgend / file revision

#
# Repository abstraction for queuing
#

import weakref
import keyring
import borgend
import config
from scheduler import QueueThread, QueuedEvent

logger=borgend.logger.getChild(__name__)

class FIFOEvent(QueuedEvent):
    def __init__(self, cond, name=None):
        self._goodtogo=False
        super().__init__(cond, name=name)

    def __lt__(self, other):
        return False

class FIFO(QueueThread):
    def __init__(self, **kwargs):
        super().__init__(target = self._fifo_thread, **kwargs)

    def _fifo_thread(self):
        with self._cond:
            while not self._terminate:
                ev=self._list
                if ev:
                    # We can only remove ev from the list when ev.cond allows
                    with ev.cond:
                        if not ev._goodtogo:
                            ev._goodtogo=True
                            ev.cond.notifyAll()
                self._cond.wait()

            # Termination cleanup
            ev=self._list
            while ev:
                # We can only remove ev from the list when ev.cond allows
                with ev.cond:
                    ev.cond.notifyAll()
                    ev=ev.next

    # cond has to be acquired on entry!
    def queue_action(self, cond, action=lambda: (), name=None):
        ev=FIFOEvent(cond, name=name)

        with self._cond:
            self._insert(ev)

        # This will release the lock on cond, allowing queue manager (scheduler)
        # thread to notify us if we are already to be released
        logger.debug("%s:Queuing %s", self.name, ev.name or 'UNKNOWN')
        ev.cond.wait()

        try:
            if ev._goodtogo:
                logger.debug("%s:Executing %s", self.name, ev.name or 'UNKNOWN')
                #
                # TODO: action() has to unlink on finish; so should maybe
                # use weak references to event.
                # Or we have to make action take all the time, so make the
                # stdout thread.
                # OR: Easiest to just move finish-waiting into __launch_check
                # instead of at the outer level of the main loop.
                #
                action()
        finally:
            with self._cond:
                self._unlink(ev)
                # Let _fifo_thread proceed to next action
                self._cond.notify()

        return ev._goodtogo

repositories=weakref.WeakValueDictionary()

class Repository(FIFO):
    def __decode_config(self, cfg):
        loc0='Repository %d' % self.identifier

        self.repository_name=config.check_string(cfg, 'name', 'Name', loc0)

        logger.debug("Configuring repository '%s'" % self.repository_name)

        loc = 'Repository "%s"'

        self.logger=logger.getChild(self.repository_name)

        self.location=config.check_string(cfg, 'location',
                                         'Target repository location', loc)

        self.borg_parameters=config.BorgParameters.from_config(cfg, loc)

        self.__keychain_account=config.check_string(cfg, 'keychain_account',
                                                    'Keychain account', loc,
                                                    default='')

        self.__passphrase=None

        if config.settings['extract_passphrases_at_startup']:
            try:
                self.extract_passphrase()
            except Exception:
                pass

    def __init__(self, identifier, cfg):
        self.identifier=identifier
        self.__decode_config(cfg)
        super().__init__(name = 'RepositoryThread %s' % self.repository_name)
        repositories[self.repository_name]=self

    def __extract_passphrase(self):
        acc=self.__keychain_account
        if not self.__passphrase:
            if acc and acc!='':
                self.logger.debug('Requesting passphrase')
                try:
                    pw=keyring.get_password("borg-backup", acc)
                except Exception as err:
                    self.logger.error('Failed to retrieve passphrase')
                    raise err
                else:
                    self.logger.debug('Received passphrase')
                self.__passphrase=pw
            else:
                self.__passphrase=None
        return self.__passphrase

    def launch_borg_instance(self, inst):
        with self._cond:
            passphrase=self.__extract_passphrase()
        inst.launch(passphrase=passphrase)

def find_repository(name):
    if name in repositories:
        return repositories[name]
    else:
        return None