Merge pull request #1891 from Exirel/job-prevent-parallel-execution

jobs: prevent parallel execution and plugin specific scheduler

Author: dgw

docs/source/api.rst

@@ -46,6 +46,11 @@ sopel.tools.events
    :members:
    :undoc-members:
 
+sopel.tools.jobs
+----------------
+.. automodule:: sopel.tools.jobs
+   :members:
+
 sopel.formatting
 ----------------
 .. automodule:: sopel.formatting

docs/source/plugin.rst

@@ -146,6 +146,12 @@ sopel.plugins.rules
       :members:
       :undoc-members:
 
+sopel.plugins.jobs
+------------------
+.. automodule:: sopel.plugins.jobs
+   :members:
+   :show-inheritance:
+
 sopel.loader
 ------------
 .. automodule:: sopel.loader

sopel/bot.py

@@ -22,7 +22,7 @@
 from sopel.db import SopelDB
 import sopel.loader
 from sopel.module import NOLIMIT
-from sopel.plugins import rules as plugin_rules
+from sopel.plugins import jobs as plugin_jobs, rules as plugin_rules
 from sopel.tools import deprecated, Identifier
 import sopel.tools.jobs
 from sopel.trigger import Trigger
@@ -61,6 +61,7 @@ def __init__(self, config, daemon=False):
         self._running_triggers_lock = threading.Lock()
         self._plugins = {}
         self._rules_manager = plugin_rules.Manager()
+        self._scheduler = plugin_jobs.Scheduler(self)
 
         self._times = {}
         """
@@ -118,13 +119,16 @@ def __init__(self, config, daemon=False):
         self.shutdown_methods = []
         """List of methods to call on shutdown."""
 
-        self.scheduler = sopel.tools.jobs.JobScheduler(self)
-        """Job Scheduler. See :func:`sopel.module.interval`."""
-
     @property
     def rules(self):
+        """Rules manager."""
         return self._rules_manager
 
+    @property
+    def scheduler(self):
+        """Job Scheduler. See :func:`sopel.module.interval`."""
+        return self._scheduler
+
     @property
     def command_groups(self):
         """A mapping of plugin names to lists of their commands.
@@ -250,7 +254,7 @@ def setup(self):
         """
         self.setup_logging()
         self.setup_plugins()
-        self.scheduler.start()
+        self._scheduler.start()
 
     def setup_logging(self):
         """Set up logging based on config options."""
@@ -414,7 +418,7 @@ def remove_plugin(self, plugin, callables, jobs, shutdowns, urls):
 
         # remove plugin rules, jobs, shutdown functions, and url callbacks
         self._rules_manager.unregister_plugin(name)
-        self.unregister_jobs(jobs)
+        self._scheduler.unregister_plugin(name)
         self.unregister_shutdowns(shutdowns)
         self.unregister_urls(urls)
 
@@ -450,10 +454,14 @@ def get_plugin_meta(self, name):
         version='7.1',
         removed_in='8.0')
     def unregister(self, obj):
-        """Unregister a job or a shutdown method.
+        """Unregister a shutdown method.
 
-        :param obj: the job or shutdown method to unregister
+        :param obj: the shutdown method to unregister
         :type obj: :term:`object`
+
+        This method was used to unregister anything (rules, commands, urls,
+        jobs, and shutdown methods), but since everything can be done by other
+        means, there is no use for it anymore.
         """
         callable_name = getattr(obj, "__name__", 'UNKNOWN')
 
@@ -544,20 +552,12 @@ def register_callables(self, callables):
 
     def register_jobs(self, jobs):
         for func in jobs:
-            for interval in func.interval:
-                job = sopel.tools.jobs.Job(interval, func)
-                self.scheduler.add_job(job)
-                callable_name = getattr(func, "__name__", 'UNKNOWN')
-                LOGGER.debug(
-                    'Job added "%s", will run every %d seconds',
-                    callable_name,
-                    interval)
+            job = sopel.tools.jobs.Job.from_callable(self.settings, func)
+            self._scheduler.register(job)
 
     def unregister_jobs(self, jobs):
         for job in jobs:
-            callable_name = getattr(job, "__name__", 'UNKNOWN')
-            self.scheduler.remove_callable_job(job)
-            LOGGER.debug('Job callable removed: %s', callable_name)
+            self._scheduler.remove_callable_job(job)
 
     def register_shutdowns(self, shutdowns):
         # Append plugin's shutdown function to the bot's list of functions to
