<commit_before>#
#  The OpenDiamond Platform for Interactive Search
#
#  Copyright (c) 2011-2019 Carnegie Mellon University
#  All rights reserved.
#
#  This software is distributed under the terms of the Eclipse Public
#  License, Version 1.0 which can be found in the file named LICENSE.
#  ANY USE, REPRODUCTION OR DISTRIBUTION OF THIS SOFTWARE CONSTITUTES
#  RECIPIENT'S ACCEPTANCE OF THIS AGREEMENT
#

'''Search state; control and blast channel handling.'''

from builtins import str
from builtins import object
from functools import wraps
import logging
import multiprocessing as mp
import os
import signal
import threading

from opendiamond import protocol
from opendiamond.blobcache import ExecutableBlobCache
from opendiamond.protocol import (
    DiamondRPCFailure, DiamondRPCFCacheMiss, DiamondRPCCookieExpired,
    DiamondRPCSchemeNotSupported)
from opendiamond.rpc import RPCHandlers, RPCError, RPCProcedureUnavailable
from opendiamond.scope import ScopeCookie, ScopeError, ScopeCookieExpired
from opendiamond.server.filter import (
    FilterStack, Filter, FilterDependencyError, FilterUnsupportedSource)
from opendiamond.server.object_ import EmptyObject, Object, ObjectLoader
from opendiamond.server.scopelist import ScopeListLoader
from opendiamond.server.sessionvars import SessionVariables
from opendiamond.server.statistics import SearchStatistics
from opendiamond.server.resource import ResourceContext

_log = logging.getLogger(__name__)


class SearchState(object):
    '''Search state that is also needed by filter code.'''

    def __init__(self, config):
        self.config = config
        self.blob_cache = ExecutableBlobCache(config.cachedir)
        self.session_vars = SessionVariables()
        self.stats = SearchStatistics()
        self.blast = None
        # TODO change to something session-dependent
        self.context = None


class Search(RPCHandlers):
    '''State for a single search, plus handlers for control channel RPCs
    to modify it.'''

    log_rpcs = True

    def __init__(self, config, blast_conn):
        RPCHandlers.__init__(self)
        self._server_id = config.serverids[0]  # Canonical server ID
        self._blast_conn = blast_conn
        self._state = SearchState(config)
        self._filters = FilterStack()
        self._running = False
        self._scope = None
        self._workers = None

    def shutdown(self):
        '''Clean up the search before the process exits.'''

        # Clean up the resource context before terminate() to avoid corrupting the shared data structures.
        self._state.context.cleanup()

        while self._workers:
            p = self._workers.pop()
            _log.debug("Terminating worker process %d", p.pid)
            try:
                p.terminate()   # send SIGTERM, which will be caught as _Signalled in the child
                p.join(1)   # allow it to clean up
                os.kill(p.pid, signal.SIGKILL)  # shoot it in the head
                p.join()
            except OSError: # perhaps process already dead
                pass

        # Log search statistics
        if self._running:
            self._state.stats.log()
            for filter in self._filters:
                filter.stats.log()

    # This is not a static method: it's only called when initializing the
    # class, and the staticmethod() decorator does not create a callable.
    # Also avoid complaints about accesses to self._running
    # pylint: disable=no-self-argument,protected-access
    def running(should_be_running):
        '''Decorator that specifies that the handler can only be called
        before, or after, the search has started running.'''

        def decorator(func):
            @wraps(func)
            def wrapper(self, *args, **kwargs):
                if self._running != should_be_running:
                    raise RPCProcedureUnavailable()
                return func(self, *args, **kwargs)

            return wrapper

        return decorator

    # pylint: enable=no-self-argument,protected-access

    def _check_runnable(self):
        '''Validate state preparatory to starting a search or reexecution.'''
        if self._scope is None:
            raise DiamondRPCFailure('No search scope configured')
        if not self._filters:
            raise DiamondRPCFailure('No filters configured')
        # Ensure we have all filter code and blob arguments
        try:
            for filter in self._filters:
                filter.resolve(self._state)
        except FilterDependencyError as e:
            raise DiamondRPCFCacheMiss(str(e))

    @RPCHandlers.handler(25, protocol.XDR_setup, protocol.XDR_blob_list)<commit_msg>allow for searching without filters<commit_after>#
