#celery

in the result backend?

asksol: could the broker connections could leave dangling references in the hub ?

dangling?

hr: if you use redis/memcached yes

yeah, the Channel or whatever closes connection but does not remove it from the hub

hr: but obviously the count is correct what is missing is the results

ionelmc: that would be possible

then there's another operation with the same fd that is something else (cause linux recycles them fds very fast) and gets closed erroneusly

i had this issue before

managed to segfault the interpreter by just mismanaging fds :)

asksol: any hints on where to look for this issue in the broker ?

i was using redis for broker

also, i was thinking, maybe the hub should log some error or raise one if you try to add a fd that's already in the hub - would raising break any existing code ?

not sure, everything is in kombu.transport.redis

i think that would *really* help tracking the source of the leak

you would have to rewrite some parts in case

several things will replace the fd afair

e.g. from 'wait for outq readable' to 'read from outq'

I was considering the same, but matched what other event loops are doing

well ok, lemme ask differently: is there any place now that will re-add a handler on a fd - with a *diffrent* callback ?

also would like to register file objects instead of raw fds

but that is also tricky as I think select returns fds not the original objects

that's annoying ::/

sure

asksol: you can't ... you only get back numbers from epoll or whatever :(

asksol: any suggestion as to where I should look so I stop asking stupid questions ::p

hr: maybe redis evicted them, this happens to me and nothing is logged

asksol: so then, is there any place now that will re-add a handler on a fd - with a *diffrent* callback ?

ionelmc: yes

are there many situations like that?

why would that happen?

can you give me some examples?

I don't remember

any way to fix the timeout (if it would be the problem)

<asksol> e.g. from 'wait for outq readable' to 'read from outq'

there will be many more in the future

asksol: well yeah, but that's a larger change, and btw, not all event loops implement the reactor pattern

tulip does a proactor iirc

this is the only key I see in my redis celery-task-meta-a4cd6f41-e02f-4cb5-8fa7-fd2639cdb4a2

hr: the timeout value is arbitrary, it shouldn't have to wait since the counter means all results should have been written

asksol: can you give me a highlevel explanation of the `consolidation` thing in the Hub ?

is it something so you can have a 'bulk callback` - eg, there's a callback that get's called with all the fds that were ready in one poller poll ?

ionelmc: that just means it will call a callback once for all fds instead of one callback for every fd

it's only used when the pool inqueues are readable

and then you need to filter out what you're interested in in said callback

as we need to choose one to write to

right?

right, but all callbacks that were flagged as 'consolidate'

aah right

it's only there for inqueue scheduling

so there is only one list of 'consolidate fds', you cannot have multiples

so it calls schedule_writes(list_of_inqueue_fds_ready_for_read)

instead of inqueue_ready(fd); inqueue_ready(fd); inqueue_ready(fd)

if did the latter we would have to maintain a set of ready fds and then hub.call_soon(schedule_writes())

asksol: Can you help me with hooking __call__ in Task? https://gist.github.com/avnik/8710313 I try different ways but them not works

which we probably will need to when it's ported to tulip, but it's an optimiziation

btw, we could be proactive like by having a second dispatch in the hub for each fd

asksol: I also created that ticket about functools and autodoc

I

I'll dig in autodoc look at how it gets the signature and try to give better ideas

avn: why does it not work?

hr: from proj.tasks import task; print(task.__doc__)

I'll explain the problem bette

ionelmc: but I guess that would lead to ambiguity as to who will register the fd

or maybe not, the second 'read from outqueue' will already know that the fd is registered

I had a hell of a time debugging that too, I logged every reregister though and worked my way backwards

asksol: proactor pattern doesn't get you much if you need to do buffering to message boundary anyway

problem is when you add too much logging the I/O slows it down and the race conditions disappear :)

and i don't see you using windows IOCP in kombu's hub any time soon

at least not sooner than when you port to tulip :p

right, but if we have a second list of callbacks we could raise an error if the fd is registered twice

asksol: ok, so you've actually tried what i said (about doing that warning on re-add) ??

yeah, in debugging but that will lead too many false positive warnings

asksol: i'm gonna check if the callback is the same

should be less false positives :)

right but moving from 'outqueue readable?' to 'outqueue read' happens for every task

I also checked if the callback.__name__ was the write_job or read from outqueue generators

the generator api needs to change to something convenient, just too bad there is no yield_from

once we have a generator api that makes sense the redis transport could be much simpler too

the yield should be based on an operation, not yielding when the fd is unavailable

that is my next task now as it coincides with the async pyamqp rewrite

asksol: which one exactly?

you've lost me :P

pyamqp will be async, and I cannot manually write the read() and write() operations each time

so the generators used in the event loop now needs to change anyway

so what do you plan to do with the hub ? make it proactor-like?

well, I guess it will be like now but there must be a second layer

so the callback points to a object that controls what happens on a read/write/error event

and the coroutines use data = yield read(size)

the amqp protocol is more like a state machine so cannot unroll that into a single inlined generator :)

asksol: would it be possible that redis.MultiChannelPoller would use the poller instance from the Hub ?

isn't that what it does now?

it does this in the __init__ meth: self.poller = poll()

but it would be possible to patch up that poller instance with the one from the Hub instance

i'm just asking if there's anything like that now

see Transport.register_with_event_loop

god damn it

def on_poll_init(self, poller):

self.poller = poller

it should use the hub directly imho, and create a custom Hub object if used blocking

:)

this will grant direct access to the poller

currently it's just a hack to make it use the hub

and create issues in the hub - cause the hub expects fds in the poll to also be in self.readers/writers

well, i put my money on this as the cause for the fd issues

not all my money tho ... ;p

hmm, that is not unlikely

it should use the hub api directly and create a private Hub instance if there isn't one

would simplify the code a bit too I would think

is it possible that the channel is not closed ?

but it's actually long-dead-and-burried

the redis transport isn't async

it just uses select to know what channel to read from

note that channel here means a redis connection

so every time it needs a message it sends BRPOP + LISTEN operations

when the channel is closed it cancels the operation and remove the fd from the poller

removes*

asksol: incorrect self if decorator used, variant with super(MyTask, self).__call__(...) give regression in tasks with progress updates

ok, but is it possible that the channel is not closed ?

where would I look ?

the channel should never close unless there is a connection failure

you would just add a print statement to wherever you expect the fd to be unregistered

asksol: File "/home/avn/work/sandbox/apu/eggs/celery-3.1.8-py2.7.egg/celery/backends/base.py", line 379, in get_key_for_task

return self.task_keyprefix + self.key_t(task_id)

TypeError: cannot concatenate 'str' and 'NoneType' objects

(I hope pasting three lines not against rules of channel?)

avn: __call__ is a bound method when you try to decorate it, you should use inheritance

asksol: I have this error in case of def __call__(self, *a, **kw): return super(MyTask, self).__call__(*a, **kw)