Merge branch 'release/v0.15'

Author: cmarat

docs/conf.py

@@ -55,9 +55,9 @@
 # built documents.
 #
 # The short X.Y version.
-version = '0.14'
+version = '0.15'
 # The full version, including alpha/beta/rc tags.
-release = '0.14'
+release = '0.15'
 
 # The language for content autogenerated by Sphinx. Refer to documentation
 # for a list of supported languages.

ers/api.py

@@ -1,17 +1,19 @@
 """
 ers.api
 
-Provides class ERS implementing API to Entity Rgistry System.
+Provides class ERS implementing API to Entity Registry System.
 
 """
 import re
 import sys
-import uuid
 
 from hashlib import md5
 from socket import gethostname
+from collections import Counter
+from random import randrange
 
 import store
+from timeout import TimeoutError
 
 class ERSReadOnly(object):
     """ ERS version with read-only methods.
@@ -26,18 +28,30 @@ def __init__(self,
                  local_only=False):
         self._local_only = local_only
         self.fixed_peers = [] if self._local_only else list(fixed_peers)
+        self._timeout_count = Counter()
         self.store = store.LocalStore()
         self._init_host_urn()
 
     def _init_host_urn(self):
-        fingerprint = md5(gethostname()).hexdigest()
-        self.host_urn = "urn:ers:host:{}".format(fingerprint)
+        # Use uuid provided by CouchDB 1.3+, fallback to hostname fingerprint
+        try:
+            uid = self.store.info()['uuid']
+        except KeyError:
+            uid = md5(gethostname()).hexdigest()
+        self.host_urn = "urn:ers:host:{}".format(uid)
 
     def get_machine_uuid(self):
         '''
         @return a unique identifier for this ERS node
         '''
-        return str(uuid.getnode())
+        return self.host_urn.split(':')[-1]
+
+    def _is_failing(self, url):
+        """
+        Returns True for url's which failed to respond with increasing probability.
+        Returns False for url's which did not fail.
+        """
+        return randrange(self._timeout_count[url] + 1) != 0
 
     def get_entity(self, entity_name):
         '''
@@ -54,15 +68,23 @@ def get_entity(self, entity_name):
                 entity.add_document(doc, source)
 
         # Get documents out of public/cache of connected peers
+        # TODO parallelize 
         for peer in self.get_peers():
+            url = peer['server_url']
+            if self._is_failing(url):
+                continue
+
             remote_docs = []
             try:
-                remote_store = store.RemoteStore(peer['server_url'])
-                remote_docs.extend(remote_store.docs_by_entity(entity_name))
-            except:
-                sys.stderr.write("Warning: failed to query remote peer {0}\n".format(peer))
-                continue
+                remote_docs = store.query_remote(url, 'docs_by_entity', entity_name)
+            except TimeoutError:
+                self._timeout_count[url] += 1
+                sys.stderr.write("Incremented timeout count for {0}: {1}\n".format(
+                    url, self._timeout_count[url]))
+            except Exception as e:
+                sys.stderr.write("Warning: failed to query remote peer {0}. Error: {1}\n".format(peer, e))
             else:
+                self._timeout_count.pop(url, 0)
                 for doc in remote_docs:
                     entity.add_document(doc, 'remote')
         return entity
@@ -81,13 +103,21 @@ def search(self, prop, value=None):
 
         # Search peers
         for peer in self.get_peers():
+            url = peer['server_url']
+            if self._is_failing(url):
+                continue
+
             remote_result = []
             try:
-                remote_store = store.RemoteStore(peer['server_url'])
-                remote_result = set(remote_store.by_property_value(prop, value))
-            except:
-                sys.stderr.write("Warning: failed to query remote peer {0}\n".format(peer))
+                remote_result = store.query_remote(url, 'by_property_value', prop, value)
+            except TimeoutError:
+                self._timeout_count[url] += 1
+                sys.stderr.write("Incremented timeout count for {0}: {1}\n".format(
+                    url, self._timeout_count[url]))
+            except Exception as e:
+                sys.stderr.write("Warning: failed to query remote peer {0}. Error: {1}\n".format(peer, e))
             else:
+                self._timeout_count.pop(url, 0)
                 result.update(remote_result)
 
         return list(result)

ers/store.py

@@ -27,8 +27,9 @@
 
 from functools import partial
 from itertools import chain
+from timeout import timeout
 
-REMOTE_SERVER_TIMEOUT = 300
+REMOTE_SERVER_TIMEOUT = 0.3
 
 DEFAULT_STORE_URI = 'http://127.0.0.1:5984'
 DEFAULT_AUTH = ['admin', 'admin']
@@ -218,6 +219,12 @@ def update_replicator_docs(self, repl_docs):
 RemoteStore = partial(Store, databases=REMOTE_DBS, timeout=REMOTE_SERVER_TIMEOUT)                                                        
 
 
+@timeout(REMOTE_SERVER_TIMEOUT)
+def query_remote(uri, method_name, *args, **kwargs):
+    # import ipdb; ipdb.set_trace()
+    remote_store = RemoteStore(uri)
+    return list(getattr(remote_store, method_name)(*args, **kwargs))
+
 def reset_local_store(auth=DEFAULT_AUTH):
     user, password = auth
     store = LocalStore(filters=[restkit.BasicAuth(user, password)])

ers/timeout.py

@@ -0,0 +1,85 @@
+# Source: http://code.activestate.com/recipes/577853-timeout-decorator-with-multiprocessing/
+
+"""
+Code to timeout with processes.
+
+>>> @timeout(.5)
+... def sleep(x):
+...     print "ABOUT TO SLEEP {0} SECONDS".format(x)
+...     time.sleep(x)
+...     return x
+
+>>> sleep(1)
+Traceback (most recent call last):
+   ...
+TimeoutError: timed out after 0 seconds
+
+>>> sleep(.2)
+0.2
+
+>>> @timeout(.5)
+... def exc():
+...     raise Exception('Houston we have problems!')
+
+>>> exc()
+Traceback (most recent call last):
+   ...
+Exception: Houston we have problems!
+
+"""
+import multiprocessing
+import time
+import logging
+logger = multiprocessing.log_to_stderr()
+logger.setLevel(logging.ERROR)
+
+
+class TimeoutError(Exception):
+    pass
+
+
+class RunableProcessing(multiprocessing.Process):
+    def __init__(self, func, *args, **kwargs):
+        self.queue = multiprocessing.Queue(maxsize=1)
+        args = (func,) + args
+        multiprocessing.Process.__init__(self, target=self.run_func, args=args, kwargs=kwargs)
+
+    def run_func(self, func, *args, **kwargs):
+        try:
+            result = func(*args, **kwargs)
+            self.queue.put((True, result))
+        except Exception as e:
+            self.queue.put((False, e))
+
+    def done(self):
+        return self.queue.full()
+
+    def result(self):
+        return self.queue.get()
+
+
+def timeout(seconds, force_kill=True):
+    def wrapper(function):
+        def inner(*args, **kwargs):
+            now = time.time()
+            proc = RunableProcessing(function, *args, **kwargs)
+            proc.start()
+            proc.join(seconds)
+            if proc.is_alive():
+                if force_kill:
+                    proc.terminate()
+                runtime = int(time.time() - now)
+                raise TimeoutError('timed out after {0} seconds'.format(runtime))
+            assert proc.done()
+            success, result = proc.result()
+            if success:
+                return result
+            else:
+                raise result
+        return inner
+    return wrapper
+
+
+if __name__ == '__main__':
+    import doctest
+    doctest.testmod()

ers/utils.py

@@ -1,9 +1,7 @@
-
 from collections import defaultdict
 from StringIO import StringIO
-import rdflib
 
-# Document model is used to store data in CouchDB. The API is independent from the choice of model.
+import rdflib
 
 def import_nt(registry, file_name, target_graph):
     """ Import N-Triples file.

setup.py

@@ -4,7 +4,7 @@
 
 requirements = open('requirements.txt').read().splitlines()
 setup(name='ERS',
-      version='0.14',
+      version='0.15',
       description='Entity Registry System',
       url='https://github.com/ers-devs/ers-node/',
       packages=['ers'],