[ARVADOS] updated: 7abed45cc9b1c5e816584b9f267f6c448384ffed

git at public.curoverse.com git at public.curoverse.com
Thu Dec 18 17:11:18 EST 2014


Summary of changes:
 sdk/python/arvados/arvfile.py        | 184 +++++++++++++-------------
 sdk/python/arvados/collection.py     |  93 ++++++++++++-
 sdk/python/arvados/stream.py         | 248 +++++++++++++++++------------------
 sdk/python/tests/test_collections.py |   8 ++
 4 files changed, 315 insertions(+), 218 deletions(-)

       via  7abed45cc9b1c5e816584b9f267f6c448384ffed (commit)
      from  cc946c07eade09974423955f8d4b080941f53c7b (commit)

Those revisions listed above that are new to this repository have
not appeared on any other notification email; so we list those
revisions in full, below.


commit 7abed45cc9b1c5e816584b9f267f6c448384ffed
Author: Peter Amstutz <peter.amstutz at curoverse.com>
Date:   Thu Dec 18 17:12:43 2014 -0500

    3198: New Collection/AravosFile design work in progress

diff --git a/sdk/python/arvados/arvfile.py b/sdk/python/arvados/arvfile.py
index ad52195..593d4fb 100644
--- a/sdk/python/arvados/arvfile.py
+++ b/sdk/python/arvados/arvfile.py
@@ -46,7 +46,7 @@ class ArvadosFileBase(object):
         self.closed = True
 
 
-class StreamFileReader(ArvadosFileBase):
+class ArvadosFileReaderBase(ArvadosFileBase):
     class _NameAttribute(str):
         # The Python file API provides a plain .name attribute.
         # Older SDK provided a name() method.
@@ -54,13 +54,11 @@ class StreamFileReader(ArvadosFileBase):
         def __call__(self):
             return self
 
-
-    def __init__(self, stream, segments, name):
-        super(StreamFileReader, self).__init__(self._NameAttribute(name), 'rb')
-        self._stream = stream
-        self.segments = segments
+    def __init__(self, name, mode, num_retries=None):
+        super(ArvadosFileReaderBase, self).__init__(self._NameAttribute(name), mode)
         self._filepos = 0L
-        self.num_retries = stream.num_retries
+        self.num_retries = num_retries
+        self.need_lock = False
         self._readline_cache = (None, None)
 
     def __iter__(self):
@@ -73,9 +71,6 @@ class StreamFileReader(ArvadosFileBase):
     def decompressed_name(self):
         return re.sub('\.(bz2|gz)$', '', self.name)
 
-    def stream_name(self):
-        return self._stream.name()
-
     @ArvadosFileBase._before_close
     def seek(self, pos, whence=os.SEEK_CUR):
         if whence == os.SEEK_CUR:
@@ -87,45 +82,11 @@ class StreamFileReader(ArvadosFileBase):
     def tell(self):
         return self._filepos
 
-    def _size(self):
-        n = self.segments[-1]
-        return n[OFFSET] + n[BLOCKSIZE]
-
     def size(self):
         return self._size()
 
     @ArvadosFileBase._before_close
     @retry_method
-    def read(self, size, num_retries=None):
-        """Read up to 'size' bytes from the stream, starting at the current file position"""
-        if size == 0:
-            return ''
-
-        data = ''
-        available_chunks = locators_and_ranges(self.segments, self._filepos, size)
-        if available_chunks:
-            locator, blocksize, segmentoffset, segmentsize = available_chunks[0]
-            data = self._stream._readfrom(locator+segmentoffset, segmentsize,
-                                         num_retries=num_retries)
-
-        self._filepos += len(data)
-        return data
-
-    @ArvadosFileBase._before_close
-    @retry_method
-    def readfrom(self, start, size, num_retries=None):
-        """Read up to 'size' bytes from the stream, starting at 'start'"""
-        if size == 0:
-            return ''
-
-        data = []
-        for locator, blocksize, segmentoffset, segmentsize in locators_and_ranges(self.segments, start, size):
-            data.append(self._stream._readfrom(locator+segmentoffset, segmentsize,
-                                              num_retries=num_retries))
-        return ''.join(data)
-
-    @ArvadosFileBase._before_close
-    @retry_method
     def readall(self, size=2**20, num_retries=None):
         while True:
             data = self.read(size, num_retries=num_retries)
