<commit_before>#!/usr/bin/env python
#
# ==============================
# Copyright 2011 Whamcloud, Inc.
# ==============================

from collections import defaultdict
from dateutil import tz
from django.db import transaction
from chroma_core.lib.job import job_log


class Transition(object):
    def __init__(self, stateful_object, old_state, new_state):
        self.stateful_object = stateful_object
        self.old_state = old_state
        self.new_state = new_state

    def __str__(self):
        return "%s %s->%s" % (self.stateful_object, self.old_state, self.new_state)

    def __eq__(self, other):
        return (isinstance(other, self.__class__)
            and self.__dict__ == other.__dict__)

    def __ne__(self, other):
        return not self.__eq__(other)

    def __hash__(self):
        return ("%s %s %s %s" % (self.stateful_object.__class__, self.stateful_object.id, self.old_state, self.new_state)).__hash__()

    def to_job(self):
        job_klass = self.stateful_object.get_job_class(self.old_state, self.new_state)
        stateful_object_attr = job_klass.stateful_object
        kwargs = {stateful_object_attr: self.stateful_object}
        return job_klass(**kwargs)


class StateManager(object):
    @classmethod
    def available_transitions(cls, stateful_object):
        """Return a list states to which the object can be set from
           its current state, or None if the object is currently
           locked by a Job"""
        if hasattr(stateful_object, 'content_type'):
            stateful_object = stateful_object.downcast()

        # If the object is subject to an incomplete StateChangeJob
        # then don't offer any other transitions.
        from chroma_core.models import StateLock
        from django.db.models import Q

        # We don't advertise transitions for anything which is currently
        # locked by an incomplete job.  We could alternatively advertise
        # which jobs would actually be legal to add by skipping this check and
        # using get_expected_state in place of .state below.
        active_locks = StateLock.filter_by_locked_item(stateful_object).filter(~Q(job__state = 'complete')).count()
        if active_locks > 0:
            return []

        # XXX: could alternatively use expected_state here if you want to advertise
        # what jobs can really be added (i.e. advertise transitions which will
        # be available when current jobs are complete)
        #from_state = self.get_expected_state(stateful_object)
        from_state = stateful_object.state
        available_states = stateful_object.get_available_states(from_state)
        transitions = []
        for to_state in available_states:
            verb = stateful_object.get_verb(from_state, to_state)
            # NB: a None verb means its an internal transition that shouldn't be advertised
            if verb != None:
                transitions.append({"state": to_state, "verb": verb})

        return transitions

    @classmethod
    def _run_opportunistic_jobs(cls, changed_stateful_object = None):
        """Call this when an object state changes, to see if anything
        in the queue of opportunistic jobs is now able to run"""
        from chroma_core.models import OpportunisticJob
        pending_opportunistic_jobs = OpportunisticJob.objects.filter(run = False)
        count = pending_opportunistic_jobs.count()
        if count > 0:
            job_log.info("Opportunistic jobs waiting: %s" % count)

        for oj in pending_opportunistic_jobs:
            import datetime

            job = oj.get_job()

            # Skip running a job if it's a state-change and the object
            # is already in the 'new' state.
            from chroma_core.lib.job import StateChangeJob
            if isinstance(job, StateChangeJob):
                stateful_object = job.get_stateful_object()
                new_state = job.state_transition[2]
                if new_state == stateful_object.state:
                    job_log.info("Opportunistic job %s: skipping (%s already in state %s)" % (
                        oj.pk, stateful_object, new_state))
                    oj.run = True
                    oj.run_at = datetime.datetime.utcnow()
                    oj.save()
                    continue

            # FIXME: should check if some of the job's dependencies are absent, e.g.
            # if you try to set a conf param on an offline MGS, then delete the MGS -- otherwise
            # jobs will linger in the opportunistic queue indefinitely in that (admittedly rare)
            # scenario.
            if job._deps_satisfied():
                job_log.info("Opportunistic job %s (%s) ready to run" % (oj.pk, job.description()))
                StateManager()._add_job(job)
                oj.run = True
                oj.run_at = datetime.datetime.utcnow()
                oj.save()

    def get_expected_state(self, stateful_object_instance):
        try:
            return self.expected_states[stateful_object_instance]
        except KeyError:
            return stateful_object_instance.state

    @classmethod
    def complete_job(cls, job_id):
        from chroma_core.tasks import complete_job
        complete_job.delay(job_id)

    @classmethod
    def _complete_job(cls, job_id):
        from chroma_core.models import Job

        job = Job.objects.get(pk = job_id)
        if job.state == 'completing':
            with transaction.commit_on_success():
                for dependent in job.wait_for_job.all():
                    dependent.notify_wait_for_complete()
                job.state = 'complete'
                job.save()
        else:
            assert job.state == 'complete'
<commit_msg>Update _add_job->add_job<commit_after>#!/usr/bin/env python
#
# ==============================
# Copyright 2011 Whamcloud, Inc.
# ==============================

