<commit_before>#
# ovirt-hosted-engine-ha -- ovirt hosted engine high availability
# Copyright (C) 2013 Red Hat, Inc.
#
# This library is free software; you can redistribute it and/or
# modify it under the terms of the GNU Lesser General Public
# License as published by the Free Software Foundation; either
# version 2.1 of the License, or (at your option) any later version.
#
# This library is distributed in the hope that it will be useful,
# but WITHOUT ANY WARRANTY; without even the implied warranty of
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the GNU
# Lesser General Public License for more details.
#
# You should have received a copy of the GNU Lesser General Public
# License along with this library; if not, write to the Free Software
# Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301 USA
#

import errno
import logging
import os
import socket
import SocketServer
import threading

from . import constants
from ..lib import util
from ..lib.exceptions import DisconnectionError
from ..lib.exceptions import RequestError


class Listener(object):
    def __init__(self, monitor_instance, storage_broker_instance):
        """
        Prepare the listener and associated locks
        """
        self._log = logging.getLogger("Listener")
        self._log.info("Initializing SocketServer")

        # Coordinate access to resources across connections
        self._conn_monitors = {}
        self._conn_monitors_access_lock = threading.Lock()
        self._monitor_instance = monitor_instance
        self._monitor_instance_access_lock = threading.Lock()
        self._storage_broker_instance = storage_broker_instance
        self._storage_broker_instance_access_lock = threading.Lock()
        self._need_exit = False

        if os.path.exists(constants.SOCKET_FILE):
            os.unlink(constants.SOCKET_FILE)
        self._server = ThreadedStreamServer(
            constants.SOCKET_FILE, ConnectionHandler, True, self)

        self._log.info("SocketServer ready")

    @property
    def conn_monitors(self):
        return self._conn_monitors

    @property
    def conn_monitors_access_lock(self):
        return self._conn_monitors_access_lock

    @property
    def monitor_instance(self):
        return self._monitor_instance

    @property
    def monitor_instance_access_lock(self):
        return self._monitor_instance_access_lock

    @property
    def storage_broker_instance(self):
        return self._storage_broker_instance

    @property
    def storage_broker_instance_access_lock(self):
        return self._storage_broker_instance_access_lock

    @property
    def need_exit(self):
        return self._need_exit

    def listen(self):
        """
        Listen briefly and return to the main loop.
        """
        # Using SocketServer.server_forever() is tempting, but it's easier to
        # shut down cleanly (and with less chance of deadlock) by returning
        # control to a main loop which checks for exit requests.
        self._server.handle_request()

    def close_connections(self):
        """
        Signal to all connection handlers that it's time to exit.
        This function must remain re-entrant.
        """
        self._need_exit = True


class ThreadedStreamServer(SocketServer.ThreadingMixIn,
                           SocketServer.UnixStreamServer):
    def __init__(self, socket_file, handler, bind_and_activate, listener):
        # Coordinate access to resources across connections through caller
        self.sp_listener = listener

        SocketServer.UnixStreamServer.__init__(self, socket_file, handler,
                                               bind_and_activate)
        self.timeout = self.sp_listener.need_exit


class ConnectionHandler(SocketServer.BaseRequestHandler):
    def setup(self):
        """
        Overridden method.
        Initialize connection handler class and identify connection.
        """
        if not hasattr(self, '_log'):
            self._log = logging.getLogger("ConnectionHandler")
        self._log.info("Connection established")
        self.request.settimeout(0.7)
        SocketServer.BaseRequestHandler.setup(self)
        self._init_monitors_for_conn()

    def handle(self):
        """
        Overridden method.
        Handle an incoming connection.
        """
        while not self.server.sp_listener.need_exit:
            try:
                data = util.socket_readline(self.request, self._log)
                self._log.debug("Input: %s", data)
                try:
                    response = "success " + self._dispatch(data)
                except RequestError as e:
                    response = "failure " + format(str(e))
                self._log.debug("Response: %s", response)
                self.request.sendall(response + "\n")
            except socket.timeout:
                pass
            except socket.error as e:
                if e.errno == errno.EPIPE:
                    self._log.info("Connection closed")
                    return
                else:
                    self._log.error("Error while serving connection",
                                    exc_info=True)
            except DisconnectionError:
                self._log.info("Connection closed")
                return
        if self.server.sp_listener.need_exit:
            self._log.info("Closing connection on server request")

    def finish(self):
        """
        Overridden method.
        Silence exception when a connection is closed remotely.
        """<commit_msg>broker: log errors during request processing<commit_after>#
