<commit_before># coding: utf-8

import time
import logging
from threading import Thread, Lock

from core.exceptions import CreationException

from core import constants
from core.utils import call_in_thread
from core.profiler import profiler

log = logging.getLogger(__name__)


class EndpointRemover(Thread):
    def __init__(self, platforms, artifact_collector, database, app_context):
        super(EndpointRemover, self).__init__()
        self.running = True
        self.daemon = True
        self.platforms = platforms
        self.artifact_collector = artifact_collector
        self.database = database
        self.app_context = app_context
        self.lock = Lock()

    @call_in_thread
    def remove_endpoint(self, endpoint, try_to_rebuild=False):
        try:
            with self.app_context():
                self.endpoint_service_mode_on(endpoint)
                session = self.database.get_session_by_endpoint_id(endpoint.id)
                if session:
                    session.restore()
                    self.artifact_collector.save_selenium_log(session)
                    self.artifact_collector.wait_for_complete(session.id)
                endpoint.delete(try_to_rebuild=try_to_rebuild)
                self.endpoint_service_mode_off(endpoint)
        except:
            log.exception("Attempt to remove {} was failed".format(endpoint))
            endpoint.send_to_service()

    def endpoint_service_mode_on(self, endpoint):
        with self.lock:
            endpoint.service_mode_on()

    def endpoint_service_mode_off(self, endpoint):
        with self.lock:
            endpoint.service_mode_off()

    def run(self):
        log.info("EndpointRemover was started...")
        while self.running:
            for endpoint in self.platforms.wait_for_service:
                self.remove_endpoint(endpoint, try_to_rebuild=True)
            time.sleep(constants.ER_SLEEP_TIME)

    def remove_all(self):
        log.info("Deleting endpoints: {}".format(self.platforms.active_endpoints))
        with self.app_context():
            for endpoint in self.platforms.active_endpoints:
                endpoint.delete(try_to_rebuild=False)

    def stop(self):
        self.running = False
        try:
            self.remove_all()
        finally:
            self.join(1)
            log.info("EndpointRemover was stopped")


class EndpointPreparer(Thread):
    def __init__(self, pool, sessions, artifact_collector, app_context):
        Thread.__init__(self)
        self.running = True
        self.daemon = True
        self.pool = pool
        self.app_context = app_context
        self.sessions = sessions
        self.artifact_collector = artifact_collector
        self.lock = Lock()

    @call_in_thread
    def prepare_endpoint(self, session, get_endpoint_attempts=constants.GET_ENDPOINT_ATTEMPTS):
        with self.app_context():
            with self.lock:
                session.set_status("preparing")

            attempt, wait_time = 0, 2
            while self.running:
                attempt += 1
                wait_time *= 2

                session.refresh()
                if session.closed:
                    log.warning("Attempt {} was aborted because session {} was closed".format(attempt, session.id))
                    break

                if session.endpoint_id:
                    log.warning("Attempt {} was aborted because session {} already have endpoint_id".format(
                        attempt, session.id, session.endpoint_id)
                    )
                    break

                log.info("Try to find endpoint for {}. Attempt {}".format(session, attempt))
                try:
                    _endpoint = self.pool.get_vm(session.platform)
                    if not self.running:
                        if _endpoint:
                            _endpoint.delete()
                        break

                    if self.running and _endpoint:
                        profiler.register_success_get_endpoint(attempt)
                        with self.lock:
                            session.set_endpoint_id(_endpoint.id)
                            log.info("Attempt {} to find endpoint({}) for session {} was succeed".format(
                                attempt, session.endpoint_id, session)
                            )
                        break
                    else:
                        raise CreationException("Got non-ready endpoint or None: {}".format(_endpoint))
                except:
                    log.exception("Attempt {} to get endpoint for session {} was failed".format(
                        attempt, session.id)
                    )
                    if attempt < get_endpoint_attempts:
                        log.debug("Waiting {} seconds before next attempt {} to get endpoint".format(
                            wait_time, attempt + 1)
                        )
                        time.sleep(wait_time)
                    else:
                        profiler.register_fail_get_endpoint()
                        break

            session.refresh()
            if not session.endpoint_id and session.is_preparing:
                with self.lock:
                    session.set_status("waiting")

    def start_screencast(self, session):
        session.set_screencast_started(True)
        session.restore()
        self.artifact_collector.record_screencast(session)

    def _run_tasks(self):
        with self.app_context():
            for session in self.sessions.active():
                session.refresh()<commit_msg>Debug message if returned None from get_vm method<commit_after># coding: utf-8

import time
import logging
from threading import Thread, Lock

from core.exceptions import CreationException, EndpointUnreachableError

from core import constants
from core.utils import call_in_thread
from core.profiler import profiler

log = logging.getLogger(__name__)