@@ -192,54 +153,89 @@ class StreamFileReader(ArvadosFileBase):
                 break
         return ''.join(data).splitlines(True)
 
+
+class StreamFileReader(ArvadosFileReaderBase):
+    def __init__(self, stream, segments, name):
+        super(StreamFileReader, self).__init__(name, 'rb')
+        self._stream = stream
+        self.segments = segments
+        self.num_retries = stream.num_retries
+        self._filepos = 0L
+        self.num_retries = stream.num_retries
+        self._readline_cache = (None, None)
+
+    def stream_name(self):
+        return self._stream.name()
+
+    def _size(self):
+        n = self.segments[-1]
+        return n[OFFSET] + n[BLOCKSIZE]
+
+    @ArvadosFileBase._before_close
+    @retry_method
+    def read(self, size, num_retries=None):
+        """Read up to 'size' bytes from the stream, starting at the current file position"""
+        if size == 0:
+            return ''
+
+        data = ''
+        available_chunks = locators_and_ranges(self.segments, self._filepos, size)
+        if available_chunks:
+            locator, blocksize, segmentoffset, segmentsize = available_chunks[0]
+            data = self._stream._readfrom(locator+segmentoffset, segmentsize,
+                                         num_retries=num_retries)
+
+        self._filepos += len(data)
+        return data
+
+    @ArvadosFileBase._before_close
+    @retry_method
+    def readfrom(self, start, size, num_retries=None):
+        """Read up to 'size' bytes from the stream, starting at 'start'"""
+        if size == 0:
+            return ''
+
+        data = []
+        for locator, blocksize, segmentoffset, segmentsize in locators_and_ranges(self.segments, start, size):
+            data.append(self._stream._readfrom(locator+segmentoffset, segmentsize,
+                                              num_retries=num_retries))
+        return ''.join(data)
+
     def as_manifest(self):
         manifest_text = ['.']
         manifest_text.extend([d[LOCATOR] for d in self._stream._data_locators])
         manifest_text.extend(["{}:{}:{}".format(seg[LOCATOR], seg[BLOCKSIZE], self.name().replace(' ', '\\040')) for seg in self.segments])
-        return arvados.CollectionReader(' '.join(manifest_text) + '\n').manifest_text(normalize=True)
+        return CollectionReader(' '.join(manifest_text) + '\n').manifest_text(normalize=True)
 
 
-class StreamFileWriter(StreamFileReader):
-    def __init__(self, stream, segments, name):
-        super(StreamFileWriter, self).__init__(stream, segments, name)
-        self.mode = 'wb'
-
-    # wrap superclass methods in mutex
-    def _proxy_method(name):
-        method = getattr(StreamFileReader, name)
-        @functools.wraps(method, ('__name__', '__doc__'))
-        def wrapper(self, *args, **kwargs):
-            with self._stream.mutex:
-                return method(self, *args, **kwargs)
-        return wrapper
-
-    for _method_name in ['__iter__', 'seek', 'tell', 'size', 'read', 'readfrom', 'readall', 'readline', 'decompress', 'readall_decompressed', 'readlines', 'as_manifest']:
-        locals()[_method_name] = _proxy_method(_method_name)
+class ArvadosFile(ArvadosFileReaderBase):
+    def __init__(self, name, mode, stream, segments):
+        super(ArvadosFile, self).__init__(name, mode)
+        self.segments = []
 
     def truncate(self, size=None):