# ovirt-hosted-engine-ha -- ovirt hosted engine high availability
# Copyright (C) 2013 Red Hat, Inc.
#
# This library is free software; you can redistribute it and/or
# modify it under the terms of the GNU Lesser General Public
# License as published by the Free Software Foundation; either
# version 2.1 of the License, or (at your option) any later version.
#
# This library is distributed in the hope that it will be useful,
# but WITHOUT ANY WARRANTY; without even the implied warranty of
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the GNU
# Lesser General Public License for more details.
#
# You should have received a copy of the GNU Lesser General Public
# License along with this library; if not, write to the Free Software
# Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301 USA
#

import errno
import logging
import os
import socket
import SocketServer
import threading

from . import constants
from ..lib import util
from ..lib.exceptions import DisconnectionError
from ..lib.exceptions import RequestError


class Listener(object):
    def __init__(self, monitor_instance, storage_broker_instance):
        """
        Prepare the listener and associated locks
        """
        self._log = logging.getLogger("Listener")
        self._log.info("Initializing SocketServer")

        # Coordinate access to resources across connections
        self._conn_monitors = {}
        self._conn_monitors_access_lock = threading.Lock()
        self._monitor_instance = monitor_instance
        self._monitor_instance_access_lock = threading.Lock()
        self._storage_broker_instance = storage_broker_instance
        self._storage_broker_instance_access_lock = threading.Lock()
        self._need_exit = False

        if os.path.exists(constants.SOCKET_FILE):
            os.unlink(constants.SOCKET_FILE)
        self._server = ThreadedStreamServer(
            constants.SOCKET_FILE, ConnectionHandler, True, self)

        self._log.info("SocketServer ready")

    @property
    def conn_monitors(self):
        return self._conn_monitors

    @property
    def conn_monitors_access_lock(self):
        return self._conn_monitors_access_lock

    @property
    def monitor_instance(self):
        return self._monitor_instance

    @property
    def monitor_instance_access_lock(self):
        return self._monitor_instance_access_lock

    @property
    def storage_broker_instance(self):
        return self._storage_broker_instance

    @property
    def storage_broker_instance_access_lock(self):
        return self._storage_broker_instance_access_lock

    @property
    def need_exit(self):
        return self._need_exit

    def listen(self):
        """
        Listen briefly and return to the main loop.
        """
        # Using SocketServer.server_forever() is tempting, but it's easier to
        # shut down cleanly (and with less chance of deadlock) by returning
        # control to a main loop which checks for exit requests.
        self._server.handle_request()

    def close_connections(self):
        """
        Signal to all connection handlers that it's time to exit.
        This function must remain re-entrant.
        """
        self._need_exit = True


class ThreadedStreamServer(SocketServer.ThreadingMixIn,
                           SocketServer.UnixStreamServer):
    def __init__(self, socket_file, handler, bind_and_activate, listener):
        # Coordinate access to resources across connections through caller
        self.sp_listener = listener

        SocketServer.UnixStreamServer.__init__(self, socket_file, handler,
                                               bind_and_activate)
        self.timeout = self.sp_listener.need_exit


class ConnectionHandler(SocketServer.BaseRequestHandler):
    def setup(self):
        """
        Overridden method.
        Initialize connection handler class and identify connection.
        """
        if not hasattr(self, '_log'):
            self._log = logging.getLogger("ConnectionHandler")
        self._log.info("Connection established")
        self.request.settimeout(0.7)
        SocketServer.BaseRequestHandler.setup(self)
        self._init_monitors_for_conn()

    def handle(self):
        """
        Overridden method.
        Handle an incoming connection.
        """
        while not self.server.sp_listener.need_exit:
            data = None
            try:
                data = util.socket_readline(self.request, self._log)
                self._log.debug("Input: %s", data)
                try:
                    response = "success " + self._dispatch(data)
                except RequestError as e:
                    response = "failure " + format(str(e))
                self._log.debug("Response: %s", response)
                self.request.sendall(response + "\n")
            except socket.timeout:
                pass
            except socket.error as e:
                if e.errno == errno.EPIPE:
                    self._log.info("Connection closed")
                    return
                else:
                    self._log.error("Error while serving connection",
                                    exc_info=True)
            except DisconnectionError:
                self._log.info("Connection closed")
                return
            except Exception:
                self._log.error("Error handling request, data: %r",
                                data, exc_info=True)
                return
        if self.server.sp_listener.need_exit:
            self._log.info("Closing connection on server request")

    def finish(self):
        """
        Overridden method.
        Silence exception when a connection is closed remotely.
        """
{'fraction_non_alphanumeric': 0.07222662092020762, 'fraction_numerical': 0.003343010468901205, 'mean_word_length': 2.33665981802172, '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': 6, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '4216955', 'n_tokens_mistral': 2920, 'n_tokens_neox': 2802, 'n_words': 861}