Skip to content

WIP: implement msgpack wrapper so that bytes and str roundtrip #973

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Closed
ThomasWaldmann wants to merge 5 commits into from
Closed

WIP: implement msgpack wrapper so that bytes and str roundtrip #973

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
5 commits
Select commit Hold shift + click to select a range
fa7ccca
WIP: implement msgpack wrapper so that bytes and str roundtrip
ThomasWaldmann Apr 23, 2016
9ad1d31
add note about msgpack usage in remote.py
ThomasWaldmann Apr 23, 2016
e9c8e47
add note about msgpack usage in key.py
ThomasWaldmann Apr 23, 2016
3a1b83d
msgpack / msg_pack: add some tests
ThomasWaldmann Apr 23, 2016
dda3c80
adapt fuse to msg_pack
ThomasWaldmann Apr 23, 2016
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
278 changes: 138 additions & 140 deletions borg/archive.py
View file
Open in desktop

Large diffs are not rendered by default.

127 changes: 64 additions & 63 deletions borg/archiver.py
View file
Open in desktop

Large diffs are not rendered by default.

36 changes: 18 additions & 18 deletions borg/cache.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -12,10 +12,9 @@
from .helpers import Error, get_cache_dir, decode_dict, int_to_bigint, \
bigint_to_int, format_file_size, yes
from .locking import UpgradableLock
from . import msg_pack
from .hashindex import ChunkIndex, ChunkIndexEntry

import msgpack

ChunkListEntry = namedtuple('ChunkListEntry', 'id size csize')
FileCacheEntry = namedtuple('FileCacheEntry', 'age inode size mtime chunk_ids')

Expand Down Expand Up @@ -179,7 +178,7 @@ def _read_files(self):
self._newest_mtime = 0
logger.debug('Reading files cache ...')
with open(os.path.join(self.path, 'files'), 'rb') as fd:
u = msgpack.Unpacker(use_list=True)
u = msg_pack.Unpacker(use_list=True)
while True:
data = fd.read(64 * 1024)
if not data:
Expand All @@ -188,7 +187,7 @@ def _read_files(self):
for path_hash, item in u:
entry = FileCacheEntry(*item)
# in the end, this takes about 240 Bytes per file
self.files[path_hash] = msgpack.packb(entry._replace(age=entry.age + 1))
self.files[path_hash] = msg_pack.packb(entry._replace(age=entry.age + 1))

def begin_txn(self):
# Initialize transaction snapshot
Expand All @@ -211,9 +210,10 @@ def commit(self):
for path_hash, item in self.files.items():
# Discard cached files with the newest mtime to avoid
# issues with filesystem snapshots and mtime precision
entry = FileCacheEntry(*msgpack.unpackb(item))
entry = FileCacheEntry(*msg_pack.unpackb(item))
if entry.age < 10 and bigint_to_int(entry.mtime) < self._newest_mtime:
msgpack.pack((path_hash, entry), fd)
data = msg_pack.packb((path_hash, entry))
fd.write(data)
self.config.set('cache', 'manifest', hexlify(self.manifest.id).decode('ascii'))
self.config.set('cache', 'timestamp', self.manifest.timestamp)
self.config.set('cache', 'key_type', str(self.key.TYPE))
Expand Down Expand Up @@ -270,7 +270,7 @@ def cached_archives():
return set()

def repo_archives():
return set(info[b'id'] for info in self.manifest.archives.values())
return set(info['id'] for info in self.manifest.archives.values())

def cleanup_outdated(ids):
for id in ids:
Expand All @@ -281,21 +281,21 @@ def fetch_and_build_idx(archive_id, repository, key):
cdata = repository.get(archive_id)
_, data = key.decrypt(archive_id, cdata)
chunk_idx.add(archive_id, 1, len(data), len(cdata))
archive = msgpack.unpackb(data)
if archive[b'version'] != 1:
archive = msg_pack.unpackb(data)
if archive['version'] != 1:
raise Exception('Unknown archive metadata version')
decode_dict(archive, (b'name',))
unpacker = msgpack.Unpacker()
for item_id, chunk in zip(archive[b'items'], repository.get_many(archive[b'items'])):
decode_dict(archive, ('name', ))
unpacker = msg_pack.Unpacker()
for item_id, chunk in zip(archive['items'], repository.get_many(archive['items'])):
_, data = key.decrypt(item_id, chunk)
chunk_idx.add(item_id, 1, len(data), len(chunk))
unpacker.feed(data)
for item in unpacker:
if not isinstance(item, dict):
logger.error('Error: Did not get expected metadata dict - archive corrupted!')
continue
if b'chunks' in item:
for chunk_id, size, csize in item[b'chunks']:
if 'chunks' in item:
for chunk_id, size, csize in item['chunks']:
chunk_idx.add(chunk_id, 1, size, csize)
if self.do_cache:
fn = mkpath(archive_id)
Expand All @@ -310,7 +310,7 @@ def fetch_and_build_idx(archive_id, repository, key):

