def murmur(data):
    '''Return a lower-case hex string representing the MurmurHash3-x64-128
    hash of the specified data using the seed 0xbb40e64d (the first 10
    digits of pi, in hex).'''
    return mmh3.hash_bytes(data, 314159).hex() def signalname(signum):
    '''Return the name for the specified signal number.'''
    for attr in dir(signal):
        if (attr.startswith('SIG') and not attr.startswith('SIG_') and
                getattr(signal, attr) == signum):
            return attr
    return 'signal %d' % signum def split_scheme(url):
    if sys.version_info[0:3] == (2, 7, 1) and url.startswith('sha256:'):
        return url.split(':')

    parts = urlparse(url)
    return (parts.scheme, parts.path) class ConnectionFailure(Exception):
    '''RPC connection died.''' ATTR_DATA = '' ATTR_OBJ_ID = '_ObjectID' class ObjectLoader(object):
    """A context for populating an Object from the dataretriever.  Allows
    network connections to be reused to fetch multiple objects.  Must not
    be used by more than one thread.

    Attributes are also loaded. There are several possible places attributes
    are loaded from. The precedence order is (later overwrites earlier):
    - 'meta' URI from scope list <object /> element, which can either be a URL or
    local file path.
    - HTTP headers starting with ATTR_HEADER_PREFIX='x-attr-', when retrieving
    the object itself. (happen only when retrieving from data retriever)
    - HTTP header ATTR_HEADER_URL='x-attributes', when retrieving the object itself.
    (happen only when retrieving from data retriever)
    """

    def __init__(self, config, blob_cache):
        self._http = _HttpLoader(config)
        self._blob_cache = blob_cache

    def source_available(self, obj):
        '''Examine the Object and return whether we think we will be able
        to load it, without actually doing so.  This can be used during
        reexecution to determine whether we should return
        DiamondRPCFCacheMiss to the client.'''
        scheme, path = split_scheme(str(obj))
        if scheme == 'sha256':
            return path in self._blob_cache
        # Assume we can always load other types of URLs
        return True

    def load(self, obj):
        '''Retrieve the Object and update it with the information we
        receive.'''

        # 0. Load src URI if not exist (mainly used for re-execution)
        self._load_src_maybe_meta(obj)

        # 1. Load object metadata from meta URI
        if hasattr(obj, 'meta'):
            uri = str(obj.meta)
            scheme, path = split_scheme(uri)
            if scheme == 'file':
                self._load_attrbutes_localfile(obj, path)
            else:  # assume http
                self._load_attributes(obj, uri)

        # 2. Load object content from src URI
        uri = str(obj.src)
        scheme, path = split_scheme(uri)
        if scheme == 'sha256':
            self._load_blobcache(obj, path)
        elif scheme == 'file':
            self._load_localfile(obj, path)
        else:  # assume http
            self._load_dataretriever(obj, uri)

        # Set display name if not already in initial attributes
        if ATTR_DISPLAY_NAME not in obj:
            obj[ATTR_DISPLAY_NAME] = str(obj) + '\0'

    def _load_blobcache(self, obj, signature):
        # Load the object data
        try:
            obj[ATTR_DATA] = self._blob_cache[signature]
        except KeyError:
            raise ObjectLoadError('Object not in cache')

    def _load_dataretriever(self, obj, url):
        headers, body = self._http.get(url)
        # Load the object data
        obj[ATTR_DATA] = body
        # Process loose initial attributes
        for key, value in headers.items():
            if key.lower().startswith(ATTR_HEADER_PREFIX):
                key = key[ATTR_HEADER_PREFIX_LEN:]
                obj[key] = value + '\0'
        # Fetch additional initial attributes if specified
        if ATTR_HEADER_URL in headers:
            attr_url = urljoin(url, headers[ATTR_HEADER_URL])
            self._load_attributes(obj, attr_url)

    def _load_localfile(self, obj, path):
        with open(path, 'rb') as f:
            obj[ATTR_DATA] = f.read()

    # The return type of json.loads() confuses pylint
    # pylint: disable=maybe-no-member
    def _load_attributes(self, obj, url):
        '''Load JSON-encoded attribute data from the specified URL.'''
        _headers, body = self._http.get(url)
        try:
            attrs = json.loads(body)
            if not isinstance(attrs, dict):
                raise ObjectLoadError("Failed to retrieve object attributes")
        except ValueError as e:
            raise ObjectLoadError(str(e))
        for k, v in attrs.items():
            obj[k] = str(v) + '\0'

    # pylint: enable=maybe-no-member

    def _load_attrbutes_localfile(self, obj, path):
        try:
            with open(path, 'r') as f:
                attrs = json.load(f)
                if not isinstance(attrs, dict):
                    raise ObjectLoadError(
                        "Failed to retrieve object attributes")
        except IOError as e:
            raise ObjectLoadError(str(e))
        except ValueError as e:
            raise ObjectLoadError(str(e))
        for k, v in attrs.items():
            obj[k] = str(v) + '\0'

    def _load_src_maybe_meta(self, obj):
        """
        Load the src and optional meta URIs from the object's id.
        :param obj:
        :return:
        """
        if not hasattr(obj, 'src'):
            uri = str(obj)
            scheme, path = split_scheme(uri)
            if scheme == 'sha256':
                # Blob has no meta URI
                obj.src = uri
            else:
                # Assume http
                _, body = self._http.get(uri)
                object_el = xmltodict.parse(body)
                obj.src = urljoin(uri, object_el['object']['@src'])
                if '@meta' in object_el['object']:
                    obj.meta = urljoin(uri, object_el['object']['@meta'])
        return class ObjectLoadError(Exception):
    '''Object failed to load.''' class FilterStatistics(_Statistics):
    '''Statistics for the execution of a single filter.'''

    attrs = (('objs_processed', 'Total objects considered'),
             ('objs_dropped', 'Total objects dropped'),
             ('objs_true_positive', 'Objects passed with label'),
             ('objs_false_negative', 'Objects dropped with label'),
             ('objs_cache_dropped', 'Objects dropped by cache'),
             ('objs_cache_passed', 'Objects skipped by cache'),
             ('objs_computed', 'Objects examined by filter'),
             ('objs_terminate', 'Objects causing filter to terminate'),
             ('execution_us', 'Filter execution time (us)'),
             )

    def __init__(self, name):
        super(FilterStatistics, self).__init__()
        self.name = name
        self.label = 'Filter statistics for %s' % name

    def xdr(self):
        '''Return an XDR statistics structure for these statistics.'''
        with self.lock:
            try:
                avg_exec_us = old_div(self.execution_us, self.objs_processed)
            except ZeroDivisionError:
                avg_exec_us = 0

            stats = []
            stats.append(XDR_stat('avg_exec_time_us', avg_exec_us))
            for name, _desc in self.attrs:
                if name != 'execution_us':
                    stats.append(XDR_stat(name, getattr(self, name)))

            return XDR_filter_stats(
                name=self.name,
                stats=stats
            ) class Timer(object):
    '''Tracks the elapsed time since the Timer object was created.'''

    def __init__(self):
        self._start = time.time()

    @property
    def elapsed_seconds(self):
        '''Elapsed time in seconds.'''
        return time.time() - self._start

    @property
    def elapsed(self):
        '''Elapsed time in us.'''
        return int(self.elapsed_seconds * 1e6)

    def reset(self):
        self._start = time.time() class FilterRunnerLogger(object):

    def __init__(self, stats):
        assert isinstance(stats, FilterStatistics)
        self.stats = stats
        self.eval_timer = Timer()

    def on_connected(self):
        pass

    def on_initialized(self):
        pass

    def on_start_evaluate(self):
        self.eval_timer.reset()

    def on_done_evaluate(self, accept, gt_present=False):
        with self.stats.lock:
            self.stats.execution_us += self.eval_timer.elapsed
            self.stats.objs_processed += 1
            self.stats.objs_computed += 1
            self.stats.objs_dropped += int(not accept)
            self.stats.objs_true_positive += int(accept and gt_present)
            self.stats.objs_false_negative += int(not accept and gt_present)

    def on_cache_hit(self, accept, gt_present=False):
        with self.stats.lock:
            self.stats.objs_processed += 1
            self.stats.objs_dropped += int(not accept)
            self.stats.objs_cache_dropped += int(not accept)
            self.stats.objs_cache_passed += int(accept)
            self.stats.objs_true_positive += int(accept and gt_present)
            self.stats.objs_false_negative += int(not accept and gt_present)

    def on_terminate(self):
        with self.stats.lock:
            self.stats.objs_terminate += 1 class FilterStackRunnerLogger(object):

    def __init__(self, stats):
        assert isinstance(stats, SearchStatistics)
        self.stats = stats
        self.eval_timer = Timer()

        # record some per-thread statistics
        self.objs_processed = self.objs_passed = self.objs_dropped = self.objs_unloadable = 0

    def on_start_evaluate(self):
        self.eval_timer.reset()

    def on_done_evaluate(self, accept, gt_present=False):
        with self.stats.lock:
            self.stats.objs_processed += 1
            self.stats.objs_passed += int(accept)
            self.stats.objs_dropped += int(not accept)
            self.stats.objs_true_positive += int(accept and gt_present)
            self.stats.objs_false_negative += int(not accept and gt_present)
            self.stats.execution_us += self.eval_timer.elapsed

        # if self.objs_processed % 1000 == 0:
        #     _log.debug('Processed %d objects', self.objs_processed)

        self.objs_processed += 1
        self.objs_passed += int(accept)
        self.objs_dropped += int(not accept)

    def on_unloadable(self):
        with self.stats.lock:
            self.stats.objs_unloadable += 1

        self.objs_unloadable += 1

    def on_finish(self):
        _log.info(
            'Worker: objs processed/passed/dropped/unloadable: %d/%d/%d/%d',
            self.objs_processed, self.objs_passed, self.objs_dropped, self.objs_unloadable) class NoLogger(object):
    def __init__(self, stats):
        pass
    
    def on_connected(self):
        pass

    def on_initialized(self):
        pass

    def on_start_evaluate(self):
        pass

    def on_done_evaluate(self, accept):
        pass

    def on_cache_hit(self, accept):
        pass

    def on_terminate(self):
        pass

    def on_unloadable(self):
        pass

    def on_finish(self):
        pass