-        with self._stream.mutex:
-            if size is None:
-                size = self._filepos
-
-            segs = locators_and_ranges(self.segments, 0, size)
-
-            newstream = []
-            self.segments = []
-            streamoffset = 0L
-            fileoffset = 0L
-
-            for seg in segs:
-                for locator, blocksize, segmentoffset, segmentsize in locators_and_ranges(self._stream._data_locators, seg[LOCATOR]+seg[OFFSET], seg[SEGMENTSIZE]):
-                    newstream.append([locator, blocksize, streamoffset])
-                    self.segments.append([streamoffset+segmentoffset, segmentsize, fileoffset])
-                    streamoffset += blocksize
-                    fileoffset += segmentsize
-            if len(newstream) == 0:
-                newstream.append(config.EMPTY_BLOCK_LOCATOR)
-                self.segments.append([0, 0, 0])
-            self._stream._data_locators = newstream
-            if self._filepos > fileoffset:
-                self._filepos = fileoffset
+        if size is None:
+            size = self._filepos
+
+        segs = locators_and_ranges(self.segments, 0, size)
+
+        newstream = []
+        self.segments = []
+        streamoffset = 0L
+        fileoffset = 0L
+
+        for seg in segs:
+            for locator, blocksize, segmentoffset, segmentsize in locators_and_ranges(self._stream._data_locators, seg[LOCATOR]+seg[OFFSET], seg[SEGMENTSIZE]):
+                newstream.append([locator, blocksize, streamoffset])
+                self.segments.append([streamoffset+segmentoffset, segmentsize, fileoffset])
+                streamoffset += blocksize
+                fileoffset += segmentsize
+        if len(newstream) == 0:
+            newstream.append(config.EMPTY_BLOCK_LOCATOR)
+            self.segments.append([0, 0, 0])
+        self._stream._data_locators = newstream
+        if self._filepos > fileoffset:
+            self._filepos = fileoffset
 
     def _writeto(self, offset, data):
         if offset > self._size():
@@ -248,19 +244,21 @@ class StreamFileWriter(StreamFileReader):
         replace_range(self.segments, self._filepos, len(data), self._stream._size()-len(data))
 
     def writeto(self, offset, data):
-        with self._stream.mutex:
-            self._writeto(offset, data)
+        self._writeto(offset, data)
 
     def write(self, data):
-        with self._stream.mutex:
-            self._writeto(self._filepos, data)
-            self._filepos += len(data)
+        self._writeto(self._filepos, data)
+        self._filepos += len(data)
 
     def writelines(self, seq):
-        with self._stream.mutex:
-            for s in seq:
-                self._writeto(self._filepos, s)
-                self._filepos += len(s)
+        for s in seq:
+            self._writeto(self._filepos, s)
+            self._filepos += len(s)
 
     def flush(self):
         pass
+
+    def add_segment(self, blocks, pos, size):
+        for locator, blocksize, segmentoffset, segmentsize in locators_and_ranges(blocks, pos, size):
+            last = self.segments[-1] if self.segments else [0, 0, 0]
+            self.segments.append([locator, segmentsize, last[OFFSET]+last[BLOCKSIZE], segmentoffset])
diff --git a/sdk/python/arvados/collection.py b/sdk/python/arvados/collection.py
index 24572ed..077115e 100644
--- a/sdk/python/arvados/collection.py
+++ b/sdk/python/arvados/collection.py
@@ -6,7 +6,7 @@ import re
 from collections import deque
 from stat import *
 
-from .arvfile import ArvadosFileBase, split
+from .arvfile import ArvadosFileBase, split, ArvadosFile
 from keep import *
 from .stream import StreamReader, normalize_stream
 import config
@@ -636,3 +636,94 @@ class ResumableCollectionWriter(CollectionWriter):
             raise errors.AssertionError(
                 "resumable writer can't accept unsourced data")
         return super(ResumableCollectionWriter, self).write(data)