def lookup_name(archive_id):
for name, info in self.manifest.archives.items():
if info[b'id'] == archive_id:
if info['id'] == archive_id:
return name

def create_master_idx(chunk_idx):
Expand Down Expand Up @@ -417,10 +417,10 @@ def file_known_and_unchanged(self, path_hash, st, ignore_inode=False):
entry = self.files.get(path_hash)
if not entry:
return None
entry = FileCacheEntry(*msgpack.unpackb(entry))
entry = FileCacheEntry(*msg_pack.unpackb(entry))
if (entry.size == st.st_size and bigint_to_int(entry.mtime) == st.st_mtime_ns and
(ignore_inode or entry.inode == st.st_ino)):
self.files[path_hash] = msgpack.packb(entry._replace(age=0))
self.files[path_hash] = msg_pack.packb(entry._replace(age=0))
return entry.chunk_ids
else:
return None
Expand All @@ -429,5 +429,5 @@ def memorize_file(self, path_hash, st, ids):
if not (self.do_files and stat.S_ISREG(st.st_mode)):
return
entry = FileCacheEntry(age=0, inode=st.st_ino, size=st.st_size, mtime=int_to_bigint(st.st_mtime_ns), chunk_ids=ids)
self.files[path_hash] = msgpack.packb(entry)
self.files[path_hash] = msg_pack.packb(entry)
self._newest_mtime = max(self._newest_mtime, st.st_mtime_ns)
10 changes: 5 additions & 5 deletions borg/constants.py
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -1,10 +1,10 @@
# this set must be kept complete, otherwise the RobustUnpacker might malfunction:
ITEM_KEYS = set([b'path', b'source', b'rdev', b'chunks', b'hardlink_master',
b'mode', b'user', b'group', b'uid', b'gid', b'mtime', b'atime', b'ctime',
b'xattrs', b'bsdflags', b'acl_nfs4', b'acl_access', b'acl_default', b'acl_extended', ])
ITEM_KEYS = set(['path', 'source', 'rdev', 'chunks', 'hardlink_master',
'mode', 'user', 'group', 'uid', 'gid', 'mtime', 'atime', 'ctime',
'xattrs', 'bsdflags', 'acl_nfs4', 'acl_access', 'acl_default', 'acl_extended', ])

ARCHIVE_TEXT_KEYS = (b'name', b'comment', b'hostname', b'username', b'time', b'time_end')
ITEM_TEXT_KEYS = (b'path', b'source', b'user', b'group')
ARCHIVE_TEXT_KEYS = ('name', 'comment', 'hostname', 'username', 'time', 'time_end')
ITEM_TEXT_KEYS = ('path', 'source', 'user', 'group')

# default umask, overriden by --umask, defaults to read/write only for owner
UMASK_DEFAULT = 0o077
Expand Down
70 changes: 35 additions & 35 deletions borg/fuse.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -9,7 +9,7 @@
from .archive import Archive
from .helpers import daemonize, bigint_to_int
from distutils.version import LooseVersion
import msgpack
from . import msg_pack

# Does this version of llfuse support ns precision?
have_fuse_xtime_ns = hasattr(llfuse.EntryAttributes, 'st_mtime_ns')
Expand All @@ -31,12 +31,12 @@ def __init__(self):

def add(self, item):
pos = self.fd.seek(0, io.SEEK_END)
self.fd.write(msgpack.packb(item))
self.fd.write(msg_pack.packb(item))
return pos + self.offset

def get(self, inode):
self.fd.seek(inode - self.offset, io.SEEK_SET)
return next(msgpack.Unpacker(self.fd, read_size=1024))
return next(msg_pack.Unpacker(self.fd, read_size=1024))