@@ -857,8 +857,8 @@ def _update_running_triggers(self, running_triggers):
     def on_scheduler_error(self, scheduler, exc):
         """Called when the Job Scheduler fails.
 
-        :param scheduler: the JobScheduler that errored
-        :type scheduler: :class:`sopel.tools.jobs.JobScheduler`
+        :param scheduler: the job scheduler that errored
+        :type scheduler: :class:`sopel.plugins.jobs.Scheduler`
         :param Exception exc: the raised exception
 
         .. seealso::
@@ -870,8 +870,8 @@ def on_scheduler_error(self, scheduler, exc):
     def on_job_error(self, scheduler, job, exc):
         """Called when a job from the Job Scheduler fails.
 
-        :param scheduler: the JobScheduler responsible for the errored ``job``
-        :type scheduler: :class:`sopel.tools.jobs.JobScheduler`
+        :param scheduler: the job scheduler responsible for the errored ``job``
+        :type scheduler: :class:`sopel.plugins.jobs.Scheduler`
         :param job: the Job that errored
         :type job: :class:`sopel.tools.jobs.Job`
         :param Exception exc: the raised exception
@@ -939,16 +939,16 @@ def _shutdown(self):
         LOGGER.info('Shutting down')
         # Stop Job Scheduler
         LOGGER.info('Stopping the Job Scheduler.')
-        self.scheduler.stop()
+        self._scheduler.stop()
 
         try:
-            self.scheduler.join(timeout=15)
+            self._scheduler.join(timeout=15)
         except RuntimeError:
             LOGGER.exception('Unable to stop the Job Scheduler.')
         else:
             LOGGER.info('Job Scheduler stopped.')
 
