X-Git-Url: https://git.arvados.org/arvados.git/blobdiff_plain/080c940d7a8134a6e277a53b7e45eb27e2b2c87f..3af38eca348413c6f11f6526b2ee2ca7cb53e348:/sdk/python/arvados/collection.py?ds=sidebyside diff --git a/sdk/python/arvados/collection.py b/sdk/python/arvados/collection.py index e38a6bd475..cf1a36f9fd 100644 --- a/sdk/python/arvados/collection.py +++ b/sdk/python/arvados/collection.py @@ -7,24 +7,26 @@ from future.utils import listitems, listvalues, viewkeys from builtins import str from past.builtins import basestring from builtins import object +import ciso8601 +import datetime +import errno import functools +import hashlib +import io import logging import os import re -import errno -import hashlib -import datetime -import ciso8601 -import time +import sys import threading +import time from collections import deque from stat import * -from .arvfile import split, _FileLikeObjectBase, ArvadosFile, ArvadosFileWriter, ArvadosFileReader, _BlockManager, synchronized, must_be_writable, NoopLock +from .arvfile import split, _FileLikeObjectBase, ArvadosFile, ArvadosFileWriter, ArvadosFileReader, WrappableFile, _BlockManager, synchronized, must_be_writable, NoopLock from .keep import KeepLocator, KeepClient from .stream import StreamReader -from ._normalize_stream import normalize_stream +from ._normalize_stream import normalize_stream, escape from ._ranges import Range, LocatorAndRange from .safeapi import ThreadSafeApiCache import arvados.config as config @@ -35,6 +37,21 @@ from arvados.retry import retry_method _logger = logging.getLogger('arvados.collection') + +if sys.version_info >= (3, 0): + TextIOWrapper = io.TextIOWrapper +else: + class TextIOWrapper(io.TextIOWrapper): + """To maintain backward compatibility, cast str to unicode in + write('foo'). + + """ + def write(self, data): + if isinstance(data, basestring): + data = unicode(data) + return super(TextIOWrapper, self).write(data) + + class CollectionBase(object): """Abstract base class for Collection classes.""" @@ -266,7 +283,7 @@ class CollectionWriter(CollectionBase): streampath, filename = split(streampath) if self._last_open and not self._last_open.closed: raise errors.AssertionError( - "can't open '{}' when '{}' is still open".format( + u"can't open '{}' when '{}' is still open".format( filename, self._last_open.name)) if streampath != self.current_stream_name(): self.start_new_stream(streampath) @@ -444,22 +461,22 @@ class ResumableCollectionWriter(CollectionWriter): writer._queued_file.seek(pos) except IOError as error: raise errors.StaleWriterStateError( - "failed to reopen active file {}: {}".format(path, error)) + u"failed to reopen active file {}: {}".format(path, error)) return writer def check_dependencies(self): for path, orig_stat in listitems(self._dependencies): if not S_ISREG(orig_stat[ST_MODE]): - raise errors.StaleWriterStateError("{} not file".format(path)) + raise errors.StaleWriterStateError(u"{} not file".format(path)) try: now_stat = tuple(os.stat(path)) except OSError as error: raise errors.StaleWriterStateError( - "failed to stat {}: {}".format(path, error)) + u"failed to stat {}: {}".format(path, error)) if ((not S_ISREG(now_stat[ST_MODE])) or (orig_stat[ST_MTIME] != now_stat[ST_MTIME]) or (orig_stat[ST_SIZE] != now_stat[ST_SIZE])): - raise errors.StaleWriterStateError("{} changed".format(path)) + raise errors.StaleWriterStateError(u"{} changed".format(path)) def dump_state(self, copy_func=lambda x: x): state = {attr: copy_func(getattr(self, attr)) @@ -475,7 +492,7 @@ class ResumableCollectionWriter(CollectionWriter): try: src_path = os.path.realpath(source) except Exception: - raise errors.AssertionError("{} not a file path".format(source)) + raise errors.AssertionError(u"{} not a file path".format(source)) try: path_stat = os.stat(src_path) except OSError as stat_error: @@ -488,10 +505,10 @@ class ResumableCollectionWriter(CollectionWriter): self._dependencies[source] = tuple(fd_stat) elif path_stat is None: raise errors.AssertionError( - "could not stat {}: {}".format(source, stat_error)) + u"could not stat {}: {}".format(source, stat_error)) elif path_stat.st_ino != fd_stat.st_ino: raise errors.AssertionError( - "{} changed between open and stat calls".format(source)) + u"{} changed between open and stat calls".format(source)) else: self._dependencies[src_path] = tuple(fd_stat) @@ -520,6 +537,7 @@ class RichCollectionBase(CollectionBase): def __init__(self, parent=None): self.parent = parent self._committed = False + self._has_remote_blocks = False self._callback = None self._items = {} @@ -544,6 +562,24 @@ class RichCollectionBase(CollectionBase): def stream_name(self): raise NotImplementedError() + + @synchronized + def has_remote_blocks(self): + """Recursively check for a +R segment locator signature.""" + + if self._has_remote_blocks: + return True + for item in self: + if self[item].has_remote_blocks(): + return True + return False + + @synchronized + def set_has_remote_blocks(self, val): + self._has_remote_blocks = val + if self.parent: + self.parent.set_has_remote_blocks(val) + @must_be_writable @synchronized def find_or_create(self, path, create_type): @@ -636,7 +672,7 @@ class RichCollectionBase(CollectionBase): return self.find_or_create(path, COLLECTION) - def open(self, path, mode="r"): + def open(self, path, mode="r", encoding=None): """Open a file-like object for access. :path: @@ -658,6 +694,7 @@ class RichCollectionBase(CollectionBase): opens for reading and writing. All writes are appended to the end of the file. Writing does not affect the file pointer for reading. + """ if not re.search(r'^[rwa][bt]?\+?$', mode): @@ -680,7 +717,12 @@ class RichCollectionBase(CollectionBase): if mode[0] == 'w': arvfile.truncate(0) - return fclass(arvfile, mode=mode, num_retries=self.num_retries) + binmode = mode[0] + 'b' + re.sub('[bt]', '', mode[1:]) + f = fclass(arvfile, mode=binmode, num_retries=self.num_retries) + if 'b' not in mode: + bufferclass = io.BufferedRandom if f.writable() else io.BufferedReader + f = TextIOWrapper(bufferclass(WrappableFile(f)), encoding=encoding) + return f def modified(self): """Determine if the collection has been modified since last commited.""" @@ -832,6 +874,8 @@ class RichCollectionBase(CollectionBase): self._items[target_name] = item self.set_committed(False) + if not self._has_remote_blocks and source_obj.has_remote_blocks(): + self.set_has_remote_blocks(True) if modified_from: self.notify(MOD, self, target_name, (modified_from, item)) @@ -1015,7 +1059,9 @@ class RichCollectionBase(CollectionBase): if stream: buf.append(" ".join(normalize_stream(stream_name, stream)) + "\n") for dirname in [s for s in sorted_keys if isinstance(self[s], RichCollectionBase)]: - buf.append(self[dirname].manifest_text(stream_name=os.path.join(stream_name, dirname), strip=strip, normalize=True, only_committed=only_committed)) + buf.append(self[dirname].manifest_text( + stream_name=os.path.join(stream_name, dirname), + strip=strip, normalize=True, only_committed=only_committed)) return "".join(buf) else: if strip: @@ -1023,6 +1069,24 @@ class RichCollectionBase(CollectionBase): else: return self._manifest_text + @synchronized + def _copy_remote_blocks(self, remote_blocks={}): + """Scan through the entire collection and ask Keep to copy remote blocks. + + When accessing a remote collection, blocks will have a remote signature + (+R instead of +A). Collect these signatures and request Keep to copy the + blocks to the local cluster, returning local (+A) signatures. + + :remote_blocks: + Shared cache of remote to local block mappings. This is used to avoid + doing extra work when blocks are shared by more than one file in + different subdirectories. + + """ + for item in self: + remote_blocks = self[item]._copy_remote_blocks(remote_blocks) + return remote_blocks + @synchronized def diff(self, end_collection, prefix=".", holding_collection=None): """Generate list of add/modify/delete actions. @@ -1257,8 +1321,12 @@ class Collection(RichCollectionBase): self._manifest_locator = manifest_locator_or_text elif re.match(arvados.util.collection_uuid_pattern, manifest_locator_or_text): self._manifest_locator = manifest_locator_or_text + if not self._has_local_collection_uuid(): + self._has_remote_blocks = True elif re.match(arvados.util.manifest_pattern, manifest_locator_or_text): self._manifest_text = manifest_locator_or_text + if '+R' in self._manifest_text: + self._has_remote_blocks = True else: raise errors.ArgumentError( "Argument to CollectionReader is not a manifest or a collection UUID") @@ -1279,7 +1347,10 @@ class Collection(RichCollectionBase): def get_trash_at(self): if self._api_response and self._api_response["trash_at"]: - return ciso8601.parse_datetime(self._api_response["trash_at"]) + try: + return ciso8601.parse_datetime(self._api_response["trash_at"]) + except ValueError: + return None else: return None @@ -1376,6 +1447,10 @@ class Collection(RichCollectionBase): def _has_collection_uuid(self): return self._manifest_locator is not None and re.match(arvados.util.collection_uuid_pattern, self._manifest_locator) + def _has_local_collection_uuid(self): + return self._has_collection_uuid and \ + self._my_api()._rootDesc['uuidPrefix'] == self._manifest_locator.split('-')[0] + def __enter__(self): return self @@ -1505,8 +1580,14 @@ class Collection(RichCollectionBase): body["trash_at"] = t if not self.committed(): + if self._has_remote_blocks: + # Copy any remote blocks to the local cluster. + self._copy_remote_blocks(remote_blocks={}) + self._has_remote_blocks = False if not self._has_collection_uuid(): raise AssertionError("Collection manifest_locator is not a collection uuid. Use save_new() for new collections.") + elif not self._has_local_collection_uuid(): + raise AssertionError("Collection manifest_locator is from a remote cluster. Use save_new() to save it on the local cluster.") self._my_block_manager().commit_all() @@ -1591,6 +1672,11 @@ class Collection(RichCollectionBase): if trash_at and type(trash_at) is not datetime.datetime: raise errors.ArgumentError("trash_at must be datetime type.") + if self._has_remote_blocks: + # Copy any remote blocks to the local cluster. + self._copy_remote_blocks(remote_blocks={}) + self._has_remote_blocks = False + self._my_block_manager().commit_all() text = self.manifest_text(strip=False) @@ -1627,6 +1713,9 @@ class Collection(RichCollectionBase): _block_re = re.compile(r'[0-9a-f]{32}\+(\d+)(\+\S+)*') _segment_re = re.compile(r'(\d+):(\d+):(\S+)') + def _unescape_manifest_path(self, path): + return re.sub('\\\\([0-3][0-7][0-7])', lambda m: chr(int(m.group(1), 8)), path) + @synchronized def _import_manifest(self, manifest_text): """Import a manifest into a `Collection`. @@ -1651,7 +1740,7 @@ class Collection(RichCollectionBase): if state == STREAM_NAME: # starting a new stream - stream_name = tok.replace('\\040', ' ') + stream_name = self._unescape_manifest_path(tok) blocks = [] segments = [] streamoffset = 0 @@ -1673,13 +1762,18 @@ class Collection(RichCollectionBase): if file_segment: pos = int(file_segment.group(1)) size = int(file_segment.group(2)) - name = file_segment.group(3).replace('\\040', ' ') - filepath = os.path.join(stream_name, name) - afile = self.find_or_create(filepath, FILE) - if isinstance(afile, ArvadosFile): - afile.add_segment(blocks, pos, size) + name = self._unescape_manifest_path(file_segment.group(3)) + if name.split('/')[-1] == '.': + # placeholder for persisting an empty directory, not a real file + if len(name) > 2: + self.find_or_create(os.path.join(stream_name, name[:-2]), COLLECTION) else: - raise errors.SyntaxError("File %s conflicts with stream of the same name.", filepath) + filepath = os.path.join(stream_name, name) + afile = self.find_or_create(filepath, FILE) + if isinstance(afile, ArvadosFile): + afile.add_segment(blocks, pos, size) + else: + raise errors.SyntaxError("File %s conflicts with stream of the same name.", filepath) else: # error! raise errors.SyntaxError("Invalid manifest format, expected file segment but did not match format: '%s'" % tok) @@ -1745,6 +1839,16 @@ class Subcollection(RichCollectionBase): self.name = newname self.lock = self.parent.root_collection().lock + @synchronized + def _get_manifest_text(self, stream_name, strip, normalize, only_committed=False): + """Encode empty directories by using an \056-named (".") empty file""" + if len(self._items) == 0: + return "%s %s 0:0:\\056\n" % ( + escape(stream_name), config.EMPTY_BLOCK_LOCATOR) + return super(Subcollection, self)._get_manifest_text(stream_name, + strip, normalize, + only_committed) + class CollectionReader(Collection): """A read-only collection object.