pyvoltha/adapters/extensions/omci/tasks/task_runner.py

#
# Copyright 2017 the original author or authors.
#
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
#
#      http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.
#
from __future__ import absolute_import
import structlog
from twisted.internet import reactor
import six


class TaskRunner(object):
    """
    Control the number of running tasks utilizing the OMCI Communications
    channel (OMCI_CC
    """
    def __init__(self, device_id, clock=None):
        self.log = structlog.get_logger(device_id=device_id)
        self._pending_queue = dict()   # task-priority -> [tasks]
        self._running_queue = dict()   # task-id -> task
        self._active = False

        self._successful_tasks = 0
        self._failed_tasks = 0
        self._watchdog_timeouts = 0
        self._last_watchdog_failure_task = ''
        self.reactor = clock if clock is not None else reactor

    def __str__(self):
        return 'TaskRunner: Pending: {}, Running:{}'.format(self.pending_tasks,
                                                            self.running_tasks)

    @property
    def active(self):
        return self._active

    @property
    def pending_tasks(self):
        """
        Get the number of tasks pending to run
        """
        count = 0
        for tasks in six.itervalues(self._pending_queue):
            count += len(tasks)
        return count

    @property
    def running_tasks(self):
        """
        Get the number of tasks currently running
        """
        return len(self._running_queue)

    @property
    def successful_tasks_completed(self):
        return self._successful_tasks

    @property
    def failed_tasks(self):
        return self._failed_tasks

    @property
    def watchdog_timeouts(self):
        return self._watchdog_timeouts

    @property
    def last_watchdog_failure_task(self):
        """ Task name of last tasks to fail due to watchdog"""
        return self._last_watchdog_failure_task

    # TODO: add properties for various stats as needed

    def start(self):
        """
        Start the Task runner
        """
        self.log.debug('starting', active=self._active)

        if not self._active:
            assert len(self._running_queue) == 0, 'Running task queue not empty'
            self._active = True
            self._run_next_task()

    def stop(self):
        """
        Stop the Task runner, first stopping any tasks and flushing the queue
        """
        self.log.debug('stopping', active=self._active)

        if self._active:
            self._active = False

            pq, self._pending_queue = self._pending_queue, dict()
            rq, self._running_queue = self._running_queue, dict()

            # Stop running tasks
            for task in six.itervalues(rq):
                try:
                    task.stop()
                except:
                    pass

            # Kill pending tasks
            for d in six.iterkeys(pq):
                try:
                    d.cancel()
                except:
                    pass

    def _run_next_task(self):
        """
        Search for next task to run, if one can
        :return:
        """
        self.log.debug('run-next', active=self._active,
                       num_running=len(self._running_queue),
                       num_pending=len(self._pending_queue))

        if self._active and len(self._pending_queue) > 0:
            # Cannot run a new task if a running one needs the OMCI_CC exclusively

            if any(task.exclusive for task in six.itervalues(self._running_queue)):
                self.log.debug('exclusive-running')
                return    # An exclusive task is already running

            try:
                priorities = [k for k in six.iterkeys(self._pending_queue)]
                priorities.sort(reverse=True)
                highest_priority = priorities[0] if len(priorities) else None

                if highest_priority is not None:
                    queue = self._pending_queue[highest_priority]
                    next_task = queue[0] if len(queue) else None

                    if next_task is not None:
                        if next_task.exclusive and len(self._running_queue) > 0:
                            self.log.debug('next-is-exclusive', task=str(next_task))
                            return  # Next task to run needs exclusive access

                        queue.pop(0)
                        if len(queue) == 0:
                            del self._pending_queue[highest_priority]

                        self.log.debug('starting-task', task=str(next_task),
                                       running=len(self._running_queue),
                                       pending=len(self._pending_queue))

                        self._running_queue[next_task.task_id] = next_task
                        self.reactor.callLater(0, next_task.start)

                # Run again if others are waiting
                if len(self._pending_queue):
                    self._run_next_task()

            except Exception as e:
                self.log.exception('run-next', e=e)

    def _on_task_success(self, results, task):
        """
        A task completed successfully callback
        :param results: deferred results
        :param task: (Task) The task that succeeded
        :return: deferred results
        """
        self.log.debug('task-success', task_id=str(task),
                       running=len(self._running_queue),
                       pending=len(self._pending_queue))
        try:
            assert task is not None and task.task_id in self._running_queue,\
                'Task not found in running queue'

            task.task_cleanup()
            self._successful_tasks += 1
            del self._running_queue[task.task_id]

        except Exception as e:
            self.log.exception('task-error', task=str(task), e=e)

        finally:
            reactor.callLater(0, self._run_next_task)

        return results

    def _on_task_failure(self, failure, task):
        """
        A task completed with failure callback
        :param failure: (Failure) Failure results
        :param task: (Task) The task that failed
        :return: (Failure) Failure results
        """
        from pyvoltha.adapters.extensions.omci.tasks.task import WatchdogTimeoutFailure

        self.log.debug('task-failure', task_id=str(task),
                       running=len(self._running_queue),
                       pending=len(self._pending_queue))
        try:
            assert task is not None and task.task_id in self._running_queue,\
                'Task not found in running queue'

            task.task_cleanup()
            self._failed_tasks += 1
            del self._running_queue[task.task_id]

            if isinstance(failure.value, WatchdogTimeoutFailure):
                self._watchdog_timeouts += 1
                self._last_watchdog_failure_task = task.name

        except Exception as e:
            # Check the pending queue

            for priority, tasks in six.iteritems(self._pending_queue):
                found = next((t for t in tasks if t.task_id == task.task_id), None)

                if found is not None:
                    self._pending_queue[task.priority].remove(task)
                    if len(self._pending_queue[task.priority]) == 0:
                        del self._pending_queue[task.priority]
                    return failure

            self.log.exception('task-error', task=str(task), e=e)
            raise

        finally:
            reactor.callLater(0, self._run_next_task)

        return failure

    def queue_task(self, task):
        """
        Place a task on the queue to run

        :param task: (Task) task to run
        :return: (deferred) Deferred that will fire on task completion
        """
        self.log.debug('queue-task', active=self._active, task=str(task),
                       running=len(self._running_queue),
                       pending=len(self._pending_queue))

        if task.priority not in self._pending_queue:
            self._pending_queue[task.priority] = []

        task.deferred.addCallbacks(self._on_task_success, self._on_task_failure,
                                   callbackArgs=[task], errbackArgs=[task])

        self._pending_queue[task.priority].append(task)
        self._run_next_task()

        return task.deferred

    def cancel_task(self, task_id):
        """
        Cancel a pending or running task.  The cancel method will be called
        for the task's deferred

        :param task_id: (int) Task identifier
        """
        task = self._running_queue.get(task_id, None)

        if task is not None:
            try:
                task.stop()
            except Exception as e:
                self.log.exception('stop-error', task=str(task), e=e)

            reactor.callLater(0, self._run_next_task)

        else:
            for priority, tasks in six.iteritems(self._pending_queue):
                task = next((t for t in tasks if t.task_id == task_id), None)

                if task is not None:
                    try:
                        task.deferred.cancel()
                    except Exception as e:
                        self.log.exception('cancel-error', task=str(task), e=e)
                    return