X-Git-Url: https://git.arvados.org/arvados.git/blobdiff_plain/cfcfa1c48d95a07bb961baeef8d7658ddecce41e..ef3e45fcc338a85432c207685567385972f79ee6:/sdk/python/arvados/arvfile.py diff --git a/sdk/python/arvados/arvfile.py b/sdk/python/arvados/arvfile.py index 7cd64aa16e..71af6445a5 100644 --- a/sdk/python/arvados/arvfile.py +++ b/sdk/python/arvados/arvfile.py @@ -129,7 +129,7 @@ class ArvadosFileReaderBase(_FileLikeObjectBase): @_FileLikeObjectBase._before_close @retry_method def decompress(self, decompress, size, num_retries=None): - for segment in self.readall(size, num_retries): + for segment in self.readall(size, num_retries=num_retries): data = decompress(segment) if data: yield data @@ -320,7 +320,7 @@ class _BufferBlock(object): @synchronized def set_state(self, nextstate, val=None): if (self._state, nextstate) not in self.STATE_TRANSITIONS: - raise StateChangeError("Invalid state change from %s to %s" % (self.state, nextstate), self.state, nextstate) + raise StateChangeError("Invalid state change from %s to %s" % (self._state, nextstate), self._state, nextstate) self._state = nextstate if self._state == _BufferBlock.PENDING: @@ -491,7 +491,7 @@ class _BlockManager(object): for i in xrange(0, self.num_put_threads): thread = threading.Thread(target=self._commit_bufferblock_worker) self._put_threads.append(thread) - thread.daemon = False + thread.daemon = True thread.start() def _block_prefetch_worker(self): @@ -543,9 +543,6 @@ class _BlockManager(object): def __exit__(self, exc_type, exc_value, traceback): self.stop_threads() - def __del__(self): - self.stop_threads() - def commit_bufferblock(self, block, sync): """Initiate a background upload of a bufferblock. @@ -564,11 +561,17 @@ class _BlockManager(object): # Mark the block as PENDING so to disallow any more appends. block.set_state(_BufferBlock.PENDING) except StateChangeError as e: - if e.state == _BufferBlock.PENDING and sync: - block.wait_for_commit.wait() - if block.state() == _BufferBlock.ERROR: - raise block.error - return + if e.state == _BufferBlock.PENDING: + if sync: + block.wait_for_commit.wait() + else: + return + if block.state() == _BufferBlock.COMMITTED: + return + elif block.state() == _BufferBlock.ERROR: + raise block.error + else: + raise if sync: try: @@ -929,7 +932,7 @@ class ArvadosFile(object): bb = self.parent._my_block_manager().get_bufferblock(s.locator) if bb: if bb.state() != _BufferBlock.COMMITTED: - self.parent._my_block_manager().commit_bufferblock(self._current_bblock, sync=True) + self.parent._my_block_manager().commit_bufferblock(bb, sync=True) to_delete.add(s.locator) s.locator = bb.locator() for s in to_delete: @@ -1070,7 +1073,7 @@ class ArvadosFileWriter(ArvadosFileReader): @retry_method def writelines(self, seq, num_retries=None): for s in seq: - self.write(s, num_retries) + self.write(s, num_retries=num_retries) @_FileLikeObjectBase._before_close def truncate(self, size=None):