from collections import defaultdict
from dateutil import tz
from django.db import transaction
from chroma_core.lib.job import job_log


class Transition(object):
    def __init__(self, stateful_object, old_state, new_state):
        self.stateful_object = stateful_object
        self.old_state = old_state
        self.new_state = new_state

    def __str__(self):
        return "%s %s->%s" % (self.stateful_object, self.old_state, self.new_state)

    def __eq__(self, other):
        return (isinstance(other, self.__class__)
            and self.__dict__ == other.__dict__)

    def __ne__(self, other):
        return not self.__eq__(other)

    def __hash__(self):
        return ("%s %s %s %s" % (self.stateful_object.__class__, self.stateful_object.id, self.old_state, self.new_state)).__hash__()

    def to_job(self):
        job_klass = self.stateful_object.get_job_class(self.old_state, self.new_state)
        stateful_object_attr = job_klass.stateful_object
        kwargs = {stateful_object_attr: self.stateful_object}
        return job_klass(**kwargs)


class StateManager(object):
    @classmethod
    def available_transitions(cls, stateful_object):
        """Return a list states to which the object can be set from
           its current state, or None if the object is currently
           locked by a Job"""
        if hasattr(stateful_object, 'content_type'):
            stateful_object = stateful_object.downcast()

        # If the object is subject to an incomplete StateChangeJob
        # then don't offer any other transitions.
        from chroma_core.models import StateLock
        from django.db.models import Q

        # We don't advertise transitions for anything which is currently
        # locked by an incomplete job.  We could alternatively advertise
        # which jobs would actually be legal to add by skipping this check and
        # using get_expected_state in place of .state below.
        active_locks = StateLock.filter_by_locked_item(stateful_object).filter(~Q(job__state = 'complete')).count()
        if active_locks > 0:
            return []

        # XXX: could alternatively use expected_state here if you want to advertise
        # what jobs can really be added (i.e. advertise transitions which will
        # be available when current jobs are complete)
        #from_state = self.get_expected_state(stateful_object)
        from_state = stateful_object.state
        available_states = stateful_object.get_available_states(from_state)
        transitions = []
        for to_state in available_states:
            verb = stateful_object.get_verb(from_state, to_state)
            # NB: a None verb means its an internal transition that shouldn't be advertised
            if verb != None:
                transitions.append({"state": to_state, "verb": verb})

        return transitions

    @classmethod
    def _run_opportunistic_jobs(cls, changed_stateful_object = None):
        """Call this when an object state changes, to see if anything
        in the queue of opportunistic jobs is now able to run"""
        from chroma_core.models import OpportunisticJob
        pending_opportunistic_jobs = OpportunisticJob.objects.filter(run = False)
        count = pending_opportunistic_jobs.count()
        if count > 0:
            job_log.info("Opportunistic jobs waiting: %s" % count)

        for oj in pending_opportunistic_jobs:
            import datetime

            job = oj.get_job()

            # Skip running a job if it's a state-change and the object
            # is already in the 'new' state.
            from chroma_core.lib.job import StateChangeJob
            if isinstance(job, StateChangeJob):
                stateful_object = job.get_stateful_object()
                new_state = job.state_transition[2]
                if new_state == stateful_object.state:
                    job_log.info("Opportunistic job %s: skipping (%s already in state %s)" % (
                        oj.pk, stateful_object, new_state))
                    oj.run = True
                    oj.run_at = datetime.datetime.utcnow()
                    oj.save()
                    continue

            # FIXME: should check if some of the job's dependencies are absent, e.g.
            # if you try to set a conf param on an offline MGS, then delete the MGS -- otherwise
            # jobs will linger in the opportunistic queue indefinitely in that (admittedly rare)
            # scenario.
            if job._deps_satisfied():
                job_log.info("Opportunistic job %s (%s) ready to run" % (oj.pk, job.description()))
                StateManager().add_job(job)
                oj.run = True
                oj.run_at = datetime.datetime.utcnow()
                oj.save()

    def get_expected_state(self, stateful_object_instance):
        try:
            return self.expected_states[stateful_object_instance]
        except KeyError:
            return stateful_object_instance.state

    @classmethod
    def complete_job(cls, job_id):
        from chroma_core.tasks import complete_job
        complete_job.delay(job_id)

    @classmethod
    def _complete_job(cls, job_id):
        from chroma_core.models import Job

        job = Job.objects.get(pk = job_id)
        if job.state == 'completing':
            with transaction.commit_on_success():
                for dependent in job.wait_for_job.all():
                    dependent.notify_wait_for_complete()
                job.state = 'complete'
                job.save()
        else:
            assert job.state == 'complete'

{'fraction_non_alphanumeric': 0.08577405857740586, 'fraction_numerical': 0.0012203626220362621, 'mean_word_length': 2.361558745971286, 'pattern_counts': {'":': 4, '<': 3, '<?xml version=': 0, '>': 10, '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': '4618860', 'n_tokens_mistral': 3216, 'n_tokens_neox': 3051, 'n_words': 917}