Extension commands must be loaded after concurrency setup (eventlet patching and so on)

celery/bin/base.py

@@ -168,6 +168,7 @@ class Command(object):
         pool_option = self.with_pool_option(argv)
         if pool_option:
             self.maybe_patch_concurrency(argv, *pool_option)
+        self.on_concurrency_setup()
 
         # Dump version and exit if '--version' arg set.
         self.early_version(argv)
@@ -196,6 +197,9 @@ class Command(object):
             # set up eventlet/gevent environments ASAP.
             concurrency.get_implementation(pool)
 
+    def on_concurrency_setup(self):
+        pass
+
     def usage(self, command):
         """Returns the command-line usage string for this app."""
         return '%%prog [options] %s' % (self.args, )

celery/bin/celery.py

@@ -68,7 +68,7 @@ def command(fun, name=None, sortpri=0):
     return fun
 
 
-def get_extension_commands(namespace='celery.commands'):
+def load_extension_commands(namespace='celery.commands'):
     try:
         from pkg_resources import iter_entry_points
     except ImportError:
@@ -83,8 +83,6 @@ def get_extension_commands(namespace='celery.commands'):
             warnings.warn('Cannot load extension %r: %r' % (sym, exc))
         else:
             command(cls, name=ep.name)
-get_extension_commands()
-
 
 class Command(BaseCommand):
     help = ''
@@ -917,6 +915,9 @@ class CeleryCommand(BaseCommand):
             # that may have to be loaded as early as possible.
             return (['-P'], ['--pool'])
 
+    def on_concurrency_setup(self):
+        load_extension_commands()
+
 
 def determine_exit_status(ret):
     if isinstance(ret, int):