+
+
+class Collection(object):
+    def __init__(self):
+        self.items = {}
+
+    def find_or_create(self, path):
+        p = path.split("/")
+        if p[0] == '.':
+            del p[0]
+
+        if len(p) > 0:
+            item = self.items.get(p[0])
+            if len(p) == 1:
+                # item must be a file
+                if item is None:
+                    # create new file
+                    item = ArvadosFile(p[0], 'wb', [], [])
+                    self.items[p[0]] = item
+                return item
+            else:
+                if item is None:
+                    # create new collection
+                    item = Collection()
+                    self.items[p[0]] = item
+                del p[0]
+                return item.find_or_create("/".join(p))
+        else:
+            return self
+
+
+def import_manifest(manifest_text):
+    c = Collection()
+
+    STREAM_NAME = 0
+    BLOCKS = 1
+    SEGMENTS = 2
+
+    stream_name = None
+    state = STREAM_NAME
+
+    for n in re.finditer(r'([^ \n]+)([ \n])', manifest_text):
+        tok = n.group(1)
+        sep = n.group(2)
+        if state == STREAM_NAME:
+            # starting a new stream
+            stream_name = tok.replace('\\040', ' ')
+            blocks = []
+            segments = []
+            streamoffset = 0L
+            state = BLOCKS
+            continue
+
+        if state == BLOCKS:
+            s = re.match(r'[0-9a-f]{32}\+(\d)+(\+\S+)*', tok)
+            if s:
+                blocksize = long(s.group(1))
+                blocks.append([tok, blocksize, streamoffset])
+                streamoffset += blocksize
+            else:
+                state = SEGMENTS
+
+        if state == SEGMENTS:
+            s = re.search(r'^(\d+):(\d+):(\S+)', tok)
+            if s:
+                pos = long(s.group(1))
+                size = long(s.group(2))
+                name = s.group(3).replace('\\040', ' ')
+                f = c.find_or_create("%s/%s" % (stream_name, name))
+                f.add_segment(blocks, pos, size)
+            else:
+                # error!
+                raise errors.SyntaxError("Invalid manifest format")
+
+        if sep == "\n":
+            stream_name = None
+            state = STREAM_NAME
+
+    return c
+
+def export_manifest(item, stream_name="."):
+    buf = ""
+    print item
+    if isinstance(item, Collection):
+        for i, j in item.items.values():
+            buf += export_manifest(j, stream_name)
+    else:
+        buf += stream_name
+        buf += " "
+        buf += item.segments
+    return buf
diff --git a/sdk/python/arvados/stream.py b/sdk/python/arvados/stream.py
index bbbbb95..ecccefa 100644
--- a/sdk/python/arvados/stream.py
+++ b/sdk/python/arvados/stream.py
@@ -7,7 +7,7 @@ import functools
 import copy
 
 from .ranges import *
-from .arvfile import ArvadosFileBase, StreamFileReader, StreamFileWriter
+from .arvfile import ArvadosFileBase, StreamFileReader
 from arvados.retry import retry_method
 from keep import *
 import config
@@ -168,126 +168,126 @@ class BufferBlock(object):
         self.locator_list_entry[1] = self.write_pointer
 
 
