celery · avikam · Aug 30, 2020 · Aug 30, 2020 · Aug 31, 2020 · Aug 31, 2020
diff --git a/celery/__init__.py b/celery/__init__.py
@@ -98,59 +98,6 @@ def _find_option_with_arg(argv, short_opts=None, long_opts=None):
     raise KeyError('|'.join(short_opts or [] + long_opts or []))
 
 
-def _patch_eventlet():
-    import eventlet.debug
-
-    eventlet.monkey_patch()
-    blockdetect = float(os.environ.get('EVENTLET_NOBLOCK', 0))
-    if blockdetect:
-        eventlet.debug.hub_blocking_detection(blockdetect, blockdetect)
-
-
-def _patch_gevent():
-    import gevent.monkey
-    import gevent.signal
-
-    gevent.monkey.patch_all()
-    if gevent.version_info[0] == 0:  # pragma: no cover
-        # Signals aren't working in gevent versions <1.0,
-        # and aren't monkey patched by patch_all()
-        import signal
-
-        signal.signal = gevent.signal
-
-
-def maybe_patch_concurrency(argv=None, short_opts=None,
-                            long_opts=None, patches=None):
-    """Apply eventlet/gevent monkeypatches.
-
-    With short and long opt alternatives that specify the command line
-    option to set the pool, this makes sure that anything that needs
-    to be patched is completed as early as possible.
-    (e.g., eventlet/gevent monkey patches).
-    """
-    argv = argv if argv else sys.argv
-    short_opts = short_opts if short_opts else ['-P']
-    long_opts = long_opts if long_opts else ['--pool']
-    patches = patches if patches else {'eventlet': _patch_eventlet,
-                                       'gevent': _patch_gevent}
-    try:
-        pool = _find_option_with_arg(argv, short_opts, long_opts)
-    except KeyError:
-        pass
-    else:
-        try:
-            patcher = patches[pool]
-        except KeyError:
-            pass
-        else:
-            patcher()
-
-        # set up eventlet/gevent environments ASAP
-        from celery import concurrency
-        concurrency.get_implementation(pool)
-
-
 # this just creates a new module, that imports stuff on first attribute
 # access.  This makes the library faster to use.
 old_module, new_module = local.recreate_module(  # pragma: no cover
@@ -174,6 +121,5 @@ def maybe_patch_concurrency(argv=None, short_opts=None,
     VERSION=VERSION, SERIES=SERIES, VERSION_BANNER=VERSION_BANNER,
     version_info_t=version_info_t,
     version_info=version_info,
-    maybe_patch_concurrency=maybe_patch_concurrency,
     _find_option_with_arg=_find_option_with_arg,
 )
diff --git a/celery/__main__.py b/celery/__main__.py
@@ -2,15 +2,12 @@
 
 import sys
 
-# from . import maybe_patch_concurrency
 
 __all__ = ('main',)
 
 
 def main():
     """Entrypoint to the ``celery`` umbrella command."""
-    # if 'multi' not in sys.argv:
-    #     maybe_patch_concurrency()
     from celery.bin.celery import main as _main
     sys.exit(_main())
 

diff --git a/celery/bin/base.py b/celery/bin/base.py
@@ -32,7 +32,7 @@ class CLIContext:
 
     def __init__(self, app, no_color, workdir, quiet=False):
         """Initialize the CLI context."""
-        self.app = app or get_current_app()
+        self._app_factory = app
         self.no_color = no_color
         self.quiet = quiet
         self.workdir = workdir
@@ -46,6 +46,13 @@ def OK(self):
     def ERROR(self):
         return self.style("ERROR", fg="red", bold=True)
 
+    @cached_property
+    def app(self):
+        """Instantiate the app if it wasn't already and return it. This lazy
+        approach gives opportunity for further initializations before the app
+        is imported."""
+        return self._app_factory() or get_current_app()
+
     def style(self, message=None, **kwargs):
         if self.no_color:
             return message

diff --git a/celery/bin/celery.py b/celery/bin/celery.py
@@ -31,10 +31,12 @@ class App(ParamType):
     name = "application"
 
     def convert(self, value, param, ctx):
-        try:
-            return find_app(value)
-        except (ModuleNotFoundError, AttributeError) as e:
-            self.fail(str(e))
+        def factory():
+            try:
+                return find_app(value)
+            except (ModuleNotFoundError, AttributeError) as e:
+                self.fail(str(e))
+        return factory
 
 
 APP = App()
@@ -107,9 +109,9 @@ def celery(ctx, app, broker, result_backend, loader, config, workdir,
     ctx.obj = CLIContext(app=app, no_color=no_color, workdir=workdir, quiet=quiet)
 
     # User options
-    worker.params.extend(ctx.obj.app.user_options.get('worker', []))
-    beat.params.extend(ctx.obj.app.user_options.get('beat', []))
-    events.params.extend(ctx.obj.app.user_options.get('events', []))
+    # worker.params.extend(ctx.obj.app.user_options.get('worker', []))
+    # beat.params.extend(ctx.obj.app.user_options.get('beat', []))
+    # events.params.extend(ctx.obj.app.user_options.get('events', []))
 
 
 @celery.command(cls=CeleryCommand)

diff --git a/celery/bin/worker.py b/celery/bin/worker.py
@@ -17,6 +17,33 @@
 logger = get_logger(__name__)
 
 
+def maybe_patch_concurrency(library):
+    """Patches gevent/eventlet libraries."""
+
+    def patch_eventlet():
+        import eventlet.debug
+
+        eventlet.monkey_patch()
+        blockdetect = float(os.environ.get('EVENTLET_NOBLOCK', 0))
+        if blockdetect:
+            eventlet.debug.hub_blocking_detection(blockdetect, blockdetect)
+
+    def patch_gevent():
+        import gevent.monkey
+        import gevent.signal
+
+        gevent.monkey.patch_all()
+
+    patches = {
+        'eventlet': patch_eventlet,
+        'gevent': patch_gevent
+    }
+
+    patcher = patches.get(library)
+    if patcher:
+        patcher()
+
+
 class CeleryBeat(ParamType):
     """Celery Beat flag."""
 
@@ -42,6 +69,8 @@ def __init__(self):
     def convert(self, value, param, ctx):
         # Pools like eventlet/gevent needs to patch libs as early
         # as possible.
+        maybe_patch_concurrency(value)
+
         return concurrency.get_implementation(
             value) or ctx.obj.app.conf.worker_pool
 
@@ -111,7 +140,10 @@ def detach(path, argv, logfile=None, pidfile=None, uid=None,
 
 
 @click.command(cls=CeleryDaemonCommand,
-               context_settings={'allow_extra_args': True})
+               context_settings={
+                   'allow_extra_args': True,
+                   'ignore_unknown_options': True
+               })
 @click.option('-n',
               '--hostname',
               default=host_format(default_nodename(None)),
@@ -173,7 +205,8 @@ def detach(path, argv, logfile=None, pidfile=None, uid=None,
               type=WORKERS_POOL,
               cls=CeleryOption,
               help_group="Pool Options",
-              help="Pool implementation.")
+              help="Pool implementation.",
+              is_eager=True)
 @click.option('-E',
               '--task-events',
               '--events',
@@ -265,12 +298,15 @@ def detach(path, argv, logfile=None, pidfile=None, uid=None,
 @click.option('--scheduler',
               cls=CeleryOption,
               help_group="Embedded Beat Options")
+@click.argument('user_extra_params', nargs=-1, type=click.UNPROCESSED)
 @click.pass_context
-def worker(ctx, hostname=None, pool_cls=None, app=None, uid=None, gid=None,
+def worker(ctx, hostname=None, pool_cls=None, uid=None, gid=None,
            loglevel=None, logfile=None, pidfile=None, statedb=None,
+           user_extra_params=None,
            **kwargs):
     """Start worker instance.
 
+    \b
     Examples
     --------
     $ celery worker --app=proj -l info
@@ -281,6 +317,23 @@ def worker(ctx, hostname=None, pool_cls=None, app=None, uid=None, gid=None,
 
     """
     app = ctx.obj.app
+
+    user_opts = app.user_options.get('worker')
+    if user_opts:
+        user_options = {}
+
+        def cb(_, param, val):
+            user_options[param.name] = val
+            return val
+
+        for x in user_opts:
+            x.callback = cb
+
+        cmd = click.Command("user", params=list(user_opts))
+        cmd.parse_args(ctx, list(user_extra_params))
+
+        kwargs.update(user_options)
+
     if ctx.args:
         try:
             app.config_from_cmdline(ctx.args, namespace='worker')

diff --git a/t/unit/concurrency/test_eventlet.py b/t/unit/concurrency/test_eventlet.py
@@ -37,17 +37,17 @@ class test_aaa_eventlet_patch(EventletCase):
 
     def test_aaa_is_patched(self):
         with patch('eventlet.monkey_patch', create=True) as monkey_patch:
-            from celery import maybe_patch_concurrency
-            maybe_patch_concurrency(['x', '-P', 'eventlet'])
+            from celery.bin.worker import maybe_patch_concurrency
+            maybe_patch_concurrency('eventlet')
             monkey_patch.assert_called_with()
 
     @patch('eventlet.debug.hub_blocking_detection', create=True)
     @patch('eventlet.monkey_patch', create=True)
     def test_aaa_blockdetecet(
             self, monkey_patch, hub_blocking_detection, patching):
         patching.setenv('EVENTLET_NOBLOCK', '10.3')
-        from celery import maybe_patch_concurrency
-        maybe_patch_concurrency(['x', '-P', 'eventlet'])
+        from celery.bin.worker import maybe_patch_concurrency
+        maybe_patch_concurrency('eventlet')
         monkey_patch.assert_called_with()
         hub_blocking_detection.assert_called_with(10.3, 10.3)
 

diff --git a/t/unit/concurrency/test_gevent.py b/t/unit/concurrency/test_gevent.py
@@ -19,8 +19,8 @@ def test_is_patched(self):
         patch_all = self.patching('gevent.monkey.patch_all')
         import gevent
         gevent.version_info = (1, 0, 0)
-        from celery import maybe_patch_concurrency
-        maybe_patch_concurrency(['x', '-P', 'gevent'])
+        from celery.bin.worker import maybe_patch_concurrency
+        maybe_patch_concurrency('gevent')
         patch_all.assert_called()