-        self.scheduler.clear_jobs()
+        self._scheduler.clear_jobs()
 
         # Shutdown plugins
         LOGGER.info(

sopel/coretasks.py

@@ -27,7 +27,7 @@
 from sopel import loader, module
 from sopel.irc import isupport
 from sopel.irc.utils import CapReq, MyInfo
-from sopel.tools import events, Identifier, iteritems, jobs, target, web
+from sopel.tools import events, Identifier, iteritems, target, web
 
 
 if sys.version_info.major >= 3:
@@ -47,11 +47,14 @@ def setup(bot):
         wait_interval = max(bot.settings.core.throttle_wait, 1)
 
         @module.interval(wait_interval)
+        @module.label('throttle_join')
         def processing_job(bot):
             _join_event_processing(bot)
 
         loader.clean_callable(processing_job, bot.settings)
-        bot.scheduler.add_job(jobs.Job(wait_interval, processing_job))
+        processing_job.plugin_name = 'coretasks'
+
+        bot.register_jobs([processing_job])
 
 
 def shutdown(bot):

sopel/irc/backends.py

@@ -16,7 +16,8 @@
 import sys
 import threading
 
-from sopel.tools.jobs import Job, JobScheduler
+from sopel import loader, module
+from sopel.tools import jobs
 from .abstract_backends import AbstractIRCBackend
 from .utils import get_cnames
 
@@ -40,6 +41,7 @@
 LOGGER = logging.getLogger(__name__)
 
 
+@module.thread(False)
 def _send_ping(backend):
     if not backend.is_connected():
         return
@@ -51,6 +53,7 @@ def _send_ping(backend):
             LOGGER.exception('Socket error on PING')
 
 
+@module.thread(False)
 def _check_timeout(backend):
     if not backend.is_connected():
         return
@@ -61,10 +64,6 @@ def _check_timeout(backend):
         backend.close_when_done()
 
 
-_send_ping.thread = False
-_check_timeout.thread = False
-
-
 class AsynchatBackend(AbstractIRCBackend, asynchat.async_chat):
     """IRC backend implementation using :mod:`asynchat` (:mod:`asyncore`).
 
@@ -85,13 +84,17 @@ def __init__(self, bot, server_timeout=None, ping_timeout=None, **kwargs):
         self.host = None
         self.port = None
         self.source_address = None
+        self.timeout_scheduler = jobs.Scheduler(self)
 
-        ping_job = Job(self.ping_timeout, _send_ping)
-        timeout_job = Job(self.server_timeout, _check_timeout)
+        # prepare interval decorator
+        ping_interval = module.interval(self.ping_timeout)
+        timeout_interval = module.interval(self.server_timeout)
 
-        self.timeout_scheduler = JobScheduler(self)
-        self.timeout_scheduler.add_job(ping_job)
-        self.timeout_scheduler.add_job(timeout_job)
+        # register timeout jobs
+        self.register_timeout_jobs([
+            ping_interval(_send_ping),
+            timeout_interval(_check_timeout),
+        ])
 
     def is_connected(self):
         return self.connected
@@ -109,8 +112,17 @@ def irc_send(self, data):
 
     def run_forever(self):
         """Run forever."""
+        LOGGER.debug('Running forever.')
         asyncore.loop()
 
+    def register_timeout_jobs(self, handlers):
+        """Register the timeout handlers for the timeout scheduler."""
+        for handler in handlers:
+            loader.clean_callable(handler, self.bot.settings)
+            job = jobs.Job.from_callable(self.bot.settings, handler)
+            self.timeout_scheduler.register(job)
+            LOGGER.debug('Timeout Job registered: %s', str(job))
+
     def initiate_connect(self, host, port, source_address):
         """Initiate IRC connection.
 

sopel/plugins/jobs.py

@@ -0,0 +1,111 @@
+# coding=utf-8
+"""Sopel's plugin jobs management.
+
+.. versionadded:: 7.1
+
+.. important::
+
+    This is all fresh and new. Its usage and documentation is for Sopel core
+    development and advanced developers. It is subject to rapid changes
+    between versions without much (or any) warning.
+
+    Do **not** build your plugin based on what is here, you do **not** need to.
+
+"""
+# Copyright 2020, Florian Strzelecki <[email protected]>
+#
+# Licensed under the Eiffel Forum License 2.
+from __future__ import absolute_import, division, print_function, unicode_literals
+
+import itertools
+import logging
+
+from sopel import tools
+from sopel.tools import jobs
+
+LOGGER = logging.getLogger(__name__)
+
+
+class Scheduler(jobs.Scheduler):
+    """Plugins's Job Scheduler
+
+    :param manager: bot instance passed to jobs as argument
+    :type manager: :class:`sopel.bot.Sopel`
+
+    Scheduler that stores plugin jobs and behaves like its
+    :class:`parent class <sopel.tools.jobs.Scheduler>`.
+
+    .. versionadded:: 7.1
+
+    .. note::
+
+        This class is a specific implementation of the scheduler, made to store
+        jobs by their plugins and be used by the bot (its ``manager``).
+        It follows a similar interface as the
+        :class:`plugin rules manager <sopel.plugins.rules.Manager>`.
+
+    .. important::
+
+        This is an internal tool used by Sopel to manage its jobs. To register
+        a job, plugin authors should use :func:`sopel.module.interval`.
+
+    """
+    def __init__(self, manager):
+        super(Scheduler, self).__init__(manager)
+        self._jobs = tools.SopelMemoryWithDefault(list)
+
+    def register(self, job):
+        with self._mutex:
+            self._jobs[job.get_plugin_name()].append(job)
+        LOGGER.debug('Job registered: %s', str(job))
+
+    def unregister_plugin(self, plugin_name):
+        """Unregister all the jobs from a plugin.
+
+        :param str plugin_name: the name of the plugin to remove
+        :return: the number of jobs unregistered for this plugin
+        :rtype: int
+
+        All jobs of that plugin will be removed from the scheduler.
+
+        This method is thread safe. However, it won't cancel or stop any
+        currently running jobs.
+        """
+        unregistered_jobs = 0
+        with self._mutex:
+            jobs_count = len(self._jobs[plugin_name])
+            del self._jobs[plugin_name]
+            unregistered_jobs = unregistered_jobs + jobs_count
+
+        LOGGER.debug(
+            '[%s] Successfully unregistered %d jobs',
+            plugin_name,
+            unregistered_jobs)
+
+        return unregistered_jobs
+
+    def clear_jobs(self):
+        with self._mutex:
+            self._jobs = tools.SopelMemoryWithDefault(list)
+
+        LOGGER.debug('Successfully unregistered all jobs')
+
+    def remove_callable_job(self, callable):
+        plugin_name = getattr(callable, 'plugin_name', None)
+        if not self._jobs[plugin_name]:
+            return
+
+        with self._mutex:
+            self._jobs[plugin_name] = [
+                job for job in self._jobs[plugin_name]
+                if job._handler != callable
+            ]
+
+    def _get_ready_jobs(self, now):
+        with self._mutex:
+            jobs = [
+                job for job in itertools.chain(*self._jobs.values())
+                if job.is_ready_to_run(now)
+            ]
+
+        return jobs