[ARVADOS] updated: 8192c879caaca60902ca362b4967cf4492d08e0c
git at public.curoverse.com
git at public.curoverse.com
Tue Feb 3 14:24:34 EST 2015
Summary of changes:
doc/sdk/cli/index.html.textile.liquid | 44 +++++++++++++++-
doc/sdk/cli/reference.html.textile.liquid | 13 +----
sdk/python/arvados/arvfile.py | 76 ++++++++++++++++-----------
sdk/python/arvados/collection.py | 87 +++++++++++++++++--------------
sdk/python/tests/test_arvfile.py | 60 +++++++++++++--------
sdk/python/tests/test_keep_client.py | 4 +-
sdk/python/tests/test_websockets.py | 21 ++++----
7 files changed, 188 insertions(+), 117 deletions(-)
via 8192c879caaca60902ca362b4967cf4492d08e0c (commit)
via f56d3a6876f246f78d5bc231a0ac5b6e4c6bdb9c (commit)
via 89fb910b523686fdf725691c44cb4c63ba464487 (commit)
via 487dc40959c4a7a8838624c6e108236320e79c8d (commit)
from 070320475e3379be2d6c79bfd581d08e50274d33 (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 8192c879caaca60902ca362b4967cf4492d08e0c
Author: Peter Amstutz <peter.amstutz at curoverse.com>
Date: Tue Feb 3 14:26:44 2015 -0500
4823: Fix bugs, fix tests, existing tests pass again. Still need to write new tests.
diff --git a/sdk/python/arvados/arvfile.py b/sdk/python/arvados/arvfile.py
index af19d1f..9d0a069 100644
--- a/sdk/python/arvados/arvfile.py
+++ b/sdk/python/arvados/arvfile.py
@@ -10,6 +10,7 @@ import hashlib
import threading
import Queue
import copy
+import errno
def split(path):
"""split(path) -> streamname, filename
@@ -218,24 +219,24 @@ class StreamFileReader(ArvadosFileReaderBase):
class BufferBlock(object):
-"""
-A BufferBlock is a stand-in for a Keep block that is in the process of being
-written. Writers can append to it, get the size, and compute the Keep locator.
+ """
+ A BufferBlock is a stand-in for a Keep block that is in the process of being
+ written. Writers can append to it, get the size, and compute the Keep locator.
-There are three valid states:
+ There are three valid states:
-WRITABLE
- Can append to block.
+ WRITABLE
+ Can append to block.
-PENDING
- Block is in the process of being uploaded to Keep, append is an error.
+ PENDING
+ Block is in the process of being uploaded to Keep, append is an error.
-COMMITTED
- The block has been written to Keep, its internal buffer has been
- released, fetching the block will fetch it via keep client (since we
- discarded the internal copy), and identifiers referring to the BufferBlock
- can be replaced with the block locator.
-"""
+ COMMITTED
+ The block has been written to Keep, its internal buffer has been
+ released, fetching the block will fetch it via keep client (since we
+ discarded the internal copy), and identifiers referring to the BufferBlock
+ can be replaced with the block locator.
+ """
WRITABLE = 0
PENDING = 1
COMMITTED = 2
@@ -319,11 +320,15 @@ class NoopLock(object):
def release(self):
pass
+SYNC_READONLY = 1
+SYNC_EXPLICIT = 2
+SYNC_LIVE = 3
+
def _must_be_writable(orig_func):
# Decorator for methods that read actual Collection data.
@functools.wraps(orig_func)
def wrapper(self, *args, **kwargs):
- if self.sync_mode() == SynchronizedCollectionBase.SYNC_READONLY:
+ if self.sync_mode() == SYNC_READONLY:
raise IOError((errno.EROFS, "Collection is read only"))
return orig_func(self, *args, **kwargs)
return wrapper
@@ -344,6 +349,9 @@ class BlockManager(object):
self._prefetch_queue = None
self._prefetch_threads = None
self.lock = threading.Lock()
+ self.prefetch_enabled = True
+ self.num_put_threads = 2
+ self.num_get_threads = 2
@_synchronized
def alloc_bufferblock(self, blockid=None, starting_capacity=2**14, owner=None):
@@ -450,9 +458,11 @@ class BlockManager(object):
# default download block cache in KeepClient.
self._put_queue = Queue.Queue(maxsize=2)
self._put_errors = Queue.Queue()
- self._put_threads = [threading.Thread(target=worker, args=(self,)),
- threading.Thread(target=worker, args=(self,))]
- for t in self._put_threads:
+
+ self._put_threads = []
+ for i in xrange(0, self.num_put_threads):
+ t = threading.Thread(target=worker, args=(self,))
+ self._put_threads.append(t)
t.daemon = True
t.start()
@@ -507,6 +517,10 @@ class BlockManager(object):
for the same block will not result in repeated downloads (unless the
block is evicted from the cache.) This method does not block.
"""
+
+ if not self.prefetch_enabled:
+ return
+
def worker(self):
"""Background downloader thread."""
while True:
@@ -523,9 +537,10 @@ class BlockManager(object):
return
if self._prefetch_threads is None:
self._prefetch_queue = Queue.Queue()
- self._prefetch_threads = [threading.Thread(target=worker, args=(self,)),
- threading.Thread(target=worker, args=(self,))]
- for t in self._prefetch_threads:
+ self._prefetch_threads = []
+ for i in xrange(0, self.num_get_threads):
+ t = threading.Thread(target=worker, args=(self,))
+ self._prefetch_threads.append(t)
t.daemon = True
t.start()
self._prefetch_queue.put(locator)
@@ -609,7 +624,7 @@ class ArvadosFile(object):
the file contents after `size` will be discarded. If `size` is greater
than the current size of the file, an IOError will be raised.
"""
- if size < self.size():
+ if size < self._size():
new_segs = []
for r in self._segments:
range_end = r.range_start+r.range_size
@@ -626,7 +641,7 @@ class ArvadosFile(object):
self._segments = new_segs
self._modified = True
- elif size > self.size():
+ elif size > self._size():
raise IOError("truncate() does not support extending the file size")
@_synchronized
@@ -634,7 +649,7 @@ class ArvadosFile(object):
"""
read upto `size` bytes from the file starting at `offset`.
"""
- if size == 0 or offset >= self.size():
+ if size == 0 or offset >= self._size():
return ''
data = []
@@ -666,7 +681,7 @@ class ArvadosFile(object):
if write_total < self._current_bblock.size():
# 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 = self.parent._my_block_manager().alloc_bufferblock(self._current_bblock.blockid, starting_size=write_total, owner=self)
+ new_bb = self.parent._my_block_manager().alloc_bufferblock(self._current_bblock.blockid, starting_capacity=write_total, owner=self)
for t in bufferblock_segs:
new_bb.append(self._current_bblock.buffer_view[t.segment_offset:t.segment_offset+t.range_size].tobytes())
t.segment_offset = new_bb.size() - t.range_size
@@ -683,7 +698,7 @@ class ArvadosFile(object):
if len(data) == 0:
return
- if offset > self.size():
+ if offset > self._size():
raise ArgumentError("Offset is past the end of the file")
if len(data) > config.KEEP_BLOCK_SIZE:
@@ -701,6 +716,7 @@ class ArvadosFile(object):
self._current_bblock = self.parent._my_block_manager().alloc_bufferblock(owner=self)
self._current_bblock.append(data)
+
replace_range(self._segments, offset, len(data), self._current_bblock.blockid, self._current_bblock.write_pointer - len(data))
@_must_be_writable
@@ -720,9 +736,7 @@ class ArvadosFile(object):
r = Range(lr.locator, last.range_start+last.range_size, lr.segment_size, lr.segment_offset)
self._segments.append(r)
-
- @_synchronized
- def size(self):
+ def _size(self):
"""Get the file size"""
if self._segments:
n = self._segments[-1]
@@ -730,6 +744,10 @@ class ArvadosFile(object):
else:
return 0
+ @_synchronized
+ def size(self):
+ """Get the file size"""
+ return self._size()
class ArvadosFileReader(ArvadosFileReaderBase):
def __init__(self, arvadosfile, name, mode="r", num_retries=None):
diff --git a/sdk/python/arvados/collection.py b/sdk/python/arvados/collection.py
index 22c4d66..dd49464 100644
--- a/sdk/python/arvados/collection.py
+++ b/sdk/python/arvados/collection.py
@@ -8,7 +8,7 @@ import time
from collections import deque
from stat import *
-from .arvfile import ArvadosFileBase, split, ArvadosFile, ArvadosFileWriter, ArvadosFileReader, BlockManager, _synchronized, _must_be_writable
+from .arvfile import ArvadosFileBase, split, ArvadosFile, ArvadosFileWriter, ArvadosFileReader, BlockManager, _synchronized, _must_be_writable, SYNC_READONLY, SYNC_EXPLICIT, SYNC_LIVE, NoopLock
from keep import *
from .stream import StreamReader, normalize_stream, locator_block_size
from .ranges import Range, LocatorAndRange
@@ -154,7 +154,6 @@ class CollectionReader(CollectionBase):
for sline in self._manifest_text.split("\n")
if sline]
- @staticmethod
def _populate_first(orig_func):
# Decorator for methods that read actual Collection data.
@functools.wraps(orig_func)
@@ -642,18 +641,13 @@ class ResumableCollectionWriter(CollectionWriter):
"resumable writer can't accept unsourced data")
return super(ResumableCollectionWriter, self).write(data)
+ADD = "add"
+DEL = "del"
class SynchronizedCollectionBase(CollectionBase):
- SYNC_READONLY = 1
- SYNC_EXPLICIT = 2
- SYNC_LIVE = 3
-
- ADD = "add"
- DEL = "del"
-
def __init__(self, parent=None):
self.parent = parent
- self._items = None
+ self._items = {}
def _my_api(self):
raise NotImplementedError()
@@ -694,7 +688,7 @@ class SynchronizedCollectionBase(CollectionBase):
component.
"""
- if create and self._sync_mode() == SynchronizedCollectionBase.SYNC_READONLY:
+ if create and self.sync_mode() == SYNC_READONLY:
raise IOError((errno.EROFS, "Collection is read only"))
p = path.split("/")
@@ -748,10 +742,11 @@ class SynchronizedCollectionBase(CollectionBase):
raise ArgumentError("Bad mode '%s'" % mode)
create = (mode != "r")
- if create and self._sync_mode() == SynchronizedCollectionBase.SYNC_READONLY:
+ if create and self.sync_mode() == SYNC_READONLY:
raise IOError((errno.EROFS, "Collection is read only"))
f = self.find(path, create=create)
+
if f is None:
raise IOError((errno.ENOENT, "File not found"))
if not isinstance(f, ArvadosFile):
@@ -761,9 +756,9 @@ class SynchronizedCollectionBase(CollectionBase):
f.truncate(0)
if mode == "r":
- return ArvadosFileReader(f, path, mode)
+ return ArvadosFileReader(f, path, mode, num_retries=self.num_retries)
else:
- return ArvadosFileWriter(f, path, mode)
+ return ArvadosFileWriter(f, path, mode, num_retries=self.num_retries)
@_synchronized
def modified(self):
@@ -851,7 +846,7 @@ class SynchronizedCollectionBase(CollectionBase):
if item is None:
raise IOError((errno.ENOENT, "File not found"))
if len(p) == 1:
- if isinstance(SynchronizedCollection, self._items[p[0]]) and len(self._items[p[0]]) > 0 and not rm_r:
+ if isinstance(self._items[p[0]], SynchronizedCollectionBase) and len(self._items[p[0]]) > 0 and not rm_r:
raise IOError((errno.ENOTEMPTY, "Subcollection not empty"))
del self._items[p[0]]
self.notify(self, DEL, p[0], None)
@@ -958,9 +953,9 @@ class Collection(SynchronizedCollectionBase):
config=None,
api_client=None,
keep_client=None,
- num_retries=0,
+ num_retries=None,
block_manager=None,
- sync=Collection.SYNC_READONLY):
+ sync=SYNC_READONLY):
""":manifest_locator_or_text:
One of Arvados collection UUID, block locator of
a manifest, raw manifest text, or None (to create an empty collection).
@@ -984,15 +979,13 @@ class Collection(SynchronizedCollectionBase):
Collection is read only. No synchronization. This mode will
also forego locking, which gives better performance.
:SYNC_EXPLICIT:
- Synchronize on explicit request via `merge()` or `save()`
+ Synchronize on explicit request via `update()` or `save()`
:SYNC_LIVE:
Synchronize with server in response to background websocket events,
on block write, or on file close.
"""
-
- self.parent = parent
- self._items = None
+ super(Collection, self).__init__(parent)
self._api_client = api_client
self._keep_client = keep_client
self._block_manager = block_manager
@@ -1004,6 +997,7 @@ class Collection(SynchronizedCollectionBase):
self._sync = sync
self.lock = threading.RLock()
self.callbacks = []
+ self.events = None
if manifest_locator_or_text:
if re.match(util.keep_locator_pattern, manifest_locator_or_text):
@@ -1021,11 +1015,11 @@ class Collection(SynchronizedCollectionBase):
if self._sync == SYNC_LIVE:
if not self._manifest_locator or not re.match(util.collection_uuid_pattern, self._manifest_locator):
raise errors.ArgumentError("Cannot SYNC_LIVE unless a collection uuid is specified")
- self.events = events.subscribe(arvados.api(), filters=[["object_uuid", "=", self._manifest_locator]], self.on_message)
+ self.events = events.subscribe(arvados.api(), [["object_uuid", "=", self._manifest_locator]], self.on_message)
@staticmethod
def create(name, owner_uuid=None, sync=SYNC_EXPLICIT):
- c = Collection(sync=SYNC_EXPLICIT)
+ c = Collection(sync=sync)
c.save_as(name, owner_uuid=owner_uuid, ensure_unique_name=True)
return c
@@ -1035,9 +1029,12 @@ class Collection(SynchronizedCollectionBase):
def sync_mode(self):
return self._sync
+ def on_message(self):
+ self.update()
+
@_synchronized
- def on_message():
- n = self._my_api().collections().get(uuid=self._manifest_locator, select=[["manifest_text"])).execute()
+ def update(self):
+ n = self._my_api().collections().get(uuid=self._manifest_locator, select=["manifest_text"]).execute()
other = import_collection(n["manifest_text"])
self.merge(other)
@@ -1092,7 +1089,6 @@ class Collection(SynchronizedCollectionBase):
return e
def _populate(self):
- self._items = {}
if self._manifest_locator is None and self._manifest_text is None:
return
error_via_api = None
@@ -1134,14 +1130,17 @@ class Collection(SynchronizedCollectionBase):
def __exit__(self, exc_type, exc_value, traceback):
"""Support scoped auto-commit in a with: block"""
- self.save(allow_no_locator=True)
+ if self._sync != SYNC_READONLY:
+ self.save(allow_no_locator=True)
if self._block_manager is not None:
self._block_manager.stop_threads()
@_synchronized
- def clone(self, new_parent=None, new_sync=Collection.SYNC_READONLY, new_config=self.config):
+ def clone(self, new_parent=None, new_sync=SYNC_READONLY, new_config=None):
+ if new_config is None:
+ new_config = self.config
c = Collection(parent=new_parent, config=new_config, sync=new_sync)
- if new_sync == Collection.SYNC_READONLY:
+ if new_sync == SYNC_READONLY:
c.lock = NoopLock()
c._items = {}
self._cloneinto(c)
@@ -1227,9 +1226,9 @@ class Collection(SynchronizedCollectionBase):
self.callbacks.remove(callback)
@_synchronized
- def notify(self, event):
+ def notify(self, collection, event, name, item):
for c in self.callbacks:
- c(event)
+ c(collection, event, name, item)
class Subcollection(SynchronizedCollectionBase):
"""This is a subdirectory within a collection that doesn't have its own API
@@ -1239,7 +1238,7 @@ class Subcollection(SynchronizedCollectionBase):
super(Subcollection, self).__init__(parent)
self.lock = parent._root_lock()
- def _root_lock():
+ def _root_lock(self):
return self.parent._root_lock()
def sync_mode(self):
@@ -1257,8 +1256,8 @@ class Subcollection(SynchronizedCollectionBase):
def _populate(self):
self.parent._populate()
- def notify(self, event):
- self.parent.notify(event)
+ def notify(self, collection, event, name, item):
+ self.parent.notify(collection, event, name, item)
@_synchronized
def clone(self, new_parent):
@@ -1267,7 +1266,12 @@ class Subcollection(SynchronizedCollectionBase):
self._cloneinto(c)
return c
-def import_manifest(manifest_text, into_collection=None, api_client=None, keep=None, num_retries=None):
+def import_manifest(manifest_text,
+ into_collection=None,
+ api_client=None,
+ keep=None,
+ num_retries=None,
+ sync=SYNC_READONLY):
"""Import a manifest into a `Collection`.
:manifest_text:
@@ -1283,15 +1287,21 @@ def import_manifest(manifest_text, into_collection=None, api_client=None, keep=N
:keep:
The keep client object that will be used when creating a new `Collection` object.
- num_retries
+ :num_retries:
the default number of api client and keep retries on error.
+
+ :sync:
+ Collection sync mode (only if into_collection is None)
"""
if into_collection is not None:
if len(into_collection) > 0:
raise ArgumentError("Can only import manifest into an empty collection")
c = into_collection
else:
- c = Collection(api_client=api_client, keep_client=keep, num_retries=num_retries)
+ c = Collection(api_client=api_client, keep_client=keep, num_retries=num_retries, sync=sync)
+
+ save_sync = c.sync_mode()
+ c._sync = None
STREAM_NAME = 0
BLOCKS = 1
@@ -1339,6 +1349,7 @@ def import_manifest(manifest_text, into_collection=None, api_client=None, keep=N
state = STREAM_NAME
c.set_unmodified()
+ c._sync = save_sync
return c
def export_manifest(item, stream_name=".", portable_locators=False):
@@ -1361,7 +1372,7 @@ def export_manifest(item, stream_name=".", portable_locators=False):
for k in [s for s in sorted_keys if isinstance(item[s], ArvadosFile)]:
v = item[k]
st = []
- for s in v.segments:
+ for s in v.segments():
loc = s.locator
if loc.startswith("bufferblock"):
loc = v.parent._my_block_manager()._bufferblocks[loc].locator()
diff --git a/sdk/python/tests/test_arvfile.py b/sdk/python/tests/test_arvfile.py
index 5d71ec4..4bc9c2e 100644
--- a/sdk/python/tests/test_arvfile.py
+++ b/sdk/python/tests/test_arvfile.py
@@ -10,6 +10,7 @@ import hashlib
import arvados
from arvados import ArvadosFile, ArvadosFileReader, Range, import_manifest, export_manifest, KeepLocator
+from arvados.arvfile import SYNC_READONLY, SYNC_EXPLICIT
import arvados_testutil as tutil
from test_stream import StreamFileReaderTestCase, StreamRetryTestMixin
@@ -56,7 +57,7 @@ class ArvadosFileWriterTestCase(unittest.TestCase):
"manifest_text":". 781e5e245d69b566979b86e28d23f2c7+10 0:8:count.txt\n"},
{"uuid":"zzzzz-4zz18-mockcollection0"})
with import_manifest('. 781e5e245d69b566979b86e28d23f2c7+10 0:10:count.txt\n',
- api_client=api, keep=keep) as c:
+ api_client=api, keep=keep, sync=SYNC_EXPLICIT) as c:
writer = c.open("count.txt", "r+")
self.assertEqual(writer.size(), 10)
writer.seek(5)
@@ -77,7 +78,7 @@ class ArvadosFileWriterTestCase(unittest.TestCase):
"manifest_text": ". 781e5e245d69b566979b86e28d23f2c7+10 acbd18db4cc2f85cedef654fccc4a4d8+3 0:13:count.txt\n"},
{"uuid":"zzzzz-4zz18-mockcollection0"})
with import_manifest('. 781e5e245d69b566979b86e28d23f2c7+10 0:10:count.txt\n',
- api_client=api, keep=keep) as c:
+ api_client=api, keep=keep, sync=SYNC_EXPLICIT) as c:
writer = c.open("count.txt", "r+")
writer.seek(5, os.SEEK_SET)
self.assertEqual("56789", writer.read(8))
@@ -98,7 +99,7 @@ class ArvadosFileWriterTestCase(unittest.TestCase):
def test_write0(self):
keep = ArvadosFileWriterTestCase.MockKeep({"781e5e245d69b566979b86e28d23f2c7+10": "0123456789"})
with import_manifest('. 781e5e245d69b566979b86e28d23f2c7+10 0:10:count.txt\n',
- keep=keep) as c:
+ keep=keep, sync=SYNC_EXPLICIT) as c:
writer = c.open("count.txt", "r+")
self.assertEqual("0123456789", writer.readfrom(0, 13))
writer.seek(0, os.SEEK_SET)
@@ -110,7 +111,7 @@ class ArvadosFileWriterTestCase(unittest.TestCase):
def test_write1(self):
keep = ArvadosFileWriterTestCase.MockKeep({"781e5e245d69b566979b86e28d23f2c7+10": "0123456789"})
with import_manifest('. 781e5e245d69b566979b86e28d23f2c7+10 0:10:count.txt\n',
- keep=keep) as c:
+ keep=keep, sync=SYNC_EXPLICIT) as c:
writer = c.open("count.txt", "r+")
self.assertEqual("0123456789", writer.readfrom(0, 13))
writer.seek(3, os.SEEK_SET)
@@ -122,7 +123,7 @@ class ArvadosFileWriterTestCase(unittest.TestCase):
def test_write2(self):
keep = ArvadosFileWriterTestCase.MockKeep({"781e5e245d69b566979b86e28d23f2c7+10": "0123456789"})
with import_manifest('. 781e5e245d69b566979b86e28d23f2c7+10 0:10:count.txt\n',
- keep=keep) as c:
+ keep=keep, sync=SYNC_EXPLICIT) as c:
writer = c.open("count.txt", "r+")
self.assertEqual("0123456789", writer.readfrom(0, 13))
writer.seek(7, os.SEEK_SET)
@@ -134,7 +135,7 @@ class ArvadosFileWriterTestCase(unittest.TestCase):
def test_write3(self):
keep = ArvadosFileWriterTestCase.MockKeep({"781e5e245d69b566979b86e28d23f2c7+10": "0123456789"})
with import_manifest('. 781e5e245d69b566979b86e28d23f2c7+10 0:10:count.txt 0:10:count.txt\n',
- keep=keep) as c:
+ keep=keep, sync=SYNC_EXPLICIT) as c:
writer = c.open("count.txt", "r+")
self.assertEqual("012345678901234", writer.readfrom(0, 15))
writer.seek(7, os.SEEK_SET)
@@ -146,7 +147,7 @@ class ArvadosFileWriterTestCase(unittest.TestCase):
def test_write4(self):
keep = ArvadosFileWriterTestCase.MockKeep({"781e5e245d69b566979b86e28d23f2c7+10": "0123456789"})
with import_manifest('. 781e5e245d69b566979b86e28d23f2c7+10 0:4:count.txt 0:4:count.txt 0:4:count.txt',
- keep=keep) as c:
+ keep=keep, sync=SYNC_EXPLICIT) as c:
writer = c.open("count.txt", "r+")
self.assertEqual("012301230123", writer.readfrom(0, 15))
writer.seek(2, os.SEEK_SET)
@@ -161,7 +162,7 @@ class ArvadosFileWriterTestCase(unittest.TestCase):
"manifest_text": ". a5de24f4417cfba9d5825eadc2f4ca49+67108000 598cc1a4ccaef8ab6e4724d87e675d78+32892000 0:100000000:count.txt\n"},
{"uuid":"zzzzz-4zz18-mockcollection0"})
with import_manifest('. ' + arvados.config.EMPTY_BLOCK_LOCATOR + ' 0:0:count.txt',
- api_client=api, keep=keep) as c:
+ api_client=api, keep=keep, sync=SYNC_EXPLICIT) as c:
writer = c.open("count.txt", "r+")
text = ''.join(["0123456789" for a in xrange(0, 100)])
for b in xrange(0, 100000):
@@ -177,7 +178,7 @@ class ArvadosFileWriterTestCase(unittest.TestCase):
def test_write_rewrite0(self):
keep = ArvadosFileWriterTestCase.MockKeep({})
with import_manifest('. ' + arvados.config.EMPTY_BLOCK_LOCATOR + ' 0:0:count.txt',
- keep=keep) as c:
+ keep=keep, sync=SYNC_EXPLICIT) as c:
writer = c.open("count.txt", "r+")
for b in xrange(0, 10):
writer.seek(0, os.SEEK_SET)
@@ -190,7 +191,7 @@ class ArvadosFileWriterTestCase(unittest.TestCase):
def test_write_rewrite1(self):
keep = ArvadosFileWriterTestCase.MockKeep({"781e5e245d69b566979b86e28d23f2c7+10": "0123456789"})
with import_manifest('. 781e5e245d69b566979b86e28d23f2c7+10 0:10:count.txt',
- keep=keep) as c:
+ keep=keep, sync=SYNC_EXPLICIT) as c:
writer = c.open("count.txt", "r+")
for b in xrange(0, 10):
writer.seek(10, os.SEEK_SET)
@@ -203,7 +204,7 @@ class ArvadosFileWriterTestCase(unittest.TestCase):
def test_write_rewrite2(self):
keep = ArvadosFileWriterTestCase.MockKeep({"781e5e245d69b566979b86e28d23f2c7+10": "0123456789"})
with import_manifest('. 781e5e245d69b566979b86e28d23f2c7+10 0:10:count.txt',
- keep=keep) as c:
+ keep=keep, sync=SYNC_EXPLICIT) as c:
writer = c.open("count.txt", "r+")
for b in xrange(0, 10):
writer.seek(5, os.SEEK_SET)
@@ -219,7 +220,7 @@ class ArvadosFileWriterTestCase(unittest.TestCase):
"manifest_text": ". 37400a68af9abdd76ca5bf13e819e42a+32892003 a5de24f4417cfba9d5825eadc2f4ca49+67108000 32892000:3:count.txt 32892006:67107997:count.txt 0:32892000:count.txt\n"},
{"uuid":"zzzzz-4zz18-mockcollection0"})
with import_manifest('. ' + arvados.config.EMPTY_BLOCK_LOCATOR + ' 0:0:count.txt',
- api_client=api, keep=keep) as c:
+ api_client=api, keep=keep, sync=SYNC_EXPLICIT) as c:
writer = c.open("count.txt", "r+")
text = ''.join(["0123456789" for a in xrange(0, 100)])
for b in xrange(0, 100000):
@@ -239,7 +240,7 @@ class ArvadosFileWriterTestCase(unittest.TestCase):
api = ArvadosFileWriterTestCase.MockApi({"name":"test_create",
"manifest_text":". 2e9ec317e197819358fbc43afca7d837+8 0:8:count.txt\n"},
{"uuid":"zzzzz-4zz18-mockcollection0"})
- with arvados.Collection(api_client=api, keep_client=keep) as c:
+ with arvados.Collection(api_client=api, keep_client=keep, sync=SYNC_EXPLICIT) as c:
writer = c.open("count.txt", "w+")
self.assertEqual(writer.size(), 0)
writer.write("01234567")
@@ -259,7 +260,7 @@ class ArvadosFileWriterTestCase(unittest.TestCase):
api = ArvadosFileWriterTestCase.MockApi({"name":"test_create",
"manifest_text":"./foo/bar 2e9ec317e197819358fbc43afca7d837+8 0:8:count.txt\n"},
{"uuid":"zzzzz-4zz18-mockcollection0"})
- with arvados.Collection(api_client=api, keep_client=keep) as c:
+ with arvados.Collection(api_client=api, keep_client=keep, sync=SYNC_EXPLICIT) as c:
writer = c.open("foo/bar/count.txt", "w+")
writer.write("01234567")
c.save_as("test_create")
@@ -270,7 +271,7 @@ class ArvadosFileWriterTestCase(unittest.TestCase):
"manifest_text":". 2e9ec317e197819358fbc43afca7d837+8 0:8:count.txt\n"},
{"uuid":"zzzzz-4zz18-mockcollection0"})
with import_manifest('. 781e5e245d69b566979b86e28d23f2c7+10 0:10:count.txt\n',
- api_client=api, keep=keep) as c:
+ api_client=api, keep=keep, sync=SYNC_EXPLICIT) as c:
writer = c.open("count.txt", "w+")
self.assertEqual(writer.size(), 0)
writer.write("01234567")
@@ -297,7 +298,7 @@ class ArvadosFileWriterTestCase(unittest.TestCase):
api = ArvadosFileWriterTestCase.MockApi({"name":"test_create_multiple",
"manifest_text":". 2e9ec317e197819358fbc43afca7d837+8 e8dc4081b13434b45189a720b77b6818+8 0:8:count1.txt 8:8:count2.txt\n"},
{"uuid":"zzzzz-4zz18-mockcollection0"})
- with arvados.Collection(api_client=api, keep_client=keep) as c:
+ with arvados.Collection(api_client=api, keep_client=keep, sync=SYNC_EXPLICIT) as c:
w1 = c.open("count1.txt", "w")
w2 = c.open("count2.txt", "w")
w1.write("01234567")
@@ -314,7 +315,7 @@ class ArvadosFileWriterTestCase(unittest.TestCase):
self.assertEqual("01234567", keep.get("2e9ec317e197819358fbc43afca7d837+8"))
def test_remove(self):
- with import_manifest('. 781e5e245d69b566979b86e28d23f2c7+10 0:10:count1.txt 0:10:count2.txt\n') as c:
+ with import_manifest('. 781e5e245d69b566979b86e28d23f2c7+10 0:10:count1.txt 0:10:count2.txt\n', sync=SYNC_EXPLICIT) as c:
self.assertEqual(". 781e5e245d69b566979b86e28d23f2c7+10 0:10:count1.txt 0:10:count2.txt\n", export_manifest(c))
self.assertTrue("count1.txt" in c)
c.remove("count1.txt")
@@ -322,22 +323,30 @@ class ArvadosFileWriterTestCase(unittest.TestCase):
self.assertEqual(". 781e5e245d69b566979b86e28d23f2c7+10 0:10:count2.txt\n", export_manifest(c))
def test_remove_in_subdir(self):
- with import_manifest('. 781e5e245d69b566979b86e28d23f2c7+10 0:10:count1.txt\n./foo 781e5e245d69b566979b86e28d23f2c7+10 0:10:count2.txt\n') as c:
+ with import_manifest('. 781e5e245d69b566979b86e28d23f2c7+10 0:10:count1.txt\n./foo 781e5e245d69b566979b86e28d23f2c7+10 0:10:count2.txt\n', sync=SYNC_EXPLICIT) as c:
c.remove("foo/count2.txt")
self.assertEqual(". 781e5e245d69b566979b86e28d23f2c7+10 0:10:count1.txt\n", export_manifest(c))
- def test_remove_subdir(self):
- with import_manifest('. 781e5e245d69b566979b86e28d23f2c7+10 0:10:count1.txt\n./foo 781e5e245d69b566979b86e28d23f2c7+10 0:10:count2.txt\n') as c:
+ def test_remove_empty_subdir(self):
+ with import_manifest('. 781e5e245d69b566979b86e28d23f2c7+10 0:10:count1.txt\n./foo 781e5e245d69b566979b86e28d23f2c7+10 0:10:count2.txt\n', sync=SYNC_EXPLICIT) as c:
+ c.remove("foo/count2.txt")
c.remove("foo")
self.assertEqual(". 781e5e245d69b566979b86e28d23f2c7+10 0:10:count1.txt\n", export_manifest(c))
+ def test_remove_empty_subdir(self):
+ with import_manifest('. 781e5e245d69b566979b86e28d23f2c7+10 0:10:count1.txt\n./foo 781e5e245d69b566979b86e28d23f2c7+10 0:10:count2.txt\n', sync=SYNC_EXPLICIT) as c:
+ with self.assertRaises(IOError):
+ c.remove("foo")
+ c.remove("foo", rm_r=True)
+ self.assertEqual(". 781e5e245d69b566979b86e28d23f2c7+10 0:10:count1.txt\n", export_manifest(c))
+
def test_prefetch(self):
keep = ArvadosFileWriterTestCase.MockKeep({"2e9ec317e197819358fbc43afca7d837+8": "01234567", "e8dc4081b13434b45189a720b77b6818+8": "abcdefgh"})
with import_manifest(". 2e9ec317e197819358fbc43afca7d837+8 e8dc4081b13434b45189a720b77b6818+8 0:16:count.txt\n", keep=keep) as c:
r = c.open("count.txt", "r")
self.assertEqual("0123", r.read(4))
- self.assertEqual(["2e9ec317e197819358fbc43afca7d837+8", "2e9ec317e197819358fbc43afca7d837+8", "e8dc4081b13434b45189a720b77b6818+8"], keep.requests)
-
+ self.assertTrue("2e9ec317e197819358fbc43afca7d837+8" in keep.requests)
+ self.assertTrue("e8dc4081b13434b45189a720b77b6818+8" in keep.requests)
class ArvadosFileReaderTestCase(StreamFileReaderTestCase):
class MockParent(object):
@@ -361,6 +370,9 @@ class ArvadosFileReaderTestCase(StreamFileReaderTestCase):
def _my_block_manager(self):
return ArvadosFileReaderTestCase.MockParent.MockBlockMgr(self.blocks, self.nocache)
+ def sync_mode(self):
+ return SYNC_READONLY
+
def make_count_reader(self, nocache=False):
stream = []
n = 0
@@ -402,7 +414,9 @@ class ArvadosFileReadTestCase(unittest.TestCase, StreamRetryTestMixin):
n += k.size
except ValueError:
pass
- af = ArvadosFile(arvados.Collection(keep_client=self.keep_client()),
+ col = arvados.Collection(keep_client=self.keep_client())
+ col._my_block_manager().prefetch_enabled = False
+ af = ArvadosFile(col,
stream=stream,
segments=segments)
return ArvadosFileReader(af, "test", **kwargs)
diff --git a/sdk/python/tests/test_keep_client.py b/sdk/python/tests/test_keep_client.py
index 39ea3c4..885ae7c 100644
--- a/sdk/python/tests/test_keep_client.py
+++ b/sdk/python/tests/test_keep_client.py
@@ -432,11 +432,11 @@ class KeepClientServiceTestCase(unittest.TestCase):
def check_errors_from_last_retry(self, verb, exc_class):
api_client = self.mock_n_keep_disks(2)
- keep_client = arvados.KeepClient(api_client=api_client)
req_mock = getattr(tutil, 'mock_{}_responses'.format(verb))(
"retry error reporting test", 500, 500, 403, 403)
with req_mock, tutil.skip_sleep, \
self.assertRaises(exc_class) as err_check:
+ keep_client = arvados.KeepClient(api_client=api_client)
getattr(keep_client, verb)('d41d8cd98f00b204e9800998ecf8427e+0',
num_retries=3)
self.assertEqual([403, 403], [
@@ -453,9 +453,9 @@ class KeepClientServiceTestCase(unittest.TestCase):
data = 'partial failure test'
data_loc = '{}+{}'.format(hashlib.md5(data).hexdigest(), len(data))
api_client = self.mock_n_keep_disks(3)
- keep_client = arvados.KeepClient(api_client=api_client)
with tutil.mock_put_responses(data_loc, 200, 500, 500) as req_mock, \
self.assertRaises(arvados.errors.KeepWriteError) as exc_check:
+ keep_client = arvados.KeepClient(api_client=api_client)
keep_client.put(data)
self.assertEqual(2, len(exc_check.exception.service_errors()))
commit f56d3a6876f246f78d5bc231a0ac5b6e4c6bdb9c
Merge: 0703204 89fb910
Author: Peter Amstutz <peter.amstutz at curoverse.com>
Date: Tue Feb 3 11:20:31 2015 -0500
Merge branch 'master' into 4823-python-sdk-writable-collection-api
-----------------------------------------------------------------------
hooks/post-receive
--
More information about the arvados-commits
mailing list