class FuseOperations(llfuse.Operations):
Expand All @@ -50,7 +50,7 @@ def __init__(self, key, repository, manifest, archive, cached_repo):
self.items = {}
self.parent = {}
self.contents = defaultdict(dict)
self.default_dir = {b'mode': 0o40755, b'mtime': int(time.time() * 1e9), b'uid': os.getuid(), b'gid': os.getgid()}
self.default_dir = {'mode': 0o40755, 'mtime': int(time.time() * 1e9), 'uid': os.getuid(), 'gid': os.getgid()}
self.pending_archives = {}
self.accounted_chunks = {}
self.cache = ItemCache()
Expand All @@ -71,13 +71,13 @@ def __init__(self, key, repository, manifest, archive, cached_repo):
def process_archive(self, archive, prefix=[]):
"""Build fuse inode hierarchy from archive metadata
"""
unpacker = msgpack.Unpacker()
for key, chunk in zip(archive.metadata[b'items'], self.repository.get_many(archive.metadata[b'items'])):
unpacker = msg_pack.Unpacker()
for key, chunk in zip(archive.metadata['items'], self.repository.get_many(archive.metadata['items'])):
_, data = self.key.decrypt(key, chunk)
unpacker.feed(data)
for item in unpacker:
segments = prefix + os.fsencode(os.path.normpath(item[b'path'])).split(b'/')
del item[b'path']
segments = prefix + os.fsencode(os.path.normpath(item['path'])).split(b'/')
del item['path']
num_segments = len(segments)
parent = 1
for i, segment in enumerate(segments, 1):
Expand All @@ -88,10 +88,10 @@ def process_archive(self, archive, prefix=[]):
self.parent[archive_inode] = parent
# Leaf segment?
if i == num_segments:
if b'source' in item and stat.S_ISREG(item[b'mode']):
inode = self._find_inode(item[b'source'], prefix)
if 'source' in item and stat.S_ISREG(item['mode']):
inode = self._find_inode(item['source'], prefix)
item = self.cache.get(inode)
item[b'nlink'] = item.get(b'nlink', 1) + 1
item['nlink'] = item.get('nlink', 1) + 1
self.items[inode] = item
else:
inode = self.cache.add(item)
Expand Down Expand Up @@ -142,7 +142,7 @@ def getattr(self, inode, ctx=None):
size = 0
dsize = 0
try:
for key, chunksize, _ in item[b'chunks']:
for key, chunksize, _ in item['chunks']:
size += chunksize
if self.accounted_chunks.get(key, inode) == inode:
self.accounted_chunks[key] = inode
Expand All @@ -154,45 +154,45 @@ def getattr(self, inode, ctx=None):
entry.generation = 0
entry.entry_timeout = 300
entry.attr_timeout = 300
entry.st_mode = item[b'mode']
entry.st_nlink = item.get(b'nlink', 1)
entry.st_uid = item[b'uid']
entry.st_gid = item[b'gid']
entry.st_rdev = item.get(b'rdev', 0)
entry.st_mode = item['mode']
entry.st_nlink = item.get('nlink', 1)
entry.st_uid = item['uid']
entry.st_gid = item['gid']
entry.st_rdev = item.get('rdev', 0)
entry.st_size = size
entry.st_blksize = 512
entry.st_blocks = dsize / 512
# note: older archives only have mtime (not atime nor ctime)
if have_fuse_xtime_ns:
entry.st_mtime_ns = bigint_to_int(item[b'mtime'])
if b'atime' in item:
entry.st_atime_ns = bigint_to_int(item[b'atime'])
entry.st_mtime_ns = bigint_to_int(item['mtime'])
if 'atime' in item:
entry.st_atime_ns = bigint_to_int(item['atime'])
else:
entry.st_atime_ns = bigint_to_int(item[b'mtime'])
if b'ctime' in item:
entry.st_ctime_ns = bigint_to_int(item[b'ctime'])
entry.st_atime_ns = bigint_to_int(item['mtime'])
if 'ctime' in item:
entry.st_ctime_ns = bigint_to_int(item['ctime'])
else:
entry.st_ctime_ns = bigint_to_int(item[b'mtime'])
entry.st_ctime_ns = bigint_to_int(item['mtime'])
else:
entry.st_mtime = bigint_to_int(item[b'mtime']) / 1e9
if b'atime' in item:
entry.st_atime = bigint_to_int(item[b'atime']) / 1e9
entry.st_mtime = bigint_to_int(item['mtime']) / 1e9
if 'atime' in item:
entry.st_atime = bigint_to_int(item['atime']) / 1e9
else:
entry.st_atime = bigint_to_int(item[b'mtime']) / 1e9
if b'ctime' in item:
entry.st_ctime = bigint_to_int(item[b'ctime']) / 1e9
entry.st_atime = bigint_to_int(item['mtime']) / 1e9
if 'ctime' in item:
entry.st_ctime = bigint_to_int(item['ctime']) / 1e9
else:
entry.st_ctime = bigint_to_int(item[b'mtime']) / 1e9
entry.st_ctime = bigint_to_int(item['mtime']) / 1e9
return entry

def listxattr(self, inode, ctx=None):
item = self.get_item(inode)
return item.get(b'xattrs', {}).keys()
return item.get('xattrs', {}).keys()

def getxattr(self, inode, name, ctx=None):
item = self.get_item(inode)
try:
return item.get(b'xattrs', {})[name]
return item.get('xattrs', {})[name]
except KeyError:
raise llfuse.FUSEError(errno.ENODATA) from None

Expand Down Expand Up @@ -224,7 +224,7 @@ def opendir(self, inode, ctx=None):
def read(self, fh, offset, size):
parts = []
item = self.get_item(fh)
for id, s, csize in item[b'chunks']:
for id, s, csize in item['chunks']:
if s < offset:
offset -= s
continue
Expand All @@ -245,7 +245,7 @@ def readdir(self, fh, off):

def readlink(self, inode, ctx=None):
item = self.get_item(inode)
return os.fsencode(item[b'source'])
return os.fsencode(item['source'])

def mount(self, mountpoint, extra_options, foreground=False):
options = ['fsname=borgfs', 'ro']
Expand Down
Loading