from builtins import map
from builtins import zip
from builtins import str
from builtins import range
from past.utils import old_div
from builtins import object
from redis import Redis
from redis.exceptions import ResponseError
from opendiamond.helpers import murmur, signalname, split_scheme
from opendiamond.rpc import ConnectionFailure
from opendiamond.server.object_ import ATTR_DATA, ATTR_OBJ_ID, ObjectLoader, ObjectLoadError
from opendiamond.server.statistics import FilterStatistics, Timer, \
    FilterRunnerLogger, FilterStackRunnerLogger, NoLogger
        from opendiamond.server.server import _Signalled
import docker
import fcntl
import logging
import os
import psutil
import simplejson as json
import signal
import socket
import subprocess
import threading
import multiprocessing as mp
import time
import uuid
import yaml
        import gc
                        level = logging.INFO
                    elif level & 0x08:
                        # LOGL_TRACE.  Very verbose; ignore.
                        continue
                    elif level & 0x10:
                        # LOGL_DEBUG
                        level = logging.DEBUG
                    else:
                        level = logging.DEBUG
                    if self._proc_initialized:
                        _log.log(level, 'object: %s. %s' % (str(obj), message))
                    else:
                        _log.log(level, 'Initialize: %s' % message)
                elif cmd == 'stdout':
                    print(proc.get_item().decode(), end=' ')
                elif cmd == 'result':
                    result.score = float(proc.get_item())
                    break
                elif cmd == 'ensure-resource':
                    # Create scoped resource here
                    scope = proc.get_item().decode()
                    rtype = proc.get_item().decode()
                    args = proc.get_array()
                    args = list(map(bytes.decode, args))
                    if scope == 'session':
                        _log.debug("Filter asks to ensure resource: %s, %s", rtype, str(args))
                        uri = self._state.context.ensure_resource(rtype, *args)
                    else:
                        raise FilterExecutionError(
                            "Unrecognized resource scope" % scope
                        )
                    proc.send_dict(uri)
                elif cmd == 'hint-large-attribute':
                    size = int(proc.get_item())
                    _log.info("Hint: attribute size %d", size)
                    proc.hint_large_attribute(size)
                    break
                elif cmd == '':
                    # Encountered EOF on pipe
                    raise IOError()
                else:
                    raise FilterExecutionError('%s: unknown command: %s' % (self, cmd))
        except IOError:
            if self._proc_initialized:
                # Filter died on an object.  Drop the object without caching
                # the result.
                _log.error('Filter %s (signature %s) died on object %s',
                           self, self._filter.signature, obj)
                self._logger.on_terminate()
                self._proc = None
                raise _DropObject()
            else:
                # Filter died during initialization.  Treat this as fatal.
                raise FilterExecutionError("Filter %s failed to initialize"
                                           % self)
        finally:
            accept = self.threshold(result)
            gt_present = ATTR_GT_LABEL in list(result.input_attrs.keys())
            self._logger.on_done_evaluate(accept, gt_present)
            lengths = [len(obj[k]) for k in result.output_attrs]
            throughput = int(old_div(sum(lengths), timer.elapsed_seconds))
            if throughput < ATTRIBUTE_CACHE_THRESHOLD:
                result.cache_output = True
        return result

    def threshold(self, result):
        return self._filter.min_score <= result.score <= self._filter.max_score


