Skip to content

P
pyfs
Commit 7de29d32 by rfkelly0
Options

S3FS re-shuffle and performance improvements: 

  * create new "fs.remote" module with generic utils for remote filesystems
  * include ThreadingTestCases in S3FS test suite
  * new listdir() mode that returns full info dict for each entry
parent 033e16c5
Showing with 411 additions and 132 deletions
fs/remote.py 0 → 100644
"""
fs.remote: utilities for interfacing with remote filesystems
"""
import time
import copy
from fs.path import *
from fs.errors import *
from fs.wrapfs import WrapFS
try:
from tempfile import SpooledTemporaryFile as TempFile
except ImportError:
from tempfile import NamedTemporaryFile as TempFile
class RemoteFileBuffer(object):
"""File-like object providing buffer for local file operations.
Instances of this class manage a local tempfile buffer corresponding
to the contents of a remote file. All reads and writes happen locally,
with the content being copied to the remote file only on flush() or
close(). Writes to the remote file are performed using the setcontents()
method on the owning FS object.
The intended use-case is for a remote filesystem (e.g. S3FS) to return
instances of this class from its open() method, and to provide the
file-uploading logic in its setcontents() method, as in the following
pseudo-code:
def open(self,path,mode="r"):
rf = self._get_remote_file(path)
return RemoteFileBuffer(self,path,mode,rf)
def setcontents(self,path,file):
self._put_remote_file(path,file)
The current implementation reads the entire contents of the file into
the buffer before returning. Future implementations may pull data into
the buffer on demand.
"""
def __init__(self,fs,path,mode,rfile=None):
"""RemoteFileBuffer constructor.
The owning filesystem, path and mode must be provided. If the
optional argument 'rfile' is provided, it must be a read()-able
object containing the initial file contents.
"""
self.file = TempFile()
self.fs = fs
self.path = path
self.mode = mode
self.closed = False
if rfile is not None:
data = rfile.read(1024*256)
while data:
self.file.write(data)
data = rfile.read(1024*256)
if "a" not in mode:
self.file.seek(0)
def __del__(self):
if not self.closed:
self.close()
# This is lifted straight from the stdlib's tempfile.py
def __getattr__(self,name):
file = self.__dict__['file']
a = getattr(file, name)
if not issubclass(type(a), type(0)):
setattr(self, name, a)
return a
def __enter__(self):
self.file.__enter__()
return self
def __exit__(self,exc,value,tb):
self.close()
return False
def __iter__(self):
return iter(self.file)
def flush(self):
self.file.flush()
if "w" in self.mode or "a" in self.mode or "+" in self.mode:
pos = self.file.tell()
self.file.seek(0)
self.fs.setcontents(self.path,self.file)
self.file.seek(pos)
def close(self):
self.closed = True
if "w" in self.mode or "a" in self.mode or "+" in self.mode:
self.file.seek(0)
self.fs.setcontents(self.path,self.file)
self.file.close()
def cached(func):
"""Method decorator that caches results for CacheFS."""
@wraps(func)
def wrapper(self,path="",*args,**kwds):
try:
(success,result) = self._cache_get(path,func.__name__,args,kwds)
except KeyError:
try:
res = func(self,path,*args,**kwds)
except Exception, e:
self._cache_set(path,func.__name__,args,kwds,(False,e))
raise
else:
self._cache_set(path,func.__name__,args,kwds,(True,res))
return res
else:
if not success:
raise copy.copy(result)
else:
return copy.copy(result)
return wrapper
class CacheFS(WrapFS):
"""Simple wrapper to cache meta-data of a remote filesystems.
This FS wrapper implements a simplistic cache that can help speedup
access to a remote filesystem. File and directory meta-data is cached
but the actual file contents are not.
"""
def __init__(self,fs,timeout=1):
"""CacheFS constructor.
The optional argument 'timeout' specifies the cache timeout in
seconds. The default timeout is 1 second. To prevent cache
entries from ever timing out, set it to None.
"""
self.timeout = timeout
self._cache = {"":{}}
super(CacheFS,self).__init__(fs)
def _path_cache(self,path):
cache = self._cache
for name in iteratepath(path):
cache = cache.setdefault(name,{"":{}})
return cache
def _cache_get(self,path,func,args,kwds):
now = time.time()
cache = self._path_cache(path)
key = (tuple(args),tuple(sorted(kwds.iteritems())))
(t,v) = cache[""][func][key]
if self.timeout is not None:
if t < now - self.timeout:
raise KeyError
return v
def _cache_set(self,path,func,args,kwds,v):
t = time.time()
cache = self._path_cache(path)
key = (tuple(args),tuple(sorted(kwds.iteritems())))
cache[""].setdefault(func,{})[key] = (t,v)
def _uncache(self,path,added=False,removed=False,unmoved=False):
cache = self._cache
names = list(iteratepath(path))
# If it's not the root dir, also clear some items for ancestors
if names:
# Clear cached 'getinfo' and 'getsize' for all ancestors
for name in names[:-1]:
cache[""].pop("getinfo",None)
cache[""].pop("getsize",None)
cache = cache.get(name,None)
if cache is None:
return
# Adjust cached 'listdir' for parent directory.
# Currently we only do this for argment-less listdir() calls.
# There's probably a better way to manage this...
cache[""].pop("getinfo",None)
cache[""].pop("getsize",None)
if added:
for (key,val) in list(cache[""].get("listdir",{}).items()):
if key == ((),()):
val[1][1].append(basename(path))
else:
cache[""].pop(key,None)
elif removed:
for (key,val) in list(cache[""].get("listdir",{}).items()):
if key == ((),()):
try:
val[1][1].remove(basename(path))
except ValueError:
pass
else:
cache[""].pop(key,None)
elif not unmoved:
cache[""].pop("listdir",None)
# Clear all cached info for the path itself.
cache[names[-1]] = {"":{}}
@cached
def exists(self,path):
return super(CacheFS,self).exists(path)
@cached
def isdir(self,path):
return super(CacheFS,self).isdir(path)
@cached
def isfile(self,path):
return super(CacheFS,self).isfile(path)
@cached
def listdir(self,path="",**kwds):
return super(CacheFS,self).listdir(path,**kwds)
@cached
def getinfo(self,path):
return super(CacheFS,self).getinfo(path)
@cached
def getsize(self,path):
return super(CacheFS,self).getsize(path)
@cached
def getxattr(self,path,name):
return super(CacheFS,self).getxattr(path,name)
@cached
def listxattrs(self,path):
return super(CacheFS,self).listxattrs(path)
def open(self,path,mode="r"):
f = super(CacheFS,self).open(path,mode)
self._uncache(path,unmoved=True)
return f
def setcontents(self,path,contents):
res = super(CacheFS,self).setcontents(path,contents)
self._uncache(path,unmoved=True)
return res
def getcontents(self,path):
res = super(CacheFS,self).getcontents(path)
self._uncache(path,unmoved=True)
return res
def makedir(self,path,**kwds):
super(CacheFS,self).makedir(path,**kwds)
self._uncache(path,added=True)
def remove(self,path):
super(CacheFS,self).remove(path)
self._uncache(path,removed=True)
def removedir(self,path,**kwds):
super(CacheFS,self).removedir(path,**kwds)
self._uncache(path,removed=True)
def rename(self,src,dst):
super(CacheFS,self).rename(src,dst)
self._uncache(src,removed=True)
self._uncache(dst,added=True)
def copy(self,src,dst,**kwds):
super(CacheFS,self).copy(src,dst,**kwds)
self._uncache(dst,added=True)
def copydir(self,src,dst,**kwds):
super(CacheFS,self).copydir(src,dst,**kwds)
self._uncache(dst,added=True)
def move(self,src,dst,**kwds):
super(CacheFS,self).move(src,dst,**kwds)
self._uncache(src,removed=True)
self._uncache(dst,added=True)
def movedir(self,src,dst,**kwds):
super(CacheFS,self).movedir(src,dst,**kwds)
self._uncache(src,removed=True)
self._uncache(dst,added=True)
def setxattr(self,path,name,value):
self._uncache(path,unmoved=True)
return super(CacheFS,self).setxattr(path,name,value)
def delxattr(self,path,name):
self._uncache(path,unmoved=True)
return super(CacheFS,self).delxattr(path,name)
fs/s3fs.py
......@@ -7,17 +7,16 @@ interface for objects stored in Amazon Simple Storage Service (S3).
"""
import boto.s3.connection
from boto.exception import S3ResponseError
import time
import datetime
try:
from tempfile import SpooledTemporaryFile as TempFile
except ImportError:
from tempfile import NamedTemporaryFile as TempFile
import stat as statinfo
import boto.s3.connection
from boto.s3.prefix import Prefix
from boto.exception import S3ResponseError
from fs.base import *
from fs.remote import *
# Boto is not thread-safe, so we need to use a per-thread S3 connection.
......@@ -37,65 +36,6 @@ else:
self._map[(threading.currentThread(),attr)] = value
class RemoteFileBuffer(object):
"""File-like object providing buffer for local file operations.
Instances of this class manage a local tempfile buffer corresponding
to the contents of a remote file. All reads and writes happen locally,
with the content being copied to the remote file only on flush() or
close().
Instances of this class are returned by S3FS.open, but it is desgined
to be usable by any FS subclass that manages remote files.
"""
def __init__(self,fs,path,mode):
self.file = TempFile()
self.fs = fs
self.path = path
self.mode = mode
self.closed = False
def __del__(self):
if not self.closed:
self.close()
# This is lifted straight from the stdlib's tempfile.py
def __getattr__(self,name):
file = self.__dict__['file']
a = getattr(file, name)
if not issubclass(type(a), type(0)):
setattr(self, name, a)
return a
def __enter__(self):
self.file.__enter__()
return self
def __exit__(self,exc,value,tb):
self.close()
return False
def __iter__(self):
return iter(self.file)
def flush(self):
self.file.flush()
if "w" in self.mode or "a" in self.mode or "+" in self.mode:
pos = self.file.tell()
self.file.seek(0)
self.fs.setcontents(self.path,self.file)
self.file.seek(pos)
def close(self):
self.closed = True
if "w" in self.mode or "a" in self.mode or "+" in self.mode:
self.file.seek(0)
self.fs.setcontents(self.path,self.file)
self.file.close()
class S3FS(FS):
"""A filesystem stored in Amazon S3.
......@@ -242,7 +182,6 @@ class S3FS(FS):
so that it can be worked on efficiently. Any changes made to the
file are only sent back to S3 when the file is flushed or closed.
"""
buf = RemoteFileBuffer(self,path,mode)
s3path = self._s3path(path)
# Truncate the file if requested
if "w" in mode:
......@@ -256,13 +195,7 @@ class S3FS(FS):
if not self.isdir(dirname(path)):
raise ParentDirectoryMissingError(path)
k = self._sync_set_contents(s3path,"")
else:
# Get the file contents into the tempfile.
if "r" in mode or "+" in mode or "a" in mode:
k.get_contents_to_file(buf)
if "a" not in mode:
buf.seek(0)
return buf
return RemoteFileBuffer(self,path,mode,k)
def exists(self,path):
"""Check whether a path exists."""
......@@ -309,55 +242,55 @@ class S3FS(FS):
return True
return False
def listdir(self,path="./",wildcard=None,full=False,absolute=False,dirs_only=False,files_only=False):
def listdir(self,path="./",wildcard=None,full=False,absolute=False,info=False,dirs_only=False,files_only=False):
"""List contents of a directory."""
s3path = self._s3path(path) + self._separator
if s3path == "/":
s3path = ""
i = len(s3path)
ks = self._s3bukt.list(prefix=s3path,delimiter=self._separator)
paths = []
keys = []
isDir = False
for k in ks:
for k in self._s3bukt.list(prefix=s3path,delimiter=self._separator):
if not isDir:
isDir = True
nm = k.name[i:]
# Skip over the entry for the directory itself, if it exists
if nm != "":
if type(path) is not unicode:
nm = nm.encode()
paths.append(nm)
if k.name[i:] != "":
k.name = k.name[i:]
keys.append(k)
if not isDir:
if s3path != self._prefix:
if self.isfile(path):
raise ResourceInvalidError(path,msg="that's not a directory: %(path)s")
raise ResourceNotFoundError(path)
return self._listdir_helper(path,paths,wildcard,full,absolute,dirs_only,files_only)
return self._listdir_helper(path,keys,wildcard,full,absolute,info,dirs_only,files_only)
def _listdir_helper(self,path,paths,wildcard,full,absolute,dirs_only,files_only):
def _listdir_helper(self,path,keys,wildcard,full,absolute,info,dirs_only,files_only):
"""Modify listdir helper to avoid additional calls to the server."""
if dirs_only and files_only:
raise ValueError("dirs_only and files_only can not both be True")
dirs = [p[:-1] for p in paths if p.endswith(self._separator)]
files = [p for p in paths if not p.endswith(self._separator)]
if dirs_only:
paths = dirs
keys = [k for k in keys if k.name.endswith(self._separator)]
elif files_only:
paths = files
else:
paths = dirs + files
keys = [k for k in keys if not k.name.endswith(self._separator)]
for k in keys:
if k.name.endswith(self._separator):
k.name = k.name[:-1]
if type(path) is not unicode:
k.name = k.name.encode()
if wildcard is not None:
match = fnmatch.fnmatch
paths = [p for p in paths if match(p, wildcard)]
keys = [k for k in keys if fnmatch.fnmatch(k.name, wildcard)]
if full:
paths = [pathjoin(path, p) for p in paths]
entries = [relpath(pathjoin(path, k.name)) for k in keys]
elif absolute:
paths = [abspath(pathjoin(path, p)) for p in paths]
return paths
entries = [abspath(pathjoin(path, k.name)) for k in keys]
elif info:
entries = [self._get_key_info(k) for k in keys]
else:
entries = [k.name for k in keys]
return entries
def makedir(self,path,recursive=False,allow_recreate=False):
"""Create a directory at the given path.
......@@ -458,19 +391,33 @@ class S3FS(FS):
def getinfo(self,path):
s3path = self._s3path(path)
if path in ("","/"):
return {}
k = Prefix(bucket=self._s3bukt,name="/")
else:
k = self._s3bukt.get_key(s3path)
if k is None:
k = self._s3bukt.get_key(s3path+"/")
if k is None:
raise ResourceNotFoundError(path)
return {}
k = Prefix(bucket=self._s3bukt,name=k.name)
return self._get_key_info(k)
def _get_key_info(self,key):
info = {}
if hasattr(k,"size"):
info['size'] = int(k.size)
info["name"] = basename(key.name)
if isinstance(key,Prefix):
info["st_mode"] = 0700 | statinfo.S_IFDIR
else:
info["st_mode"] = 0700 | statinfo.S_IFREG
if hasattr(key,"size"):
info['size'] = int(key.size)
if hasattr(key,"last_modified"):
# TODO: does S3 use any other formats?
fmt = "%a, %d %b %Y %H:%M:%S %Z"
if hasattr(k,"last_modified"):
info['modified_time'] = datetime.datetime.strptime(k.last_modified,fmt)
try:
mtime = datetime.datetime.strptime(key.last_modified,fmt)
info['modified_time'] = mtime
except ValueError:
pass
return info
def desc(self,path):
......
fs/sftpfs.py
......@@ -119,6 +119,8 @@ class SFTPFS(FS):
@convert_os_errors
def open(self,path,mode="r",bufsize=-1):
npath = self._normpath(path)
# paramiko implements its own buffering and write-back logic,
# so we don't need to use a RemoteFileBuffer here.
f = self.client.open(npath,mode,bufsize)
if self.isdir(path):
msg = "that's a directory: %(path)s"
......
fs/tests/__init__.py
......@@ -462,16 +462,16 @@ class FSTestCases:
class ThreadingTestCases:
"""Testcases for thread-safety of FS implementations."""
_ThreadingTestCasesLock = threading.RLock()
__lock = threading.RLock()
def _yield(self):
time.sleep(0.01)
def _lock(self):
self._ThreadingTestCasesLock.acquire()
self.__lock.acquire()
def _unlock(self):
self._ThreadingTestCasesLock.release()
self.__lock.release()
def _makeThread(self,func,errors):
def runThread():
......@@ -537,7 +537,7 @@ class ThreadingTestCases:
try:
self._runThreads(thread1,thread2,thread3)
except ResourceLockedError:
# that's ok, some platforms don't support concurrent writes
# that's ok, some implementations don't support concurrent writes
pass
def test_FSTestCases_in_separate_dirs(self):
......
fs/tests/test_remote.py 0 → 100644
"""
fs.tests.test_remote: testcases for FS remote support
"""
from fs.tests import FSTestCases, ThreadingTestCases
import unittest
from fs.remote import *
from fs.tempfs import TempFS
from fs.path import *
class TestCacheFS(unittest.TestCase,FSTestCases,ThreadingTestCases):
def setUp(self):
sys.setcheckinterval(1)
self.fs = CacheFS(TempFS())
def tearDown(self):
self.fs.close()
fs/tests/test_s3fs.py
#!/usr/bin/env python
"""
fs.tests.test_s3fs: testcases for the S3FS module
......@@ -11,39 +10,39 @@ attribute the True on te TestS3FS class to get them running.
import unittest
from fs.tests import FSTestCases
from fs.tests import FSTestCases, ThreadingTestCases
from fs.path import *
from fs import s3fs
class TestS3FS(unittest.TestCase,FSTestCases):
class TestS3FS(unittest.TestCase,FSTestCases,ThreadingTestCases):
# Disable the tests by default
__test__ = False
#__test__ = False
bucket = "test-s3fs.rfk.id.au"
def setUp(self):
self.fs = s3fs.S3FS(self.bucket)
self._clear()
def _clear(self):
for (path,files) in self.fs.walk(search="depth"):
for fn in files:
self.fs.remove(pathjoin(path,fn))
if path and path != "/":
self.fs.removedir(path)
def tearDown(self):
self._clear()
for k in self.fs._s3bukt.list():
self.fs._s3bukt.delete_key(k)
self.fs._s3conn.delete_bucket(self.bucket)
def test_concurrent_copydir(self):
# makdir() on S3FS is currently not atomic
pass
def test_makedir_winner(self):
# makdir() on S3FS is currently not atomic
pass
def test_multiple_overwrite(self):
# S3's eventual-consistency seems to be breaking this test
pass
class TestS3FS_prefix(TestS3FS):
def setUp(self):
self.fs = s3fs.S3FS(self.bucket,"/unittest/files")
self._clear()
for k in self.fs._s3bukt.list():
self.fs._s3bukt.delete_key(k)
fs/wrapfs.py
......@@ -13,10 +13,11 @@ directory listings.
"""
from fnmatch import fnmatch
from fs.base import FS, threading, synchronize
from fs.errors import *
def rewrite_errors(func):
@wraps(func)
def wrapper(self,*args,**kwds):
......@@ -136,11 +137,21 @@ class WrapFS(FS):
return self.wrapped_fs.isfile(self._encode(path))
@rewrite_errors
def listdir(self,path="",wildcard=None,full=False,absolute=False,dirs_only=False,files_only=False):
def listdir(self,path="",**kwds):
wildcard = kwds.pop("wildcard","*")
info = kwds.get("info",False)
entries = []
for name in self.wrapped_fs.listdir(self._encode(path),wildcard=None,full=full,absolute=absolute,dirs_only=dirs_only,files_only=files_only):
entries.append(self._decode(name))
return self._listdir_helper(path,entries,wildcard=wildcard,full=False,absolute=False,dirs_only=False,files_only=False)
for e in self.wrapped_fs.listdir(self._encode(path),**kwds):
if info:
e["name"] = self._decode(e["name"])
if wildcard is not None and not fnmatch(e["name"],wildcard):
continue
else:
e = self._decode(e)
if wildcard is not None and not fnmatch(e,wildcard):
continue
entries.append(e)
return entries
@rewrite_errors
def makedir(self,path,*args,**kwds):
......@@ -226,13 +237,13 @@ class HideDotFiles(WrapFS):
path = pathjoin(current_path, filename)
if self.isdir(path):
if dir_wildcard is not None:
if fnmatch.fnmatch(path, dir_wilcard):
if fnmatch(path, dir_wilcard):
dirs.append(path)
else:
dirs.append(path)
else:
if wildcard is not None:
if fnmatch.fnmatch(path, wildcard):
if fnmatch(path, wildcard):
paths.append(filename)
else:
paths.append(filename)
......
Markdown is supported
0% or
You are about to add 0 people to the discussion. Proceed with caution.
Finish editing this message first!
Please register or to comment