Windows: Chains broken with Celery 3.1.16 and Redis broker

[flavianh]

File "[python install]\lib\site-packages\amqp\transport.py", line
95, in init
raise socket.error(last_err)

Yes, it calls a function from amqp even though I set redis as a broker and a backend
-------------- celery@Jack_Dalton v3.1.1 (Cipater)
---- **** -----
--- * *** * -- Windows-8-6.2.9200
-- * - **** ---

• ** ---------- [config]
• ** ---------- .> broker: redis://:**@distant.redis.server:6379
  0
• ** ---------- .> app: my_app:0x2a65a50
• ** ---------- .> concurrency: 2 (prefork)
• *** --- * --- .> events: OFF (enable -E to monitor this worker)
  -- ******* ----
  --- ***** ----- [queues]
  -------------- .> celery exchange=celery(direct) key=celery

I reverted to Celery 3.1.1 and it works fine.

Is 3.1.16 really stable?

[sshinault]

I have a similar issue. I use amqp as the broker and redis as the backend. This works fine on OS X, but when I move the worker to Windows 2012 R2, it blows up (traceback below).

I added some debug to the celery code, and it shows it is trying to hit localhost amqp, instead of the remote machine. The worker has no problem pulling the task chain from the remote amqp initially, so connectivity is not an issue.

If run the worker with --pool=solo command line option, it will work fine. I have seen this (solo pool) as a solution to a number of similar issues (here and on stackoverflow), so something obviously is wrong with multiprocessing (pre-fork) on Windows.

Note the debug I added at the top. It shows:

• no kwargs are passed down to the subprocess
• this means defaults are use for host, so instead of the remote host, the code defaults to '127.0.0.1'
• I'm guessing lack of kwargs may also mean it attempts to use default 'amqp://' instead of redis)

[2014-11-08 09:12:34,451: DEBUG/Worker-1] delay kwarghs = {}
[2014-11-08 09:12:34,467: DEBUG/Worker-1] publish_task kwargs = {}
[2014-11-08 09:12:34,482: DEBUG/Worker-1] getaddrinfo for host=127.0.0.1 port=5672
[2014-11-08 09:12:35,529: DEBUG/Worker-1] getaddrinfo for host=127.0.0.1 port=5672
[2014-11-08 09:12:36,670: DEBUG/Worker-1] getaddrinfo for host=127.0.0.1 port=5672
[2014-11-08 09:12:37,920: DEBUG/Worker-1] getaddrinfo for host=127.0.0.1 port=5672
[2014-11-08 09:12:39,342: DEBUG/Worker-1] getaddrinfo for host=127.0.0.1 port=5672
[2014-11-08 09:12:40,372: WARNING/Worker-1] C:\Python27\lib\site-packages\celery\app\trace.py:364: RuntimeWarning: Excep
tion raised outside body: error(error(10061, 'No connection could be made because the target machine actively refused it
'),):
Traceback (most recent call last):
  File "C:\Python27\lib\site-packages\celery\app\trace.py", line 280, in trace_task
    signature(callbacks[0], app=app).delay(retval)
  File "C:\Python27\lib\site-packages\celery\canvas.py", line 171, in delay
    return self.apply_async(partial_args, partial_kwargs)
  File "C:\Python27\lib\site-packages\celery\canvas.py", line 246, in apply_async
    return _apply(args, kwargs, **options)
  File "C:\Python27\lib\site-packages\celery\app\task.py", line 555, in apply_async
    **dict(self._get_exec_options(), **options)
  File "C:\Python27\lib\site-packages\celery\app\base.py", line 353, in send_task
    reply_to=reply_to or self.oid, **options
  File "C:\Python27\lib\site-packages\celery\app\amqp.py", line 308, in publish_task
    **kwargs
  File "C:\Python27\lib\site-packages\kombu\messaging.py", line 168, in publish
    routing_key, mandatory, immediate, exchange, declare)
  File "C:\Python27\lib\site-packages\kombu\connection.py", line 457, in _ensured
    interval_max)
  File "C:\Python27\lib\site-packages\kombu\connection.py", line 369, in ensure_connection
    interval_start, interval_step, interval_max, callback)
  File "C:\Python27\lib\site-packages\kombu\utils\__init__.py", line 243, in retry_over_time
    return fun(*args, **kwargs)
  File "C:\Python27\lib\site-packages\kombu\connection.py", line 237, in connect
    return self.connection
  File "C:\Python27\lib\site-packages\kombu\connection.py", line 741, in connection
    self._connection = self._establish_connection()
  File "C:\Python27\lib\site-packages\kombu\connection.py", line 696, in _establish_connection
    conn = self.transport.establish_connection()
  File "C:\Python27\lib\site-packages\kombu\transport\pyamqp.py", line 112, in establish_connection
    conn = self.Connection(**opts)
  File "C:\Python27\lib\site-packages\amqp\connection.py", line 165, in __init__
    self.transport = self.Transport(host, connect_timeout, ssl)
  File "C:\Python27\lib\site-packages\amqp\connection.py", line 186, in Transport
    return create_transport(host, connect_timeout, ssl)
  File "C:\Python27\lib\site-packages\amqp\transport.py", line 301, in create_transport
    return TCPTransport(host, connect_timeout)
  File "C:\Python27\lib\site-packages\amqp\transport.py", line 97, in __init__
    raise socket.error(last_err)
error: [Errno 10061] No connection could be made because the target machine actively refused it

[ask]

How do you configure your app?

Pickling an app will only include configuration changes, so the initial configuration must be reconstructed by simply importing the module.

[sshinault]

Thanks, @ask - I moved the config to a module and now it works fine.

[RemiDesgrange]

@ask can you elaborate please ? I still have the bug on win7/8 and celery 3.1.19 with this sample :

# file celeryTest.py
import json
from flask import Flask, Blueprint, abort, jsonify, request, session
from celery import Celery


app = Flask('celeryTest')
app.config.update(
SECRET_KEY = 'not_a_secret',
REDIS_URL = 'redis://localhost',
CELERY_BROKER_URL='redis://localhost',
CELERY_RESULT_BACKEND='redis://localhost',
)

def make_celery(app):
    celery = Celery(app.import_name, broker=app.config['CELERY_BROKER_URL'])
    celery.conf.update(app.config)
    TaskBase = celery.Task
    class ContextTask(TaskBase):
        abstract = True
        def __call__(self, *args, **kwargs):
            with app.app_context():
                return TaskBase.__call__(self, *args, **kwargs)
    celery.Task = ContextTask
    return celery

celery = make_celery(app)

@celery.task
def add(x, y):
    return x + y

@app.route("/")
def hello_world():
    res = add.apply_async((1,2), link=add.s(10))
    return str(res.get())

if __name__ == "__main__":
    app.run(debug=True)

Work perfecly on linux and not on windows, in advance, thanks !!

[ask]

You're using app.conf.update here which means the configuration changes need to be pickled,
it's better to use app.config_from_object('celeryTest.celeryconfig') for a configuration module namedceleryTest/celeryconfig.py, or app.config_from_object('celeryTest:app.conf') to use the flask app config directly.

[ask]

Closing this, as we don't have the resources to support Windows.