class Filter(object):
    '''A filter with arguments.'''

    def __init__(self, name, code_source, blob_source, min_score, max_score,
                 arguments, dependencies):
        self.name = name
        self.code_source = code_source
        self.blob_source = blob_source
        self.min_score = min_score
        self.max_score = max_score
        self.arguments = arguments
        self.dependencies = dependencies
        self.stats = FilterStatistics(name)

        # Will be initialized during resolve()
        self.code_path = None
        self.signature = None
        self.blob = None
        self.cache_digest = None
        self.mode = None

    def connect(self):
        """Return a FilterConnection. To be reloaded during resolve()"""
        assert self is not None
        raise FilterDependencyError('Filter mode is not resolved yet.')

    def resolve(self, state):
        '''Ensure filter code and blob argument are available in the blob
        cache, load the blob argument, and initialize the cache digest.'''
        if self.code_path is not None:
            return
        # Get path to filter code
        code_path, code_signature = self._resolve_code(state)
        # Get contents of blob argument
        blob, blob_signature = self._resolve_blob(state)
        # Initialize digest
        summary = ([code_signature, self.name] + self.arguments +
                   [blob_signature])
        cache_digest = murmur(' '.join(summary))
        # Commit
        self.code_path = code_path
        self.signature = code_signature
        self.blob = blob
        self.cache_digest = cache_digest

        # Resolve mode and reload connect()
        connector = self._resolve_mode(state)
        self.connect = connector.__get__(self, None)  # bind to instance

    def _resolve_code(self, state):
        '''Returns (code_path, signature).'''

        scheme, path = split_scheme(self.code_source)