| | from collections import deque |
| |
|
| |
|
| | class Transaction: |
| | """Filesystem transaction write context |
| | |
| | Gathers files for deferred commit or discard, so that several write |
| | operations can be finalized semi-atomically. This works by having this |
| | instance as the ``.transaction`` attribute of the given filesystem |
| | """ |
| |
|
| | def __init__(self, fs, **kwargs): |
| | """ |
| | Parameters |
| | ---------- |
| | fs: FileSystem instance |
| | """ |
| | self.fs = fs |
| | self.files = deque() |
| |
|
| | def __enter__(self): |
| | self.start() |
| | return self |
| |
|
| | def __exit__(self, exc_type, exc_val, exc_tb): |
| | """End transaction and commit, if exit is not due to exception""" |
| | |
| | self.complete(commit=exc_type is None) |
| | if self.fs: |
| | self.fs._intrans = False |
| | self.fs._transaction = None |
| | self.fs = None |
| |
|
| | def start(self): |
| | """Start a transaction on this FileSystem""" |
| | self.files = deque() |
| | self.fs._intrans = True |
| |
|
| | def complete(self, commit=True): |
| | """Finish transaction: commit or discard all deferred files""" |
| | while self.files: |
| | f = self.files.popleft() |
| | if commit: |
| | f.commit() |
| | else: |
| | f.discard() |
| | self.fs._intrans = False |
| | self.fs._transaction = None |
| | self.fs = None |
| |
|
| |
|
| | class FileActor: |
| | def __init__(self): |
| | self.files = [] |
| |
|
| | def commit(self): |
| | for f in self.files: |
| | f.commit() |
| | self.files.clear() |
| |
|
| | def discard(self): |
| | for f in self.files: |
| | f.discard() |
| | self.files.clear() |
| |
|
| | def append(self, f): |
| | self.files.append(f) |
| |
|
| |
|
| | class DaskTransaction(Transaction): |
| | def __init__(self, fs): |
| | """ |
| | Parameters |
| | ---------- |
| | fs: FileSystem instance |
| | """ |
| | import distributed |
| |
|
| | super().__init__(fs) |
| | client = distributed.default_client() |
| | self.files = client.submit(FileActor, actor=True).result() |
| |
|
| | def complete(self, commit=True): |
| | """Finish transaction: commit or discard all deferred files""" |
| | if commit: |
| | self.files.commit().result() |
| | else: |
| | self.files.discard().result() |
| | self.fs._intrans = False |
| | self.fs = None |
| |
|
