Add file based locks

Author: mrocklin

continuous_integration/Dockerfile

@@ -9,10 +9,10 @@ ENV PATH /opt/conda/bin:$PATH
 
 # hdfs3 - python
 ENV LIBHDFS3_CONF /etc/hadoop/conf/hdfs-site.xml
-RUN conda install -y -q ipython pytest
+RUN conda install -y -q ipython pytest locket
 RUN conda install -y -q libhdfs3 -c conda-forge
 RUN conda create -y -n py3 python=3
-RUN conda install -y -n py3 ipython pytest
+RUN conda install -y -n py3 ipython pytest locket
 RUN conda install -y -n py3 libhdfs3 -c conda-forge
 
 # Cloudera repositories

docs/source/limitations.rst

@@ -6,24 +6,6 @@ Forked processes
 
 The ``libhdfs3`` library may fail when an ``HDFileSystem`` is copied to a new
 forked process.  This happens in some cases when using ``hdfs3`` with
-``multiprocessing`` in Python 2.  Common solutions include the following:
+``multiprocessing`` in Python 2.
 
-*  Use threads
-*  Use Python 3 and a multiprocessing context using spawn with
-   ``multiprocessing.get_context(method='spawn')`` see `multiprocessing docs`_
-*  Only instantiate ``HDFileSystem`` within the forked processes, do not start
-   an ``HDFileSystem`` within the parent processes or do not use that
-   ``HDFileSystem`` within the child processes.
-*  Use a file based lock.  We recommend ``locket``::
-
-        $ pip install locket
-
-    .. code-block:: python
-
-        import locket
-
-        with locket.lock_file('.lock'):
-            # do hdfs3 work
-
-
-.. _`multiprocessing docs`: https://docs.python.org/3/library/multiprocessing.html#contexts-and-start-methods
+We get around this by using file-based locks, which slightly limit concurrency.

hdfs3/core.py

@@ -16,6 +16,9 @@
 from .compatibility import FileNotFoundError, urlparse, ConnectionError
 from .utils import read_block
 
+import locket
+
+lock = locket.lock_file('.libhdfs3.lock')
 
 logger = logging.getLogger(__name__)
 
@@ -132,9 +135,6 @@ class HDFileSystem(object):
 
     >>> hdfs = HDFileSystem(host='127.0.0.1', port=8020)  # doctest: +SKIP
     """
-
-    CONNECT_RETRIES = 5
-
     def __init__(self, host=DEFAULT_HOST, port=DEFAULT_PORT, user=None,
                  ticket_cache=None, token=None, pars=None, connect=True):
         """
@@ -185,44 +185,41 @@ def connect(self):
         if self._handle:
             return
 
-        o = _lib.hdfsNewBuilder()
-        if self.port is not None:
-            _lib.hdfsBuilderSetNameNodePort(o, self.port)
-        _lib.hdfsBuilderSetNameNode(o, ensure_bytes(self.host))
-        if self.user:
-            _lib.hdfsBuilderSetUserName(o, ensure_bytes(self.user))
+        with lock:
+            o = _lib.hdfsNewBuilder()
+            if self.port is not None:
+                _lib.hdfsBuilderSetNameNodePort(o, self.port)
+            _lib.hdfsBuilderSetNameNode(o, ensure_bytes(self.host))
+            if self.user:
+                _lib.hdfsBuilderSetUserName(o, ensure_bytes(self.user))
 
-        if self.ticket_cache:
-            _lib.hdfsBuilderSetKerbTicketCachePath(o, ensure_bytes(self.ticket_cache))
+            if self.ticket_cache:
+                _lib.hdfsBuilderSetKerbTicketCachePath(o, ensure_bytes(self.ticket_cache))
 
-        if self.token:
-            _lib.hdfsBuilderSetToken(o, ensure_bytes(self.token))
+            if self.token:
+                _lib.hdfsBuilderSetToken(o, ensure_bytes(self.token))
 
-        for par, val in self.pars.items():
-            if not  _lib.hdfsBuilderConfSetStr(o, ensure_bytes(par), ensure_bytes(val)) == 0:
-                warnings.warn('Setting conf parameter %s failed' % par)
+            for par, val in self.pars.items():
+                if not  _lib.hdfsBuilderConfSetStr(o, ensure_bytes(par), ensure_bytes(val)) == 0:
+                    warnings.warn('Setting conf parameter %s failed' % par)
 
-        trial = 0
-        while trial < self.CONNECT_RETRIES:
             fs = _lib.hdfsBuilderConnect(o)
-            trial += 1
             if fs:
