<commit_before>import platform

import gevent
from gevent.lock import BoundedSemaphore
from sqlalchemy.exc import OperationalError

from inbox.providers import providers
from inbox.config import config
from inbox.contacts.remote_sync import ContactSync
from inbox.events.remote_sync import EventSync, GoogleEventSync
from inbox.heartbeat.status import clear_heartbeat_status
from nylas.logging import get_logger
from nylas.logging.sentry import log_uncaught_errors
from inbox.models.session import session_scope
from inbox.models import Account
from inbox.scheduling.queue import QueueClient
from inbox.util.concurrency import retry_with_logging
from inbox.util.stats import statsd_client

from inbox.mailsync.backends import module_registry

USE_GOOGLE_PUSH_NOTIFICATIONS = \
    'GOOGLE_PUSH_NOTIFICATIONS' in config.get('FEATURE_FLAGS', [])


class SyncService(object):
    """
    Parameters
    ----------
    process_identifier: string
        Unique identifying string for this process (currently
        <hostname>:<process_number>)
    cpu_id: int
        If a system has 4 cores, value from 0-3. (Each sync service on the
        system should get a different value.)
    poll_interval : int
        Seconds between polls for account changes.
    """
    def __init__(self, process_identifier, cpu_id, poll_interval=10):
        self.host = platform.node()
        self.cpu_id = cpu_id
        self.process_identifier = process_identifier
        self.monitor_cls_for = {mod.PROVIDER: getattr(
            mod, mod.SYNC_MONITOR_CLS) for mod in module_registry.values()
            if hasattr(mod, 'SYNC_MONITOR_CLS')}

        for p_name, p in providers.iteritems():
            if p_name not in self.monitor_cls_for:
                self.monitor_cls_for[p_name] = self.monitor_cls_for["generic"]

        self.log = get_logger()
        self.log.bind(cpu_id=cpu_id)
        self.log.info('starting mail sync process',
                      supported_providers=module_registry.keys())

        self.syncing_accounts = set()
        self.email_sync_monitors = {}
        self.contact_sync_monitors = {}
        self.event_sync_monitors = {}
        self.poll_interval = poll_interval
        self.semaphore = BoundedSemaphore(1)

        self.stealing_enabled = config.get('SYNC_STEAL_ACCOUNTS', True)
        self.zone = config.get('ZONE')
        self.queue_client = QueueClient(self.zone)

    def run(self):
        retry_with_logging(self._run_impl, self.log)

    def _run_impl(self):
        """
        Polls for newly registered accounts and checks for start/stop commands.

        """
        while True:
            self.poll()
            gevent.sleep(self.poll_interval)

    def poll(self):
        if self.stealing_enabled:
            r = self.queue_client.claim_next(self.process_identifier)
            if r:
                self.log.info('Claimed new account sync', account_id=r)<commit_msg>Don't take on new processes when CPUs are pegged<commit_after>import platform

import gevent
from gevent.lock import BoundedSemaphore
from sqlalchemy.exc import OperationalError
import psutil

from inbox.providers import providers
from inbox.config import config
from inbox.contacts.remote_sync import ContactSync
from inbox.events.remote_sync import EventSync, GoogleEventSync
from inbox.heartbeat.status import clear_heartbeat_status
from nylas.logging import get_logger
from nylas.logging.sentry import log_uncaught_errors
from inbox.models.session import session_scope
from inbox.models import Account
from inbox.scheduling.queue import QueueClient
from inbox.util.concurrency import retry_with_logging
from inbox.util.stats import statsd_client

from inbox.mailsync.backends import module_registry

USE_GOOGLE_PUSH_NOTIFICATIONS = \
    'GOOGLE_PUSH_NOTIFICATIONS' in config.get('FEATURE_FLAGS', [])


class SyncService(object):
    """
    Parameters
    ----------
    process_identifier: string
        Unique identifying string for this process (currently
        <hostname>:<process_number>)
    cpu_id: int
        If a system has 4 cores, value from 0-3. (Each sync service on the
        system should get a different value.)
    poll_interval : int
        Seconds between polls for account changes.
    """
    def __init__(self, process_identifier, cpu_id, poll_interval=10):
        self.host = platform.node()
        self.cpu_id = cpu_id
        self.process_identifier = process_identifier
        self.monitor_cls_for = {mod.PROVIDER: getattr(
            mod, mod.SYNC_MONITOR_CLS) for mod in module_registry.values()
            if hasattr(mod, 'SYNC_MONITOR_CLS')}

        for p_name, p in providers.iteritems():
            if p_name not in self.monitor_cls_for:
                self.monitor_cls_for[p_name] = self.monitor_cls_for["generic"]

        self.log = get_logger()
        self.log.bind(cpu_id=cpu_id)
        self.log.info('starting mail sync process',
                      supported_providers=module_registry.keys())

        self.syncing_accounts = set()
        self.email_sync_monitors = {}
        self.contact_sync_monitors = {}
        self.event_sync_monitors = {}
        self.poll_interval = poll_interval
        self.semaphore = BoundedSemaphore(1)

        self.stealing_enabled = config.get('SYNC_STEAL_ACCOUNTS', True)
        self.zone = config.get('ZONE')
        self.queue_client = QueueClient(self.zone)

        # We call cpu_percent in a non-blocking way. Because of the way
        # this function works, it'll always return 0.0 the first time
        # we call it. See: https://pythonhosted.org/psutil/#psutil.cpu_percent
        # for more details.
        psutil.cpu_percent(percpu=True)

    def run(self):
        retry_with_logging(self._run_impl, self.log)

    def _run_impl(self):
        """
        Polls for newly registered accounts and checks for start/stop commands.

        """
        while True:
            self.poll()
            gevent.sleep(self.poll_interval)

    def poll(self):
        # We really don't want to take on more load than we can bear, so we need
        # to check the CPU usage before accepting new accounts.
        # Note that we can't check this for the current core because the kernel
        # transparently moves programs across cores.
        usage_per_cpu = psutil.cpu_percent(percpu=True)

        # Conservatively, stop accepting accounts if the CPU usage is over 90%
        # for every core.
        overloaded_cpus = all([cpu_usage > 90.0 for cpu_usage in usage_per_cpu])

        if self.stealing_enabled and not overloaded_cpus:
            r = self.queue_client.claim_next(self.process_identifier)
            if r:
                self.log.info('Claimed new account sync', account_id=r)
{'fraction_non_alphanumeric': 0.07747211895910781, 'fraction_numerical': 0.002825278810408922, 'mean_word_length': 2.9036564132327336, 'pattern_counts': {'":': 0, '<': 7, '<?xml version=': 0, '>': 8, 'https://': 1, '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': '740627', 'n_tokens_mistral': 2003, 'n_tokens_neox': 1919, 'n_words': 516}