multi_query error temporarily closes down worker

multi_query raises errors in case of severe problems only. In those
cases cases continued querying causes more problems. With this commit
an error that is raised out of multi_query will cause the worker to
temporarily be closed. Further calls to multi_query will be logged as
warning and raise a MultiQueryException. This allows the scoop main
process to early terminate and handle the exception.

Waiting times are now configurable and in general increased to 30s as
they are more direct now. This for example allows the endpoint to
restart in case of buuuuugs.

Author: joernhees

config/defaults.py

@@ -19,6 +19,7 @@
 PAUSE_FILE = 'pause.lck'
 
 ERROR_RETRIES = 5  # in case an unexpected error occurs retry? (real world!)
+ERROR_WAIT = 30  # seconds to wait in case of error before retry
 
 NRUNS = 64  # number of whole coverage runs of the evolutionary algorithm
 NRUNS_NO_IMPROVEMENT = 5  # stop if no more coverage patterns found in n runs

gp_learner.py

@@ -1447,7 +1447,7 @@ def find_graph_pattern_coverage(
             raise
         except Exception as e:
             error_count += 1
-            logger.error('uncaught exception in run %d' % run)
+            logger.error('uncaught exception in run %d', run)
             log_wrapped_exception(logger, e)
             if error_count > error_retries:
                 logger.error(
@@ -1456,9 +1456,13 @@ def find_graph_pattern_coverage(
                 )
                 raise
             else:
-                logger.error('will retry in 15s despite error...')
+                logger.error(
+                    'this was uncaught exception number %d, will retry in %ds '
+                    'despite error...',
+                    error_count, config.ERROR_WAIT
+                )
                 logging_config.save_error_logs()
-                sleep(15)
+                sleep(config.ERROR_WAIT)
 
     # sort patterns by fitness, run and then pattern
     patterns = sorted(

gp_query.py

@@ -7,9 +7,13 @@
 from collections import defaultdict
 from collections import Counter
 from collections import Sequence
+from datetime import datetime
+from datetime import timedelta
+from functools import wraps
 import logging
 import re
 import socket
+import sys
 from time import sleep
 
 from cachetools import LRUCache
@@ -18,6 +22,7 @@
 from rdflib.term import Identifier
 import SPARQLWrapper
 from SPARQLWrapper.SPARQLExceptions import EndPointNotFound
+from SPARQLWrapper.SPARQLExceptions import QueryBadFormed
 from SPARQLWrapper.SPARQLExceptions import SPARQLWrapperException
 from xml.sax.expatreader import SAXParseException
 # noinspection PyUnresolvedReferences
@@ -40,13 +45,15 @@
 logger = logging.getLogger(__name__)
 
 
+class QueryException(Exception):
+    pass
 
 
-class EvalException(Exception):
+class IncompleteQueryException(Exception):
     pass
 
 
-class QueryException(EvalException):
+class MultiQueryException(Exception):
     pass
 
 
@@ -221,7 +228,7 @@ def _get_vars_values_mapping(graph_pattern, source_target_pairs):
             _values = [(t,) for t in sorted(set(targets))]
             _val_idx = 1
         else:
-            raise QueryException(
+            raise IncompleteQueryException(
                 "tried to run a query on a graph pattern without "
                 "%s and %s vars:\n%s" % (SOURCE_VAR, TARGET_VAR, graph_pattern)
             )
@@ -276,7 +283,42 @@ def _ask_chunk_result_extractor(q_res, _vars, _ret_val_mapping):
     return chunk_res
 
 
-# noinspection PyBroadException
+def _exception_closes_worker_guard(func):
+    """Temporarily closes _multi_query for current worker.
+
+    This is a workaround for SCOOP's map otherwise having already dispatched
+    further work to this worker, despite an exception of a previous _multi_query
+    not being handled in origin yet.
+
+    An exception being raised out of _multi_query would normally cause origin to
+    back-off for config.ERROR_WAIT and retry. This "quick fails" all remaining
+    work in the time frame.
+    """
+    closed = []
+    wait = timedelta(
+        seconds=config.ERROR_WAIT * .75  # rather don't close too long
+    )
+
+    @wraps(func)
+    def _multi_query_wrapper(*args, **kwds):
+        if closed:
+            if datetime.utcnow() - closed[0] < wait:
+                logger.warning(
+                    '_multi_query temporarily closed for worker due to '
+                    'previous exception'
+                )
+                raise MultiQueryException('closed for worker')
+            else:
+                closed.pop()
+        try:
+            return func(*args, **kwds)
+        except:
+            closed.append(datetime.utcnow())
+            raise
+    return _multi_query_wrapper
+
+
+@_exception_closes_worker_guard
 def _multi_query(
         sparql, timeout, graph_pattern, source_target_pairs,
         batch_size,
@@ -303,15 +345,16 @@ def _multi_query(
                 t, q_res = _query(sparql, timeout, q, **kwds)
                 chunk_res = _chunk_res(
                     q_res, _vars, _ret_val_mapping, **kwds)
-            except EndPointNotFound:
+            except EndPointNotFound as e:
                 # happens if the endpoint reports a 404...
                 # as virtuoso in rare cases seems to report a 404 let's
                 # retry after some time but then cancel
                 if retry:
                     logger.info(
-                        'SPARQL endpoint reports a 404, will retry once in 10s'
+                        'SPARQL endpoint reports a 404, will retry in %ds',
+                        config.ERROR_WAIT
                     )
-                    sleep(10)
+                    sleep(config.ERROR_WAIT)
                     continue
                 else:
                     logger.exception(
@@ -320,7 +363,7 @@ def _multi_query(
                         'could not perform query:\n%s for %s\nException:',
                         q, val_chunk,
                     )
-                    raise
+                    six.reraise(MultiQueryException, e, sys.exc_info()[2])
             except (SPARQLWrapperException, SAXParseException, URLError) as e:
                 if (isinstance(e, SPARQLWrapperException) and
                         re.search(
@@ -346,45 +389,47 @@ def _multi_query(
                     # error. It is very likely that the endpoint is dead...
                     if retry:
                         logger.warning(
-                            'could not perform query, retry in 10s:\n'
+                            'URLError, seems we cannot reach SPARQL endpoint, '
+                            'retry in %ds. Tried to perform query:\n'
                             '%s for %s\nException:',
-                            q, val_chunk,
+                            config.ERROR_WAIT, q, val_chunk,
                             exc_info=1,  # appends exception to message
                         )
-                        sleep(10)
+                        sleep(config.ERROR_WAIT)
                         continue
                     else:
                         logger.exception(
-                            'could not perform query:\n%s for %s\nException:',
+                            'URLError, seems we cannot reach SPARQL endpoint, '
+                            'giving up after 3 retries. Tried to perform query:'
+                            '\n%s for %s\nException:',
                             q, val_chunk,
-                            exc_info=1,  # appends exception to message
                         )
-                        raise
+                        six.reraise(MultiQueryException, e, sys.exc_info()[2])
                 else:
                     logger.warning(
-                        'could not perform query:\n%s for %s\nException:',
+                        'could not perform query, replacing with 0 result:\n'
+                        '%s for %s\nException:',
                         q, val_chunk,
                         exc_info=1,  # appends exception to message
                     )
                     t, chunk_res = timer() - _start_time, {}
-            except Exception:
+            except Exception as e:
                 if retry:
                     logger.warning(
-                        'unhandled exception, retry in 10s:\n'
+                        'unhandled exception, retry in %ds:\n'
                         'Query:\n%s\nChunk:%r\nException:',
-                        q, val_chunk,
+                        config.ERROR_WAIT, q, val_chunk,
                         exc_info=1,  # appends exception to message
                     )
-                    sleep(10)
+                    sleep(config.ERROR_WAIT)
                     continue
                 else:
                     logger.exception(
-                        'unhandled exception:\n'
+                        'unhandled exception, giving up after 3 retries:\n'
                         'Query:\n%s\nChunk:%r\nException:',
                         q, val_chunk,
-                        exc_info=1,  # appends exception to message
                     )
-                    t, chunk_res = timer() - _start_time, {}
+                    six.reraise(MultiQueryException, e, sys.exc_info()[2])
             break
         _res_update(res, chunk_res, **kwds)
         total_time += t

utils.py

@@ -169,17 +169,18 @@ def exception_stack_wrapper(*args, **kwds):
                         str(e)
                     except UnicodeEncodeError:
                         scoop.logger.warning(
-                            're-packing exception for scoop, see'
+                            're-packing exception for scoop, see '
                             'https://github.com/soravux/scoop/pull/24'
                         )
-                        e_msg = repr(e.message)
+                        e_msg = repr(e)
                         six.reraise(type(e), e_msg, exc_info[2])
                     else:
                         raise
                 except BaseException as err:
                     # append the stack as field to the re-raised exception
                     err._exc_fmt = 'error in worker:\n%s' % (
                         ''.join(traceback.format_exception(*exc_info)))
+                    six.reraise(type(err), err, exc_info[2])
             raise
     return exception_stack_wrapper