<commit_before># - # Copyright 2016 <NAME>
# - #
# - # Licensed under the Apache License, Version 2.0 (the "License");
# - # you may not use this file except in compliance with the License.
# - # You may obtain a copy of the License at
# - #
# - #     http://www.apache.org/licenses/LICENSE-2.0
# - #
# - # Unless required by applicable law or agreed to in writing, software
# - # distributed under the License is distributed on an "AS IS" BASIS,
# - # WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# - # See the License for the specific language governing permissions and
# - # limitations under the License.
"""
The kernel is the main thread of execution running inside a twinterpreter.

Any connection between twinterpreters is handled by two kernels. Each
consists of client and server side residing in the different interpreters.

The kernels assume that they have been setup properly. Use
:py:class:`~cpy2py.twinterpreter.twin_master.TwinMaster` start kernel pairs.
"""
import sys
import os
import time
import logging
import threading

from cpy2py.utility.compat import pickle
from cpy2py.kernel import kernel_state

from cpy2py.utility.exceptions import format_exception
from cpy2py.ipyc import ipyc_exceptions
from cpy2py.kernel.kernel_exceptions import StopTwinterpreter
from cpy2py.proxy import proxy_tracker
from cpy2py.kernel.kernel_requesthandler import RequestDispatcher, RequestHandler


def _connect_ipyc(ipyc, pickle_protocol):
    """Connect pickle/unpickle trackers to a duplex IPyC"""
    pickler = pickle.Pickler(ipyc.writer, pickle_protocol)
    pickler.persistent_id = proxy_tracker.persistent_twin_id
    send = pickler.dump
    unpickler = pickle.Unpickler(ipyc.reader)
    unpickler.persistent_load = proxy_tracker.persistent_twin_load
    recv = unpickler.load
    return send, recv


class SingleThreadKernelServer(object):
    """
    Default kernel server for sending requests to other interpreter

    :param peer_id: id of the kernel/twinterpreter this kernel is peered with
    :type peer_id: str
    :param ipyc: :py:mod:`~IPyC` for incoming requests
    :type ipyc: :py:class:`~DuplexFifoIPyC`
    :param pickle_protocol: protocol number for :py:mod:`pickle`
    :type pickle_protocol: int
    """
    def __new__(cls, peer_id, *args, **kwargs):  # pylint: disable=unused-argument
        assert peer_id not in kernel_state.KERNEL_SERVERS, 'Twinterpreters must have unique IDs'
        kernel_state.KERNEL_SERVERS[peer_id] = object.__new__(cls)
        return kernel_state.KERNEL_SERVERS[peer_id]

    def __init__(self, peer_id, ipyc, pickle_protocol=2):
        self._logger = logging.getLogger('__cpy2py__.kernel.%s_to_%s.server' % (kernel_state.TWIN_ID, peer_id))
        self.peer_id = peer_id
        self._ipyc = ipyc
        self._ipyc.open()
        self._server_send, self._server_recv = _connect_ipyc(ipyc, pickle_protocol)
        self._terminate = threading.Event()
        self._terminate.set()
        self._request_handler = RequestHandler(peer_id=self.peer_id, kernel_server=self)

    def run(self):
        """
        Run the kernel request server

        :returns: exit code indicating potential failure
        """
        assert self._terminate.is_set(), 'Kernel already active'
        self._terminate.clear()
        exit_code, request_id = 1, None
        self._logger.warning('Starting kernel %s @ %s', kernel_state.TWIN_ID, time.asctime())
        try:
            while not self._terminate.is_set():
                self._logger.warning('Listening [%s]', kernel_state.TWIN_ID)
                request_id, directive = self._server_recv()
                self._request_handler.serve_request(request_id, directive)
        except StopTwinterpreter as err:
            exit_code = err.exit_code
        # cPickle may raise EOFError by itself
        except (ipyc_exceptions.IPyCTerminated, EOFError):
            exit_code = 0
        except Exception as err:  # pylint: disable=broad-except
            self._logger.critical('TWIN KERNEL INTERNAL EXCEPTION')
            format_exception(self._logger, 3)
        finally:
            self._terminate.set()
            self._logger.critical('TWIN KERNEL SHUTDOWN: %s => %d', kernel_state.TWIN_ID, exit_code)
            self._ipyc.close()
            del kernel_state.KERNEL_SERVERS[self.peer_id]
        return exit_code

    def send_reply(self, request_id, reply_body):
        self._server_send((request_id, reply_body))

    def stop(self):
        """Shutdown the local server"""
        # just tell server thread to stop, it'll cleanup automatically
        self._terminate.set()
        return True

    def __repr__(self):
        return '<%s[%s@%s]>' % (self.__class__.__name__, sys.executable, os.getpid())


class SingleThreadKernelClient(object):
    """
    Default kernel client for sending requests to other interpreter

    :param peer_id: id of the kernel/twinterpreter this kernel is peered with
    :type peer_id: str
    :param ipyc: :py:mod:`~IPyC` for outgoing requests
    :type ipyc: :py:class:`~DuplexFifoIPyC`
    :param pickle_protocol: protocol number for :py:mod:`pickle`
    :type pickle_protocol: int
    """
    def __new__(cls, peer_id, *args, **kwargs):  # pylint: disable=unused-argument<commit_msg>factored out request serving on server<commit_after># - # Copyright 2016 <NAME>
