borg/attic/cache.py

228 lines
9.1 KiB
Python
Raw Normal View History

2013-06-03 11:45:48 +00:00
from configparser import RawConfigParser
from attic.remote import cache_if_remote
import msgpack
2010-10-20 19:08:46 +00:00
import os
from binascii import hexlify
2010-12-21 20:29:09 +00:00
import shutil
2014-06-03 20:00:34 +00:00
from .helpers import Error, get_cache_dir, decode_dict, st_mtime_ns, unhexlify, UpgradableLock, int_to_bigint, \
bigint_to_int
from .hashindex import ChunkIndex
2010-03-06 17:25:35 +00:00
class Cache(object):
"""Client Side cache
"""
class RepositoryReplay(Error):
"""Cache is newer than repository, refusing to continue"""
def __init__(self, repository, key, manifest, path=None, sync=True):
self.timestamp = None
2010-12-21 20:29:09 +00:00
self.txn_active = False
2013-06-20 10:44:58 +00:00
self.repository = repository
self.key = key
self.manifest = manifest
self.path = path or os.path.join(get_cache_dir(), hexlify(repository.id).decode('ascii'))
2010-12-21 20:29:09 +00:00
if not os.path.exists(self.path):
self.create()
2010-03-06 17:25:35 +00:00
self.open()
if sync and self.manifest.id != self.manifest_id:
# If repository is older than the cache something fishy is going on
if self.timestamp and self.timestamp > manifest.timestamp:
raise self.RepositoryReplay()
2010-12-21 20:29:09 +00:00
self.sync()
self.commit()
2010-03-06 17:25:35 +00:00
2013-06-24 20:41:05 +00:00
def __del__(self):
self.close()
2010-12-21 20:29:09 +00:00
def create(self):
"""Create a new empty cache at `path`
2010-12-21 20:29:09 +00:00
"""
2011-01-04 22:00:39 +00:00
os.makedirs(self.path)
2013-06-03 11:45:48 +00:00
with open(os.path.join(self.path, 'README'), 'w') as fd:
2013-07-08 21:38:27 +00:00
fd.write('This is an Attic cache')
2010-12-21 20:29:09 +00:00
config = RawConfigParser()
config.add_section('cache')
config.set('cache', 'version', '1')
2013-06-20 10:44:58 +00:00
config.set('cache', 'repository', hexlify(self.repository.id).decode('ascii'))
config.set('cache', 'manifest', '')
2013-06-03 11:45:48 +00:00
with open(os.path.join(self.path, 'config'), 'w') as fd:
2010-12-21 20:29:09 +00:00
config.write(fd)
ChunkIndex().write(os.path.join(self.path, 'chunks').encode('utf-8'))
2013-06-03 11:45:48 +00:00
with open(os.path.join(self.path, 'files'), 'w') as fd:
pass # empty file
2010-12-21 20:29:09 +00:00
2010-03-06 17:25:35 +00:00
def open(self):
2010-12-21 20:29:09 +00:00
if not os.path.isdir(self.path):
2013-07-08 21:38:27 +00:00
raise Exception('%s Does not look like an Attic cache' % self.path)
self.lock = UpgradableLock(os.path.join(self.path, 'config'), exclusive=True)
2010-12-21 20:29:09 +00:00
self.rollback()
self.config = RawConfigParser()
self.config.read(os.path.join(self.path, 'config'))
if self.config.getint('cache', 'version') != 1:
2013-07-08 21:38:27 +00:00
raise Exception('%s Does not look like an Attic cache')
2013-06-20 10:44:58 +00:00
self.id = self.config.get('cache', 'repository')
self.manifest_id = unhexlify(self.config.get('cache', 'manifest'))
self.timestamp = self.config.get('cache', 'timestamp', fallback=None)
self.chunks = ChunkIndex.read(os.path.join(self.path, 'chunks').encode('utf-8'))
self.files = None
2011-07-02 18:39:35 +00:00
2013-06-24 20:41:05 +00:00
def close(self):
self.lock.release()
2013-06-24 20:41:05 +00:00
2011-07-02 18:39:35 +00:00
def _read_files(self):
2011-07-06 20:23:41 +00:00
self.files = {}
self._newest_mtime = 0
2010-12-21 20:29:09 +00:00
with open(os.path.join(self.path, 'files'), 'rb') as fd:
u = msgpack.Unpacker(use_list=True)
2010-12-21 20:29:09 +00:00
while True:
data = fd.read(64 * 1024)
if not data:
break
u.feed(data)
for path_hash, item in u:
item[0] += 1
self.files[path_hash] = msgpack.packb(item)
2010-12-21 20:29:09 +00:00
def begin_txn(self):
# Initialize transaction snapshot
txn_dir = os.path.join(self.path, 'txn.tmp')
os.mkdir(txn_dir)
shutil.copy(os.path.join(self.path, 'config'), txn_dir)
shutil.copy(os.path.join(self.path, 'chunks'), txn_dir)
shutil.copy(os.path.join(self.path, 'files'), txn_dir)
os.rename(os.path.join(self.path, 'txn.tmp'),
os.path.join(self.path, 'txn.active'))
self.txn_active = True
def commit(self):
"""Commit transaction
"""
2011-01-04 22:16:55 +00:00
if not self.txn_active:
return
if self.files is not None:
with open(os.path.join(self.path, 'files'), 'wb') as fd:
2014-06-03 21:10:52 +00:00
for path_hash, item in self.files.items():
# Discard cached files with the newest mtime to avoid
# issues with filesystem snapshots and mtime precision
2014-06-03 21:10:52 +00:00
item = msgpack.unpackb(item)
if item[0] < 10 and bigint_to_int(item[3]) < self._newest_mtime:
msgpack.pack((path_hash, item), fd)
2013-06-03 11:45:48 +00:00
self.config.set('cache', 'manifest', hexlify(self.manifest.id).decode('ascii'))
self.config.set('cache', 'timestamp', self.manifest.timestamp)
2010-12-21 20:29:09 +00:00
with open(os.path.join(self.path, 'config'), 'w') as fd:
self.config.write(fd)
self.chunks.write(os.path.join(self.path, 'chunks').encode('utf-8'))
2010-12-21 20:29:09 +00:00
os.rename(os.path.join(self.path, 'txn.active'),
os.path.join(self.path, 'txn.tmp'))
shutil.rmtree(os.path.join(self.path, 'txn.tmp'))
self.txn_active = False
def rollback(self):
"""Roll back partial and aborted transactions
"""
# Remove partial transaction
if os.path.exists(os.path.join(self.path, 'txn.tmp')):
shutil.rmtree(os.path.join(self.path, 'txn.tmp'))
2010-12-21 20:29:09 +00:00
# Roll back active transaction
txn_dir = os.path.join(self.path, 'txn.active')
if os.path.exists(txn_dir):
shutil.copy(os.path.join(txn_dir, 'config'), self.path)
shutil.copy(os.path.join(txn_dir, 'chunks'), self.path)
shutil.copy(os.path.join(txn_dir, 'files'), self.path)
os.rename(txn_dir, os.path.join(self.path, 'txn.tmp'))
if os.path.exists(os.path.join(self.path, 'txn.tmp')):
shutil.rmtree(os.path.join(self.path, 'txn.tmp'))
2010-12-21 20:29:09 +00:00
self.txn_active = False
def sync(self):
2010-03-06 17:25:35 +00:00
"""Initializes cache by fetching and reading all archive indicies
"""
def add(id, size, csize):
try:
count, size, csize = self.chunks[id]
self.chunks[id] = count + 1, size, csize
except KeyError:
self.chunks[id] = 1, size, csize
2010-12-21 20:29:09 +00:00
self.begin_txn()
2013-06-03 11:45:48 +00:00
print('Initializing cache...')
self.chunks.clear()
unpacker = msgpack.Unpacker()
repository = cache_if_remote(self.repository)
for name, info in self.manifest.archives.items():
archive_id = info[b'id']
cdata = repository.get(archive_id)
data = self.key.decrypt(archive_id, cdata)
add(archive_id, len(data), len(cdata))
archive = msgpack.unpackb(data)
if archive[b'version'] != 1:
raise Exception('Unknown archive metadata version')
decode_dict(archive, (b'name',))
2013-06-03 11:45:48 +00:00
print('Analyzing archive:', archive[b'name'])
for key, chunk in zip(archive[b'items'], repository.get_many(archive[b'items'])):
data = self.key.decrypt(key, chunk)
add(key, len(data), len(chunk))
unpacker.feed(data)
for item in unpacker:
if b'chunks' in item:
for chunk_id, size, csize in item[b'chunks']:
add(chunk_id, size, csize)
2010-03-06 17:25:35 +00:00
def add_chunk(self, id, data, stats):
2010-12-21 20:29:09 +00:00
if not self.txn_active:
self.begin_txn()
if self.seen_chunk(id):
return self.chunk_incref(id, stats)
size = len(data)
data = self.key.encrypt(data)
csize = len(data)
2013-06-20 10:44:58 +00:00
self.repository.put(id, data, wait=False)
self.chunks[id] = (1, size, csize)
stats.update(size, csize, True)
return id, size, csize
2010-03-06 17:25:35 +00:00
def seen_chunk(self, id):
return self.chunks.get(id, (0, 0, 0))[0]
2010-03-06 17:25:35 +00:00
def chunk_incref(self, id, stats):
2010-12-21 20:29:09 +00:00
if not self.txn_active:
self.begin_txn()
count, size, csize = self.chunks[id]
self.chunks[id] = (count + 1, size, csize)
stats.update(size, csize, False)
return id, size, csize
2010-03-06 17:25:35 +00:00
def chunk_decref(self, id, stats):
2010-12-21 20:29:09 +00:00
if not self.txn_active:
self.begin_txn()
count, size, csize = self.chunks[id]
if count == 1:
2010-12-21 20:29:09 +00:00
del self.chunks[id]
2013-06-20 10:44:58 +00:00
self.repository.delete(id, wait=False)
stats.update(-size, -csize, True)
else:
self.chunks[id] = (count - 1, size, csize)
stats.update(-size, -csize, False)
def file_known_and_unchanged(self, path_hash, st):
if self.files is None:
2011-07-02 18:39:35 +00:00
self._read_files()
2010-12-21 20:29:09 +00:00
entry = self.files.get(path_hash)
if not entry:
return None
entry = msgpack.unpackb(entry)
2014-06-03 20:00:34 +00:00
if entry[2] == st.st_size and bigint_to_int(entry[3]) == st_mtime_ns(st) and entry[1] == st.st_ino:
2010-10-26 18:50:30 +00:00
# reset entry age
entry[0] = 0
self.files[path_hash] = msgpack.packb(entry)
return entry[4]
else:
return None
2010-12-21 20:29:09 +00:00
def memorize_file(self, path_hash, st, ids):
# Entry: Age, inode, size, mtime, chunk ids
mtime_ns = st_mtime_ns(st)
2014-06-03 20:00:34 +00:00
self.files[path_hash] = msgpack.packb((0, st.st_ino, st.st_size, int_to_bigint(mtime_ns), ids))
self._newest_mtime = max(self._newest_mtime, mtime_ns)