class EndpointRemover(Thread):
    def __init__(self, platforms, artifact_collector, database, app_context):
        super(EndpointRemover, self).__init__()
        self.running = True
        self.daemon = True
        self.platforms = platforms
        self.artifact_collector = artifact_collector
        self.database = database
        self.app_context = app_context
        self.lock = Lock()

    @call_in_thread
    def remove_endpoint(self, endpoint, try_to_rebuild=False):
        try:
            with self.app_context():
                self.endpoint_service_mode_on(endpoint)
                session = self.database.get_session_by_endpoint_id(endpoint.id)
                if session:
                    session.restore()
                    self.artifact_collector.save_selenium_log(session)
                    self.artifact_collector.wait_for_complete(session.id)
                endpoint.delete(try_to_rebuild=try_to_rebuild)
                self.endpoint_service_mode_off(endpoint)
        except:
            log.exception("Attempt to remove {} was failed".format(endpoint))
            endpoint.send_to_service()

    def endpoint_service_mode_on(self, endpoint):
        with self.lock:
            endpoint.service_mode_on()

    def endpoint_service_mode_off(self, endpoint):
        with self.lock:
            endpoint.service_mode_off()

    def run(self):
        log.info("EndpointRemover was started...")
        while self.running:
            for endpoint in self.platforms.wait_for_service:
                self.remove_endpoint(endpoint, try_to_rebuild=True)
            time.sleep(constants.ER_SLEEP_TIME)

    def remove_all(self):
        log.info("Deleting endpoints: {}".format(self.platforms.active_endpoints))
        with self.app_context():
            for endpoint in self.platforms.active_endpoints:
                endpoint.delete(try_to_rebuild=False)

    def stop(self):
        self.running = False
        try:
            self.remove_all()
        finally:
            self.join(1)
            log.info("EndpointRemover was stopped")


class EndpointPreparer(Thread):
    def __init__(self, pool, sessions, artifact_collector, app_context):
        Thread.__init__(self)
        self.running = True
        self.daemon = True
        self.pool = pool
        self.app_context = app_context
        self.sessions = sessions
        self.artifact_collector = artifact_collector
        self.lock = Lock()

    @call_in_thread
    def prepare_endpoint(self, session, get_endpoint_attempts=constants.GET_ENDPOINT_ATTEMPTS):
        with self.app_context():
            with self.lock:
                session.set_status("preparing")

            attempt, wait_time = 0, 2
            while self.running:
                attempt += 1
                wait_time *= 2

                session.refresh()
                if session.closed:
                    log.warning("Attempt {} was aborted because session {} was closed".format(attempt, session.id))
                    break

                if session.endpoint_id:
                    log.warning("Attempt {} was aborted because session {} already have endpoint_id".format(
                        attempt, session.id, session.endpoint_id)
                    )
                    break

                log.info("Try to find endpoint for {}. Attempt {}".format(session, attempt))
                try:
                    _endpoint = self.pool.get_vm(session.platform)
                    if not self.running:
                        if _endpoint:
                            _endpoint.delete()
                        break

                    if self.running and _endpoint:
                        profiler.register_success_get_endpoint(attempt)
                        with self.lock:
                            session.set_endpoint_id(_endpoint.id)
                            log.info("Attempt {} to find endpoint({}) for session {} was succeed".format(
                                attempt, session.endpoint_id, session)
                            )
                        break

                    if _endpoint and getattr(_endpoint, "ready", False):
                        raise CreationException("Got non-ready endpoint or None: {}".format(_endpoint))
                    else:
                        raise EndpointUnreachableError("Endpoint wasn't returned, was returned None")
                except EndpointUnreachableError as e:
                    log.debug("Attempt {} to get endpoint for session {} was failed: {}".format(
                        attempt, session.id, e.message)
                    )
                except:
                    log.exception("Attempt {} to get endpoint for session {} was failed".format(
                        attempt, session.id)
                    )

                if attempt < get_endpoint_attempts:
                    log.debug("Waiting {} seconds before next attempt {} to get endpoint".format(
                        wait_time, attempt + 1)
                    )
                    time.sleep(wait_time)
                else:
                    profiler.register_fail_get_endpoint()
                    break

            session.refresh()
            if not session.endpoint_id and session.is_preparing:
                with self.lock:
                    session.set_status("waiting")

    def start_screencast(self, session):
        session.set_screencast_started(True)
        session.restore()
        self.artifact_collector.record_screencast(session)

    def _run_tasks(self):
        with self.app_context():
            for session in self.sessions.active():
                session.refresh()
{'fraction_non_alphanumeric': 0.07725395732966277, 'fraction_numerical': 0.0012044046799724708, 'mean_word_length': 1.6828986845141933, 'pattern_counts': {'":': 0, '<': 5, '<?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': '2533272', 'n_tokens_mistral': 3002, 'n_tokens_neox': 2798, 'n_words': 628}