-                break
-        if fs:
-            logger.debug("Connect to handle %d", fs.contents.filesystem)
-            self._handle = fs
-            #if self.token:   # TODO: find out what a delegation token is
-            #    self._token = _lib.hdfsGetDelegationToken(self._handle,
-            #                                             ensure_bytes(self.user))
-        else:
-            msg = ensure_string(_lib.hdfsGetLastError())
-            raise ConnectionError('Connection Failed: {}'.format(msg))
+                logger.debug("Connect to handle %d", fs.contents.filesystem)
+                self._handle = fs
+                #if self.token:   # TODO: find out what a delegation token is
+                #    self._token = _lib.hdfsGetDelegationToken(self._handle,
+                #                                             ensure_bytes(self.user))
+            else:
+                msg = ensure_string(_lib.hdfsGetLastError())
+                raise ConnectionError('Connection Failed: {}'.format(msg))
 
     def disconnect(self):
         """ Disconnect from name node """
         if self._handle:
             logger.debug("Disconnect from handle %d", self._handle.contents.filesystem)
-            _lib.hdfsDisconnect(self._handle)
+            with lock:
+                _lib.hdfsDisconnect(self._handle)
         self._handle = None
 
     def open(self, path, mode='rb', replication=0, buff=0, block_size=0):
@@ -276,8 +273,9 @@ def du(self, path, total=False, deep=False):
 
     def df(self):
         """ Used/free disc space on the HDFS system """
-        cap = _lib.hdfsGetCapacity(self._handle)
-        used = _lib.hdfsGetUsed(self._handle)
+        with lock:
+            cap = _lib.hdfsGetCapacity(self._handle)
+            used = _lib.hdfsGetUsed(self._handle)
         return {'capacity': cap, 'used': used, 'percent-free': 100*(cap-used)/cap}
 
     def get_block_locations(self, path, start=0, length=0):
@@ -287,26 +285,29 @@ def get_block_locations(self, path, start=0, length=0):
         start = int(start) or 0
         length = int(length) or self.info(path)['size']
         nblocks = ctypes.c_int(0)
-        out = _lib.hdfsGetFileBlockLocations(self._handle, ensure_bytes(path),
-                                ctypes.c_int64(start), ctypes.c_int64(length),
-                                ctypes.byref(nblocks))
+        with lock:
+            out = _lib.hdfsGetFileBlockLocations(self._handle, ensure_bytes(path),
+                                    ctypes.c_int64(start), ctypes.c_int64(length),
+                                    ctypes.byref(nblocks))
         locs = []
         for i in range(nblocks.value):
             block = out[i]
             hosts = [block.hosts[i] for i in
                      range(block.numOfNodes)]
             locs.append({'hosts': hosts, 'length': block.length,
                          'offset': block.offset})
-        _lib.hdfsFreeFileBlockLocations(out, nblocks)
+        with lock:
+            _lib.hdfsFreeFileBlockLocations(out, nblocks)
         return locs
 
     def info(self, path):
         """ File information (as a dict) """
         if not self.exists(path):
             raise FileNotFoundError(path)
-        fi = _lib.hdfsGetPathInfo(self._handle, ensure_bytes(path)).contents
-        out = info_to_dict(fi)
-        _lib.hdfsFreeFileInfo(ctypes.byref(fi), 1)
+        with lock:
+            fi = _lib.hdfsGetPathInfo(self._handle, ensure_bytes(path)).contents
+            out = info_to_dict(fi)
+            _lib.hdfsFreeFileInfo(ctypes.byref(fi), 1)
         return ensure_string(out)
 
     def walk(self, path):
@@ -358,9 +359,10 @@ def ls(self, path, detail=True):
         if not self.exists(path):
             raise FileNotFoundError(path)
         num = ctypes.c_int(0)
-        fi = _lib.hdfsListDirectory(self._handle, ensure_bytes(path), ctypes.byref(num))
-        out = [ensure_string(info_to_dict(fi[i])) for i in range(num.value)]
-        _lib.hdfsFreeFileInfo(fi, num.value)
+        with lock:
+            fi = _lib.hdfsListDirectory(self._handle, ensure_bytes(path), ctypes.byref(num))
+            out = [ensure_string(info_to_dict(fi[i])) for i in range(num.value)]
+            _lib.hdfsFreeFileInfo(fi, num.value)
         if detail:
             return out
         else:
@@ -376,10 +378,11 @@ def __del__(self):
 
     def mkdir(self, path):
         """ Make directory at path """
