#celery

is there an easy function to kill all scheduled tasks?

myusuf3: should it be using more than one?

asksol weill I have 6 conncurrent workers

and it used to use more connections

and looks like its waiting on connection to rabbitmq

myusuf3: one or two connections are used for consuming, and one or more connections (< pool limit) is used for publishing

hmm

asksol it changed since the last release

I used to use more connection

asksol also when I used gevent in production

the workers never return

they get assigned to worker, but never finish

Heh, gevent sounds like trouble if there's non-monkeypatched shared connectioons

so all the workers are busy but they never some back

nanonyme the issue I was talking about above isn't using gevent

Indeed

myusuf3, I just commented to the case you mentioned

nanonyme hmm that might be it

but I have the same setup locally

and it works fine

Well, you don't notice gevent bugs until things start blocking for longer than expected

asksol, are the said connections monkeypatched in Celery, btw?

(ie if gevent is used)

nanonyme well gevent pool

monkey patches everything

hopefully thats not how it is in celery

:)

myusuf3, well, it's about timing here. If you monkeypatch things too late, it came have dramatic outcomes

nanonyme well I am doing it on invocation

at the very beginning

so in the celery process passing -p

I dont think I can get earlier than that

Right

what would it use the connections for? sending subtasks? results? definitely uses more than one connection here with 4 procs

gevent patches celery, yeah

I am doing alot of chaining using |

I dont know just trying to understand why it went from using up 5 connections to using 1

also Substantial drift from celery@16a3fb13-116f-43b4-9be0-1f8dfad8194f may mean clocks are out of sync. Current drift is

2013-05-29T19:31:08.893410+00:00 app[worker.4]: 29.1138370037 seconds (including message overhead).

I am getting that now more recently

asksol ^

so would me setting broker pool limit to 1 account for this

because it wasn't being honoured

previously

Hi guys! Hey, can anyone help me with using "headers" exchanges with Celery 2.5? I can't get it working at all :-(

Is it ridiculous to use a RabbitMQ server with the only worker being itself?

Basically, I'm looking for a queue with a nice interface.

Imo not

nanonyme: is that to me?

Right now I can't see myself expanding to other worker servers, but I guess that might be a nice plus

cmwslw: no, even one worker is useful

myusuf3: a pool limit of 1 means it will use one connection

asksol: Ok, just wondering if I was doing it way wrong

myusuf3: substantial drift is interesting, but not critical. would be interesting with some stats for that

myusuf3: it means that the clocks are out of sync, but also maybe it just received it too late

hard to know the difference ;)

Regarding "headers" exchanges, does anyone know where there's an example of how to use them with Celery? I can't get it to work here, no matter what I set in my CELERY_QUEUES

if it's on the same machine then it's definitely the latter

diogobaeder: no example afaik, I don't remember if .apply_async forwards the headers argument

diogobaeder: you should try setting envvar KOMBU_LOG_DEBUG=1 to see what the kombu calls ends up being

e.g. KOMBU_LOG_DEBUG=1 python; >>> add.apply_async(exchange='hhh', headers={...})

asksol, ah, interesting, thanks. Alternatively, is there any example of setting up the exchanges, queues and bindings manually, by using the AMQP classes directly?

diogobaeder: if headers is not passed then you can add it to extract_exec_options in celery/app/task.py and open an issue

diogobaeder: not sure what you mean by manually, but if you mean using the channel directly then no

asksol, ack, thanks. But how do I define the bindings with the appropriate arguments, in order to go through the "headers" matchers?

Ah, k

you can subclass Queue's .declare method and add that to CELERY_QUEUES

but should be able to use the existing Queue

Exchange has a new passive attribute in kombu master, that could be added to Queue as well

asksol, interesting, but even in 2.5? 2.5 doesn't accept Queue instances in the CELERY_QUEUES config, does it?

diogobaeder: no, not on 2.5 that's true

Btw, CELERY_QUEUES should accept "arguments" items in each queue dict, shouldn't it? It doesn't seem to be passing them to Kombu

I did try using header exchanges with celery 2.x so should be possible

asksol, this is what I'm having: http://paste.ubuntu.com/5714546/

diogobaeder: it's queue_arguments, exchange_arguments and binding_arguments for queue_declare/exchange_declare and queue_bind respectively

(this mess is why it now uses the instances...)

asksol, oh, that makes a lot of difference :-)

asksol, yeah, they're much better, I agree :-)

(And congratz for that change ;-)

someone should document this, it's still in use by the add_consumer control command

asksol, thanks, I'll try with queue_arguments and binding_arguments, as with exchange_arguments it didn't work

asksol, one more question, sorry to bother you with this: how do I use current_app.send_task, to send a message, but specifying only the exchange to send to and the appropriate arguments, but not the specific queue to which it should route the message?

I want the message to be routed according to the headers arguments

If I try specifying the exchange in the args for the method call, all I get is "celery.exceptions.QueueNotFound: "Queue 'celery' is not defined in CELERY_QUEUES""

diogobaeder: the queue= argument makes it take default arguments from the dict in CELERY_QUEUES

so just specify exchange and other options directly and omit the queue

asksol, I'm trying with these args, and it doesn't work: http://paste.ubuntu.com/5714593/

Did I miss something?

In that example, the "node" variable is a dict, so that I can try matching the headers arguments