# - #
# - # Licensed under the Apache License, Version 2.0 (the "License");
# - # you may not use this file except in compliance with the License.
# - # You may obtain a copy of the License at
# - #
# - #     http://www.apache.org/licenses/LICENSE-2.0
# - #
# - # Unless required by applicable law or agreed to in writing, software
# - # distributed under the License is distributed on an "AS IS" BASIS,
# - # WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# - # See the License for the specific language governing permissions and
# - # limitations under the License.
"""
The kernel is the main thread of execution running inside a twinterpreter.

Any connection between twinterpreters is handled by two kernels. Each
consists of client and server side residing in the different interpreters.

The kernels assume that they have been setup properly. Use
:py:class:`~cpy2py.twinterpreter.twin_master.TwinMaster` start kernel pairs.
"""
import sys
import os
import time
import logging
import threading

from cpy2py.utility.compat import pickle
from cpy2py.kernel import kernel_state

from cpy2py.utility.exceptions import format_exception
from cpy2py.ipyc import ipyc_exceptions
from cpy2py.kernel.kernel_exceptions import StopTwinterpreter
from cpy2py.proxy import proxy_tracker
from cpy2py.kernel.kernel_requesthandler import RequestDispatcher, RequestHandler


def _connect_ipyc(ipyc, pickle_protocol):
    """Connect pickle/unpickle trackers to a duplex IPyC"""
    pickler = pickle.Pickler(ipyc.writer, pickle_protocol)
    pickler.persistent_id = proxy_tracker.persistent_twin_id
    send = pickler.dump
    unpickler = pickle.Unpickler(ipyc.reader)
    unpickler.persistent_load = proxy_tracker.persistent_twin_load
    recv = unpickler.load
    return send, recv


class SingleThreadKernelServer(object):
    """
    Default kernel server for sending requests to other interpreter

    :param peer_id: id of the kernel/twinterpreter this kernel is peered with
    :type peer_id: str
    :param ipyc: :py:mod:`~IPyC` for incoming requests
    :type ipyc: :py:class:`~DuplexFifoIPyC`
    :param pickle_protocol: protocol number for :py:mod:`pickle`
    :type pickle_protocol: int
    """
    def __new__(cls, peer_id, *args, **kwargs):  # pylint: disable=unused-argument
        assert peer_id not in kernel_state.KERNEL_SERVERS, 'Twinterpreters must have unique IDs'
        kernel_state.KERNEL_SERVERS[peer_id] = object.__new__(cls)
        return kernel_state.KERNEL_SERVERS[peer_id]

    def __init__(self, peer_id, ipyc, pickle_protocol=2):
        self._logger = logging.getLogger('__cpy2py__.kernel.%s_to_%s.server' % (kernel_state.TWIN_ID, peer_id))
        self.peer_id = peer_id
        self._ipyc = ipyc
        self._ipyc.open()
        self._server_send, self._server_recv = _connect_ipyc(ipyc, pickle_protocol)
        self._terminate = threading.Event()
        self._terminate.set()
        self._request_handler = RequestHandler(peer_id=self.peer_id, kernel_server=self)

    def run(self):
        """
        Run the kernel request server

        :returns: exit code indicating potential failure
        """
        assert self._terminate.is_set(), 'Kernel already active'
        self._terminate.clear()
        exit_code, request_id = 1, None
        self._logger.warning('Starting kernel %s @ %s', kernel_state.TWIN_ID, time.asctime())
        try:
            self._serve_requests()
        except StopTwinterpreter as err:
            exit_code = err.exit_code
        # cPickle may raise EOFError by itself
        except (ipyc_exceptions.IPyCTerminated, EOFError):
            exit_code = 0
        except Exception as err:  # pylint: disable=broad-except
            self._logger.critical('TWIN KERNEL INTERNAL EXCEPTION')
            format_exception(self._logger, 3)
        finally:
            self._terminate.set()
            self._logger.critical('TWIN KERNEL SHUTDOWN: %s => %d', kernel_state.TWIN_ID, exit_code)
            self._ipyc.close()
            del kernel_state.KERNEL_SERVERS[self.peer_id]
        return exit_code

    def _serve_requests(self):
        while not self._terminate.is_set():
            self._logger.warning('Listening [%s]', kernel_state.TWIN_ID)
            request_id, directive = self._server_recv()
            self._request_handler.serve_request(request_id, directive)

    def send_reply(self, request_id, reply_body):
        self._server_send((request_id, reply_body))

    def stop(self):
        """Shutdown the local server"""
        # just tell server thread to stop, it'll cleanup automatically
        self._terminate.set()
        return True

    def __repr__(self):
        return '<%s[%s@%s]>' % (self.__class__.__name__, sys.executable, os.getpid())


class SingleThreadKernelClient(object):
    """
    Default kernel client for sending requests to other interpreter

    :param peer_id: id of the kernel/twinterpreter this kernel is peered with
    :type peer_id: str
    :param ipyc: :py:mod:`~IPyC` for outgoing requests
    :type ipyc: :py:class:`~DuplexFifoIPyC`
    :param pickle_protocol: protocol number for :py:mod:`pickle`
    :type pickle_protocol: int
    """
    def __new__(cls, peer_id, *args, **kwargs):  # pylint: disable=unused-argument
{'fraction_non_alphanumeric': 0.09037845980041423, 'fraction_numerical': 0.003954057616268123, 'mean_word_length': 3.384234420140322, 'pattern_counts': {'":': 0, '<': 7, '<?xml version=': 0, '>': 9, 'https://': 0, 'lorem ipsum': 0, 'www.': 2, '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': '2860953', 'n_tokens_mistral': 3257, 'n_tokens_neox': 3166, 'n_words': 901}