Initial checkin of media test matrix class, which will be used for media perf test later.

third_party/APScheduler/apscheduler/scheduler.py

+"""
+This module is the main part of the library, and is the only module that
+regular users should be concerned with.
+"""
+from threading import Thread, Event, Lock
+from datetime import datetime, timedelta
+from logging import getLogger
+import os
+
+from apscheduler.util import time_difference, asbool
+from apscheduler.triggers import DateTrigger, IntervalTrigger, CronTrigger
+
+
+logger = getLogger(__name__)
+
+
+class Job(object):
+    """
+    Represents a task scheduled in the scheduler.
+    """
+
+    def __init__(self, trigger, func, args, kwargs):
+        self.thread = None
+        self.trigger = trigger
+        self.func = func
+        self.args = args
+        self.kwargs = kwargs
+        if hasattr(func, '__name__'):
+            self.name = func.__name__
+        else:
+            self.name = str(func)
+
+    def run(self):
+        """
+        Starts the execution of this job in a separate thread.
+        """
+        if (self.thread and self.thread.isAlive()):
+            logger.info('Skipping run of job %s (previously triggered '
+                        'instance is still running)', self)
+        else:
+            self.thread = Thread(target=self.run_in_thread)
+            self.thread.setDaemon(False)
+            self.thread.start()
+
+    def run_in_thread(self):
+        """
+        Runs the associated callable.
+        This method is executed in a dedicated thread.
+        """
+        try:
+            self.func(*self.args, **self.kwargs)
+        except:
+            logger.exception('Error executing job "%s"', self)
+            raise
+
+    def __str__(self):
+        return '%s: %s' % (self.name, repr(self.trigger))
+
+    def __repr__(self):
+        return '%s(%s, %s)' % (self.__class__.__name__, self.name,
+                               repr(self.trigger))
+
+
+class SchedulerShutdownError(Exception):
+    """
+    Thrown when attempting to use the scheduler after
+    it's been shut down.
+    """
+
+    def __init__(self):
+        Exception.__init__(self, 'Scheduler has already been shut down')
+
+
+class SchedulerAlreadyRunningError(Exception):
+    """
+    Thrown when attempting to start the scheduler, but it's already running.
+    """
+
+    def __init__(self):
+        Exception.__init__(self, 'Scheduler is already running')
+
+
+class Scheduler(object):
+    """
+    This class is responsible for scheduling jobs and triggering
+    their execution.
+    """
+
+    stopped = False
+    thread = None
+    misfire_grace_time = 1
+    daemonic = True
+
+    def __init__(self, **config):
+        self.jobs = []
+        self.jobs_lock = Lock()
+        self.wakeup = Event()
+        self.configure(config)
+
+    def configure(self, config):
+        """
+        Updates the configuration with the given options.
+        """
+        for key, val in config.items():
+            if key.startswith('apscheduler.'):
+                key = key[12:]
+            if key == 'misfire_grace_time':
+                self.misfire_grace_time = int(val)
+            elif key == 'daemonic':
+                self.daemonic = asbool(val)
+
+    def start(self):
+        """
+        Starts the scheduler in a new thread.
+        """
+        if self.thread and self.thread.isAlive():
+            raise SchedulerAlreadyRunningError
+        
+        self.stopped = False
+        self.thread = Thread(target=self.run, name='APScheduler')
+        self.thread.setDaemon(self.daemonic)
+        self.thread.start()
+        logger.info('Scheduler started')
+
+    def shutdown(self, timeout=0):
+        """
+        Shuts down the scheduler and terminates the thread.
+        Does not terminate any currently running jobs.
+
+        :param timeout: time (in seconds) to wait for the scheduler thread to
+            terminate, 0 to wait forever, None to skip waiting
+        """
+        if self.stopped or not self.thread.isAlive():
+            return
+
+        logger.info('Scheduler shutting down')
+        self.stopped = True
+        self.wakeup.set()
+        if timeout is not None:
+            self.thread.join(timeout)
+        self.jobs = []
+
+    def cron_schedule(self, year='*', month='*', day='*', week='*',
+                      day_of_week='*', hour='*', minute='*', second='*',
+                      args=None, kwargs=None):
+        """
+        Decorator that causes its host function to be scheduled
+        according to the given parameters.
+        This decorator does not wrap its host function.
+        The scheduled function will be called without any arguments.
+        See :meth:`add_cron_job` for more information.
+        """
+        def inner(func):
+            self.add_cron_job(func, year, month, day, week, day_of_week, hour,
+                              minute, second, args, kwargs)
+            return func
+        return inner
+
+    def interval_schedule(self, weeks=0, days=0, hours=0, minutes=0, seconds=0,
+                          start_date=None, repeat=0, args=None, kwargs=None):
+        """
+        Decorator that causes its host function to be scheduled
+        for execution on specified intervals.
+        This decorator does not wrap its host function.
+        The scheduled function will be called without any arguments.
+        Note that the default repeat value is 0, which means to repeat forever.
+        See :meth:`add_delayed_job` for more information.
+        """
+        def inner(func):
+            self.add_interval_job(func, weeks, days, hours, minutes, seconds,
+                                  start_date, repeat, args, kwargs)
+            return func
+        return inner
+
+    def _add_job(self, trigger, func, args, kwargs):
+        """
+        Adds a Job to the job list and notifies the scheduler thread.
+
+        :param trigger: trigger for the given callable
+        :param args: list of positional arguments to call func with
+        :param kwargs: dict of keyword arguments to call func with
+        :return: the scheduled job
+        :rtype: Job
+        """
+        if self.stopped:
+            raise SchedulerShutdownError
+        if not hasattr(func, '__call__'):
+            raise TypeError('func must be callable')
+
+        if args is None:
+            args = []
+        if kwargs is None:
+            kwargs = {}
+
+        job = Job(trigger, func, args, kwargs)
+        self.jobs_lock.acquire()
+        try:
+            self.jobs.append(job)
+        finally:
+            self.jobs_lock.release()
+        logger.info('Added job "%s"', job)
+
+        # Notify the scheduler about the new job
+        self.wakeup.set()
+
+        return job
+
+    def add_date_job(self, func, date, args=None, kwargs=None):
+        """
+        Adds a job to be completed on a specific date and time.
+
+        :param func: callable to run
+        :param args: positional arguments to call func with
+        :param kwargs: keyword arguments to call func with
+        """
+        trigger = DateTrigger(date)
+        return self._add_job(trigger, func, args, kwargs)
+
+    def add_interval_job(self, func, weeks=0, days=0, hours=0, minutes=0,
+                         seconds=0, start_date=None, repeat=0, args=None,
+                         kwargs=None):
+        """
+        Adds a job to be completed on specified intervals.
+
+        :param func: callable to run
+        :param weeks: number of weeks to wait
+        :param days: number of days to wait
+        :param hours: number of hours to wait
+        :param minutes: number of minutes to wait
+        :param seconds: number of seconds to wait
+        :param start_date: when to first execute the job and start the
+            counter (default is after the given interval)
+        :param repeat: number of times the job will be run (0 = repeat
+            indefinitely)
+        :param args: list of positional arguments to call func with
+        :param kwargs: dict of keyword arguments to call func with
+        """
+        interval = timedelta(weeks=weeks, days=days, hours=hours,
+                             minutes=minutes, seconds=seconds)
+        trigger = IntervalTrigger(interval, repeat, start_date)
+        return self._add_job(trigger, func, args, kwargs)
+
+    def add_cron_job(self, func, year='*', month='*', day='*', week='*',
+                     day_of_week='*', hour='*', minute='*', second='*',
+                     args=None, kwargs=None):
+        """
+        Adds a job to be completed on times that match the given expressions.
+
+        :param func: callable to run
+        :param year: year to run on
+        :param month: month to run on (0 = January)
+        :param day: day of month to run on
+        :param week: week of the year to run on
+        :param day_of_week: weekday to run on (0 = Monday)
+        :param hour: hour to run on
+        :param second: second to run on
+        :param args: list of positional arguments to call func with
+        :param kwargs: dict of keyword arguments to call func with
+        :return: the scheduled job
+        :rtype: Job
+        """
+        trigger = CronTrigger(year=year, month=month, day=day, week=week,
+                              day_of_week=day_of_week, hour=hour,
+                              minute=minute, second=second)
+        return self._add_job(trigger, func, args, kwargs)
+
+    def is_job_active(self, job):
+        """
+        Determines if the given job is still on the job list.
+
+        :return: True if the job is still active, False if not
+        """
+        self.jobs_lock.acquire()
+        try:
+            return job in self.jobs
+        finally:
+            self.jobs_lock.release()
+
+    def unschedule_job(self, job):
+        """
+        Removes a job, preventing it from being fired any more.
+        """
+        self.jobs_lock.acquire()
+        try:
+            self.jobs.remove(job)
+        finally:
+            self.jobs_lock.release()
+        logger.info('Removed job "%s"', job)
+        self.wakeup.set()
+
+    def unschedule_func(self, func):
+        """
+        Removes all jobs that would execute the given function.
+        """
+        self.jobs_lock.acquire()
+        try:
+            remove_list = [job for job in self.jobs if job.func == func]
+            for job in remove_list:
+                self.jobs.remove(job)
+                logger.info('Removed job "%s"', job)
+        finally:
+            self.jobs_lock.release()
+
+        # Have the scheduler calculate a new wakeup time
+        self.wakeup.set()
+
+    def dump_jobs(self):
+        """
+        Gives a textual listing of all jobs currently scheduled on this
+        scheduler.
+
+        :rtype: str
+        """
+        job_strs = []
+        now = datetime.now()
+        self.jobs_lock.acquire()
+        try:
+            for job in self.jobs:
+                next_fire_time = job.trigger.get_next_fire_time(now)
+                job_str = '%s (next fire time: %s)' % (str(job),
+                                                       next_fire_time)
+                job_strs.append(job_str)
+        finally:
+            self.jobs_lock.release()
+
+        if job_strs:
+            return os.linesep.join(job_strs)
+        return 'No jobs currently scheduled.'
+
+    def _get_next_wakeup_time(self, now):
+        """
+        Determines the time of the next job execution, and removes finished
+        jobs.
+
+        :param now: the result of datetime.now(), generated elsewhere for
+            consistency.
+        """
+        next_wakeup = None
+        finished_jobs = []
+
+        self.jobs_lock.acquire()
+        try:
+            for job in self.jobs:
+                next_run = job.trigger.get_next_fire_time(now)
+                if next_run is None:
+                    finished_jobs.append(job)
+                elif next_run and (next_wakeup is None or \
+                                   next_run < next_wakeup):
+                    next_wakeup = next_run
+
+            # Clear out any finished jobs
+            for job in finished_jobs:
+                self.jobs.remove(job)
+                logger.info('Removed finished job "%s"', job)
+        finally:
+            self.jobs_lock.release()
+
+        return next_wakeup
+
+    def _get_current_jobs(self):
+        """
+        Determines which jobs should be executed right now.
+        """
+        current_jobs = []
+        now = datetime.now()
+        start = now - timedelta(seconds=self.misfire_grace_time)
+
+        self.jobs_lock.acquire()
+        try:
+            for job in self.jobs:
+                next_run = job.trigger.get_next_fire_time(start)
+                if next_run:
+                    time_diff = time_difference(now, next_run)
+                    if next_run < now and time_diff <= self.misfire_grace_time:
+                        current_jobs.append(job)
+        finally:
+            self.jobs_lock.release()
+
+        return current_jobs
+
+    def run(self):
+        """
+        Runs the main loop of the scheduler.
+        """
+        self.wakeup.clear()
+        while not self.stopped:
+            # Execute any jobs scheduled to be run right now
+            for job in self._get_current_jobs():
+                logger.debug('Executing job "%s"', job)
+                job.run()
+
+            # Figure out when the next job should be run, and
+            # adjust the wait time accordingly
+            now = datetime.now()
+            next_wakeup_time = self._get_next_wakeup_time(now)
+
+            # Sleep until the next job is scheduled to be run,
+            # or a new job is added, or the scheduler is stopped
+            if next_wakeup_time is not None:
+                wait_seconds = time_difference(next_wakeup_time, now)
+                logger.debug('Next wakeup is due at %s (in %f seconds)',
+                             next_wakeup_time, wait_seconds)
+                self.wakeup.wait(wait_seconds)
+            else:
+                logger.debug('No jobs; waiting until a job is added')
+                self.wakeup.wait()
+            self.wakeup.clear()