#  The OpenDiamond Platform for Interactive Search
#
#  Copyright (c) 2011-2019 Carnegie Mellon University
#  All rights reserved.
#
#  This software is distributed under the terms of the Eclipse Public
#  License, Version 1.0 which can be found in the file named LICENSE.
#  ANY USE, REPRODUCTION OR DISTRIBUTION OF THIS SOFTWARE CONSTITUTES
#  RECIPIENT'S ACCEPTANCE OF THIS AGREEMENT
#

'''Search state; control and blast channel handling.'''

from builtins import str
from builtins import object
from functools import wraps
import logging
import multiprocessing as mp
import os
import signal
import threading

from opendiamond import protocol
from opendiamond.blobcache import ExecutableBlobCache
from opendiamond.protocol import (
    DiamondRPCFailure, DiamondRPCFCacheMiss, DiamondRPCCookieExpired,
    DiamondRPCSchemeNotSupported)
from opendiamond.rpc import RPCHandlers, RPCError, RPCProcedureUnavailable
from opendiamond.scope import ScopeCookie, ScopeError, ScopeCookieExpired
from opendiamond.server.filter import (
    FilterStack, Filter, FilterDependencyError, FilterUnsupportedSource)
from opendiamond.server.object_ import EmptyObject, Object, ObjectLoader
from opendiamond.server.scopelist import ScopeListLoader
from opendiamond.server.sessionvars import SessionVariables
from opendiamond.server.statistics import SearchStatistics
from opendiamond.server.resource import ResourceContext

_log = logging.getLogger(__name__)


class SearchState(object):
    '''Search state that is also needed by filter code.'''

    def __init__(self, config):
        self.config = config
        self.blob_cache = ExecutableBlobCache(config.cachedir)
        self.session_vars = SessionVariables()
        self.stats = SearchStatistics()
        self.blast = None
        # TODO change to something session-dependent
        self.context = None


class Search(RPCHandlers):
    '''State for a single search, plus handlers for control channel RPCs
    to modify it.'''

    log_rpcs = True

    def __init__(self, config, blast_conn):
        RPCHandlers.__init__(self)
        self._server_id = config.serverids[0]  # Canonical server ID
        self._blast_conn = blast_conn
        self._state = SearchState(config)
        self._filters = FilterStack()
        self._running = False
        self._scope = None
        self._workers = None

    def shutdown(self):
        '''Clean up the search before the process exits.'''

        # Clean up the resource context before terminate() to avoid corrupting the shared data structures.
        self._state.context.cleanup()

        while self._workers:
            p = self._workers.pop()
            _log.debug("Terminating worker process %d", p.pid)
            try:
                p.terminate()   # send SIGTERM, which will be caught as _Signalled in the child
                p.join(1)   # allow it to clean up
                os.kill(p.pid, signal.SIGKILL)  # shoot it in the head
                p.join()
            except OSError: # perhaps process already dead
                pass

        # Log search statistics
        if self._running:
            self._state.stats.log()
            for filter in self._filters:
                filter.stats.log()

    # This is not a static method: it's only called when initializing the
    # class, and the staticmethod() decorator does not create a callable.
    # Also avoid complaints about accesses to self._running
    # pylint: disable=no-self-argument,protected-access
    def running(should_be_running):
        '''Decorator that specifies that the handler can only be called
        before, or after, the search has started running.'''

        def decorator(func):
            @wraps(func)
            def wrapper(self, *args, **kwargs):
                if self._running != should_be_running:
                    raise RPCProcedureUnavailable()
                return func(self, *args, **kwargs)

            return wrapper

        return decorator

    # pylint: enable=no-self-argument,protected-access

    def _check_runnable(self):
        '''Validate state preparatory to starting a search or reexecution.'''
        if self._scope is None:
            raise DiamondRPCFailure('No search scope configured')
        if not self._filters:
            _log.warning('No filters configured')
        # Ensure we have all filter code and blob arguments
        try:
            for filter in self._filters:
                filter.resolve(self._state)
        except FilterDependencyError as e:
            raise DiamondRPCFCacheMiss(str(e))

    @RPCHandlers.handler(25, protocol.XDR_setup, protocol.XDR_blob_list)
{'fraction_non_alphanumeric': 0.06851950828433992, 'fraction_numerical': 0.0029930518439337255, 'mean_word_length': 2.8391464915880182, 'pattern_counts': {'":': 0, '<': 3, '<?xml version=': 0, '>': 3, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '95360', 'n_tokens_mistral': 2525, 'n_tokens_neox': 2389, 'n_words': 834}