-class StreamWriter(StreamReader):
-    def __init__(self, tokens, keep=None, debug=False, _empty=False,
-                 num_retries=0):
-        super(StreamWriter, self).__init__(tokens, keep, debug, _empty, num_retries)
-
-        if len(self._files) != 1:
-            raise AssertionError("StreamWriter can only have one file at a time")
-        sr = self._files.popitem()[1]
-        self._files[sr.name] = StreamFileWriter(self, sr.segments, sr.name)
-
-        self.mutex = threading.Lock()
-        self.current_bblock = None
-        self.bufferblocks = {}
-
-    # wrap superclass methods in mutex
-    def _proxy_method(name):
-        method = getattr(StreamReader, name)
-        @functools.wraps(method, ('__name__', '__doc__'))
-        def wrapper(self, *args, **kwargs):
-            with self.mutex:
-                return method(self, *args, **kwargs)
-        return wrapper
-
-    for _method_name in ['files', 'all_files', 'size', 'locators_and_ranges', 'readfrom', 'manifest_text']:
-        locals()[_method_name] = _proxy_method(_method_name)
-
-    @retry_method
-    def _keepget(self, locator, num_retries=None):
-        if locator in self.bufferblocks:
-            bb = self.bufferblocks[locator]
-            return str(bb.buffer_block[0:bb.write_pointer])
-        else:
-            return self._keep.get(locator, num_retries=num_retries)
-
-    def _init_bufferblock(self):
-        last = self._data_locators[-1]
-        streamoffset = last[OFFSET] + last[BLOCKSIZE]
-        if last[BLOCKSIZE] == 0:
-            del self._data_locators[-1]
-        self.current_bblock = BufferBlock("bufferblock%i" % len(self.bufferblocks), streamoffset)
-        self.bufferblocks[self.current_bblock.locator] = self.current_bblock
-        self._data_locators.append(self.current_bblock.locator_list_entry)
-
-    def _repack_writes(self):
-        '''Test if the buffer block has more data than is referenced by actual segments
-        (this happens when a buffered write over-writes a file range written in
-        a previous buffered write).  Re-pack the buffer block for efficiency
-        and to avoid leaking information.
-        '''
-        segs = self._files.values()[0].segments
-
-        bufferblock_segs = []
-        i = 0
-        tmp_segs = copy.copy(segs)
-        while i < len(tmp_segs):
-            # Go through each segment and identify segments that include the buffer block
-            s = tmp_segs[i]
-            if s[LOCATOR] < self.current_bblock.locator_list_entry[OFFSET] and (s[LOCATOR] + s[BLOCKSIZE]) > self.current_bblock.locator_list_entry[OFFSET]:
-                # The segment straddles the previous block and the current buffer block.  Split the segment.
-                b1 = self.current_bblock.locator_list_entry[OFFSET] - s[LOCATOR]
-                b2 = (s[LOCATOR] + s[BLOCKSIZE]) - self.current_bblock.locator_list_entry[OFFSET]
-                bb_seg = [self.current_bblock.locator_list_entry[OFFSET], b2, s[OFFSET]+b1]
-                tmp_segs[i] = [s[LOCATOR], b1, s[OFFSET]]
-                tmp_segs.insert(i+1, bb_seg)
-                bufferblock_segs.append(bb_seg)
-                i += 1
-            elif s[LOCATOR] >= self.current_bblock.locator_list_entry[OFFSET]:
-                # The segment's data is in the buffer block.
-                bufferblock_segs.append(s)
-            i += 1
-
-        # Now sum up the segments to get the total bytes
-        # of the file referencing into the buffer block.
-        write_total = sum([s[BLOCKSIZE] for s in bufferblock_segs])
-
-        if write_total < self.current_bblock.locator_list_entry[BLOCKSIZE]:
-            # There is more data in the buffer block than is actually accounted for by segments, so
-            # re-pack into a new buffer by copying over to a new buffer block.
-            new_bb = BufferBlock(self.current_bblock.locator,
-                                 self.current_bblock.locator_list_entry[OFFSET],
-                                 starting_size=write_total)
-            for t in bufferblock_segs:
-                t_start = t[LOCATOR] - self.current_bblock.locator_list_entry[OFFSET]
-                t_end = t_start + t[BLOCKSIZE]
-                t[0] = self.current_bblock.locator_list_entry[OFFSET] + new_bb.write_pointer
-                new_bb.append(self.current_bblock.buffer_block[t_start:t_end])
-
-            self.current_bblock = new_bb
-            self.bufferblocks[self.current_bblock.locator] = self.current_bblock
-            self._data_locators[-1] = self.current_bblock.locator_list_entry
-            self._files.values()[0].segments = tmp_segs
-
-    def _commit(self):
-        # commit buffer block
-
-        # TODO: do 'put' in the background?
-        pdh = self._keep.put(self.current_bblock.buffer_block[0:self.current_bblock.write_pointer])
-        self._data_locators[-1][0] = pdh
-        self.current_bblock = None
-
-    def commit(self):
-        with self.mutex:
-            self._repack_writes()
-            self._commit()
-
-    def _append(self, data):
-        if len(data) > config.KEEP_BLOCK_SIZE:
-            raise ArgumentError("Please append data chunks smaller than config.KEEP_BLOCK_SIZE")
-
-        if self.current_bblock is None:
-            self._init_bufferblock()
-
-        if (self.current_bblock.write_pointer + len(data)) > config.KEEP_BLOCK_SIZE:
-            self._repack_writes()
-            if (self.current_bblock.write_pointer + len(data)) > config.KEEP_BLOCK_SIZE:
-                self._commit()
-                self._init_bufferblock()
-
-        self.current_bblock.append(data)
-
-    def append(self, data):
-        with self.mutex:
-            self._append(data)
+# class StreamWriter(StreamReader):
+#     def __init__(self, tokens, keep=None, debug=False, _empty=False,
+#                  num_retries=0):
+#         super(StreamWriter, self).__init__(tokens, keep, debug, _empty, num_retries)
+
+#         if len(self._files) != 1:
+#             raise AssertionError("StreamWriter can only have one file at a time")
+#         sr = self._files.popitem()[1]
+#         self._files[sr.name] = StreamFileWriter(self, sr.segments, sr.name)
+
+#         self.mutex = threading.Lock()
+#         self.current_bblock = None
+#         self.bufferblocks = {}
+
+#     # wrap superclass methods in mutex
+#     def _proxy_method(name):
+#         method = getattr(StreamReader, name)
+#         @functools.wraps(method, ('__name__', '__doc__'))
+#         def wrapper(self, *args, **kwargs):
+#             with self.mutex:
+#                 return method(self, *args, **kwargs)
+#         return wrapper
+
+#     for _method_name in ['files', 'all_files', 'size', 'locators_and_ranges', 'readfrom', 'manifest_text']:
+#         locals()[_method_name] = _proxy_method(_method_name)
+
+#     @retry_method
+#     def _keepget(self, locator, num_retries=None):
+#         if locator in self.bufferblocks:
+#             bb = self.bufferblocks[locator]
+#             return str(bb.buffer_block[0:bb.write_pointer])
+#         else:
+#             return self._keep.get(locator, num_retries=num_retries)
+
+#     def _init_bufferblock(self):
+#         last = self._data_locators[-1]
+#         streamoffset = last[OFFSET] + last[BLOCKSIZE]
+#         if last[BLOCKSIZE] == 0:
+#             del self._data_locators[-1]
+#         self.current_bblock = BufferBlock("bufferblock%i" % len(self.bufferblocks), streamoffset)
+#         self.bufferblocks[self.current_bblock.locator] = self.current_bblock
+#         self._data_locators.append(self.current_bblock.locator_list_entry)
+
+#     def _repack_writes(self):
+#         '''Test if the buffer block has more data than is referenced by actual segments
+#         (this happens when a buffered write over-writes a file range written in
+#         a previous buffered write).  Re-pack the buffer block for efficiency
+#         and to avoid leaking information.
+#         '''
+#         segs = self._files.values()[0].segments
+
+#         bufferblock_segs = []
+#         i = 0
+#         tmp_segs = copy.copy(segs)
+#         while i < len(tmp_segs):
+#             # Go through each segment and identify segments that include the buffer block
+#             s = tmp_segs[i]
+#             if s[LOCATOR] < self.current_bblock.locator_list_entry[OFFSET] and (s[LOCATOR] + s[BLOCKSIZE]) > self.current_bblock.locator_list_entry[OFFSET]:
+#                 # The segment straddles the previous block and the current buffer block.  Split the segment.
+#                 b1 = self.current_bblock.locator_list_entry[OFFSET] - s[LOCATOR]
+#                 b2 = (s[LOCATOR] + s[BLOCKSIZE]) - self.current_bblock.locator_list_entry[OFFSET]
+#                 bb_seg = [self.current_bblock.locator_list_entry[OFFSET], b2, s[OFFSET]+b1]
+#                 tmp_segs[i] = [s[LOCATOR], b1, s[OFFSET]]
+#                 tmp_segs.insert(i+1, bb_seg)
+#                 bufferblock_segs.append(bb_seg)
+#                 i += 1
+#             elif s[LOCATOR] >= self.current_bblock.locator_list_entry[OFFSET]:
+#                 # The segment's data is in the buffer block.
+#                 bufferblock_segs.append(s)
+#             i += 1
+
+#         # Now sum up the segments to get the total bytes
+#         # of the file referencing into the buffer block.
+#         write_total = sum([s[BLOCKSIZE] for s in bufferblock_segs])
+
+#         if write_total < self.current_bblock.locator_list_entry[BLOCKSIZE]:
+#             # There is more data in the buffer block than is actually accounted for by segments, so
+#             # re-pack into a new buffer by copying over to a new buffer block.
+#             new_bb = BufferBlock(self.current_bblock.locator,
+#                                  self.current_bblock.locator_list_entry[OFFSET],
+#                                  starting_size=write_total)
+#             for t in bufferblock_segs:
+#                 t_start = t[LOCATOR] - self.current_bblock.locator_list_entry[OFFSET]
+#                 t_end = t_start + t[BLOCKSIZE]
+#                 t[0] = self.current_bblock.locator_list_entry[OFFSET] + new_bb.write_pointer
+#                 new_bb.append(self.current_bblock.buffer_block[t_start:t_end])
+
+#             self.current_bblock = new_bb
+#             self.bufferblocks[self.current_bblock.locator] = self.current_bblock
+#             self._data_locators[-1] = self.current_bblock.locator_list_entry
+#             self._files.values()[0].segments = tmp_segs
+
+#     def _commit(self):
+#         # commit buffer block
+
+#         # TODO: do 'put' in the background?
+#         pdh = self._keep.put(self.current_bblock.buffer_block[0:self.current_bblock.write_pointer])
+#         self._data_locators[-1][0] = pdh
+#         self.current_bblock = None
+
+#     def commit(self):
+#         with self.mutex:
+#             self._repack_writes()
+#             self._commit()
+
+#     def _append(self, data):
+#         if len(data) > config.KEEP_BLOCK_SIZE:
+#             raise ArgumentError("Please append data chunks smaller than config.KEEP_BLOCK_SIZE")
+
+#         if self.current_bblock is None:
+#             self._init_bufferblock()
+
+#         if (self.current_bblock.write_pointer + len(data)) > config.KEEP_BLOCK_SIZE:
+#             self._repack_writes()
+#             if (self.current_bblock.write_pointer + len(data)) > config.KEEP_BLOCK_SIZE:
+#                 self._commit()
+#                 self._init_bufferblock()
+
+#         self.current_bblock.append(data)
+
+#     def append(self, data):
+#         with self.mutex:
+#             self._append(data)
diff --git a/sdk/python/tests/test_collections.py b/sdk/python/tests/test_collections.py
index c991154..9882db9 100644
--- a/sdk/python/tests/test_collections.py
+++ b/sdk/python/tests/test_collections.py
@@ -806,5 +806,13 @@ class CollectionWriterTestCase(unittest.TestCase, CollectionTestMixin):
         self.assertRaises(arvados.errors.AssertionError, writer.open, 'two')
 
 
+class NewCollectionTestCase(unittest.TestCase, CollectionTestMixin):
+    def test_import_manifest(self):
+        m1 = """. 5348b82a029fd9e971a811ce1f71360b+43 0:43:md5sum.txt
+. 085c37f02916da1cad16f93c54d899b7+41 0:41:md5sum.txt
+. 8b22da26f9f433dea0a10e5ec66d73ba+43 0:43:md5sum.txt
+"""
+        print arvados.export_manifest(arvados.import_manifest(m1))
+
 if __name__ == '__main__':
     unittest.main()

-----------------------------------------------------------------------


hooks/post-receive
-- 




More information about the arvados-commits mailing list