-        out = _lib.hdfsCreateDirectory(self._handle, ensure_bytes(path))
-        if out != 0:
-            msg = ensure_string(_lib.hdfsGetLastError())
-            raise IOError('Create directory failed: {}'.format(msg))
+        with lock:
+            out = _lib.hdfsCreateDirectory(self._handle, ensure_bytes(path))
+            if out != 0:
+                msg = ensure_string(_lib.hdfsGetLastError())
+                raise IOError('Create directory failed: {}'.format(msg))
 
     def set_replication(self, path, replication):
         """ Instruct HDFS to set the replication for the given file.
@@ -391,31 +394,35 @@ def set_replication(self, path, replication):
         """
         if replication < 0:
             raise ValueError('Replication must be positive, or 0 for system default')
-        out = _lib.hdfsSetReplication(self._handle, ensure_bytes(path),
-                                     ctypes.c_int16(int(replication)))
-        if out != 0:
-            msg = ensure_string(_lib.hdfsGetLastError())
-            raise IOError('Set replication failed: {}'.format(msg))
+        with lock:
+            out = _lib.hdfsSetReplication(self._handle, ensure_bytes(path),
+                                         ctypes.c_int16(int(replication)))
+            if out != 0:
+                msg = ensure_string(_lib.hdfsGetLastError())
+                raise IOError('Set replication failed: {}'.format(msg))
 
     def mv(self, path1, path2):
         """ Move file at path1 to path2 """
         if not self.exists(path1):
             raise FileNotFoundError(path1)
-        out = _lib.hdfsRename(self._handle, ensure_bytes(path1), ensure_bytes(path2))
+        with lock:
+            out = _lib.hdfsRename(self._handle, ensure_bytes(path1), ensure_bytes(path2))
         return out == 0
 
     def rm(self, path, recursive=True):
         "Use recursive for `rm -r`, i.e., delete directory and contents"
         if not self.exists(path):
             raise FileNotFoundError(path)
-        out = _lib.hdfsDelete(self._handle, ensure_bytes(path), bool(recursive))
-        if out != 0:
-            msg = ensure_string(_lib.hdfsGetLastError())
-            raise IOError('Remove failed on %s %s' % (path, msg))
+        with lock:
+            out = _lib.hdfsDelete(self._handle, ensure_bytes(path), bool(recursive))
+            if out != 0:
+                msg = ensure_string(_lib.hdfsGetLastError())
+                raise IOError('Remove failed on %s %s' % (path, msg))
 
     def exists(self, path):
         """ Is there an entry at path? """
-        out = _lib.hdfsExists(self._handle, ensure_bytes(path) )
+        with lock:
+            out = _lib.hdfsExists(self._handle, ensure_bytes(path) )
         return out == 0
 
     def chmod(self, path, mode):
@@ -441,20 +448,22 @@ def chmod(self, path, mode):
         """
         if not self.exists(path):
             raise FileNotFoundError(path)
-        out = _lib.hdfsChmod(self._handle, ensure_bytes(path), ctypes.c_short(mode))
-        if out != 0:
-            msg = ensure_string(_lib.hdfsGetLastError())
-            raise IOError("chmod failed on %s %s" % (path, msg))
+        with lock:
+            out = _lib.hdfsChmod(self._handle, ensure_bytes(path), ctypes.c_short(mode))
+            if out != 0:
+                msg = ensure_string(_lib.hdfsGetLastError())
+                raise IOError("chmod failed on %s %s" % (path, msg))
 
     def chown(self, path, owner, group):
         """ Change owner/group """
         if not self.exists(path):
             raise FileNotFoundError(path)
-        out = _lib.hdfsChown(self._handle, ensure_bytes(path), ensure_bytes(owner),
-                            ensure_bytes(group))
-        if out != 0:
-            msg = ensure_string(_lib.hdfsGetLastError())
-            raise IOError("chown failed on %s %s" % (path, msg))
+        with lock:
+            out = _lib.hdfsChown(self._handle, ensure_bytes(path), ensure_bytes(owner),
+                                ensure_bytes(group))
+            if out != 0:
+                msg = ensure_string(_lib.hdfsGetLastError())
+                raise IOError("chown failed on %s %s" % (path, msg))
 
     def cat(self, path):
         """ Return contents of file """

hdfs3/tests/test_hdfs3.py

@@ -66,6 +66,7 @@ def test_token_and_ticket_cache_in_same_time():
     assert msg in str(ctx.value)
 
 
+@pytest.mark.slow
 def test_connection_error():
     with pytest.raises(ConnectionError) as ctx:
         hdfs = HDFileSystem(host='localhost', port=9999, connect=False)