Kombu: Test failure with Python 3.6.0

[felixonmars]

The failures are not present in 3.5.0:

=================================== FAILURES ===================================
___________________________ test_Consumer.test_purge ___________________________

self = <t.unit.test_messaging.test_Consumer object at 0xf263c08c>

    def test_purge(self):
        channel = self.connection.channel()
        b1 = Queue('qname1', self.exchange, 'rkey')
        b2 = Queue('qname2', self.exchange, 'rkey')
        b3 = Queue('qname3', self.exchange, 'rkey')
        b4 = Queue('qname4', self.exchange, 'rkey')
>       consumer = Consumer(channel, [b1, b2, b3, b4], auto_declare=True)

t/unit/test_messaging.py:498: 
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ 
kombu/messaging.py:386: in __init__
    self.revive(self.channel)
kombu/messaging.py:407: in revive
    self.declare()
kombu/messaging.py:420: in declare
    queue.declare()
kombu/entity.py:604: in declare
    self._create_exchange(nowait=nowait, channel=channel)
kombu/entity.py:611: in _create_exchange
    self.exchange.declare(nowait=nowait, channel=channel)
kombu/entity.py:182: in declare
    return (channel or self.channel).exchange_declare(
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ 

self = <unbound Exchange foo(direct)>

    @property
    def channel(self):
        """Current channel if the object is bound."""
        channel = self._channel
        if channel is None:
            raise NotBoundError(
                "Can't call method on {0} not bound to a channel".format(
>                   type(self).__name__))
E           kombu.exceptions.NotBoundError: Can't call method on Exchange not bound to a channel

kombu/abstract.py:119: NotBoundError
______________________ test_Consumer.test_multiple_queues ______________________

self = <t.unit.test_messaging.test_Consumer object at 0xf247f88c>

    def test_multiple_queues(self):
        channel = self.connection.channel()
        b1 = Queue('qname1', self.exchange, 'rkey')
        b2 = Queue('qname2', self.exchange, 'rkey')
        b3 = Queue('qname3', self.exchange, 'rkey')
        b4 = Queue('qname4', self.exchange, 'rkey')
>       consumer = Consumer(channel, [b1, b2, b3, b4])

t/unit/test_messaging.py:508: 
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ 
kombu/messaging.py:386: in __init__
    self.revive(self.channel)
kombu/messaging.py:407: in revive
    self.declare()
kombu/messaging.py:420: in declare
    queue.declare()
kombu/entity.py:604: in declare
    self._create_exchange(nowait=nowait, channel=channel)
kombu/entity.py:611: in _create_exchange
    self.exchange.declare(nowait=nowait, channel=channel)
kombu/entity.py:182: in declare
    return (channel or self.channel).exchange_declare(
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ 

self = <unbound Exchange foo(direct)>

    @property
    def channel(self):
        """Current channel if the object is bound."""
        channel = self._channel
        if channel is None:
            raise NotBoundError(
                "Can't call method on {0} not bound to a channel".format(
>                   type(self).__name__))
E           kombu.exceptions.NotBoundError: Can't call method on Exchange not bound to a channel

kombu/abstract.py:119: NotBoundError
============== 2 failed, 834 passed, 303 skipped in 11.37 seconds ==============

[bartekbrak]

Thanks for fixing this, I confirm this bug exists, and it's confusing to debug. We had two machines serving 2 up to three queues each, all fine:

celery worker --app app.celery_app.celery -n default@%h -Q default_computing,default_s3,default
celery worker --app app.celery_app.celery -n backlog@%h -Q backlog_computing,backlog_s3

Moving all five to one worker:

celery worker --app app.celery_app.celery -Q default,default_computing,default_s3,backlog_computing,backlog_s3
...
kombu.exceptions.NotBoundError: Can't call method on Exchange not bound to a channel

We tested on 3.5.3 where the problem ceased to exist.

Can someone explain how this happens? Why 3 (or 4) queues was fine, but not 5, I'm aware 3.6 has ordered dictionaries but this shouldn't matter, should it?

I can also confirm that the connected patch solved the problem.

[dgabriele]

Where are we with this? @auvipy I'm running up against this bug in Python 3.6. Is someone looking into this issue? This is the full kombu traceback when starting a Celery worker:

ile "/Users/yzhao/Dev/celery/lib/python3.6/site-packages/celery/worker/worker.py", line 203, in start
    self.blueprint.start(self)
  File "/Users/yzhao/Dev/celery/lib/python3.6/site-packages/celery/bootsteps.py", line 119, in start
    step.start(parent)
  File "/Users/yzhao/Dev/celery/lib/python3.6/site-packages/celery/bootsteps.py", line 370, in start
    return self.obj.start()
  File "/Users/yzhao/Dev/celery/lib/python3.6/site-packages/celery/worker/consumer/consumer.py", line 318, in start
    blueprint.start(self)
  File "/Users/yzhao/Dev/celery/lib/python3.6/site-packages/celery/bootsteps.py", line 119, in start
    step.start(parent)
  File "/Users/yzhao/Dev/celery/lib/python3.6/site-packages/celery/worker/consumer/tasks.py", line 37, in start
    c.connection, on_decode_error=c.on_decode_error,
  File "/Users/yzhao/Dev/celery/lib/python3.6/site-packages/celery/app/amqp.py", line 302, in TaskConsumer
    **kw
  File "/Users/yzhao/Dev/celery/lib/python3.6/site-packages/kombu/messaging.py", line 386, in __init__
    self.revive(self.channel)
  File "/Users/yzhao/Dev/celery/lib/python3.6/site-packages/kombu/messaging.py", line 407, in revive
    self.declare()
  File "/Users/yzhao/Dev/celery/lib/python3.6/site-packages/kombu/messaging.py", line 420, in declare
    queue.declare()
  File "/Users/yzhao/Dev/celery/lib/python3.6/site-packages/kombu/entity.py", line 604, in declare
    self._create_exchange(nowait=nowait, channel=channel)
  File "/Users/yzhao/Dev/celery/lib/python3.6/site-packages/kombu/entity.py", line 611, in _create_exchange
    self.exchange.declare(nowait=nowait, channel=channel)
  File "/Users/yzhao/Dev/celery/lib/python3.6/site-packages/kombu/entity.py", line 182, in declare
    return (channel or self.channel).exchange_declare(
  File "/Users/yzhao/Dev/celery/lib/python3.6/site-packages/kombu/abstract.py", line 119, in channel
    type(self).__name__))
kombu.exceptions.NotBoundError: Can't call method on Exchange not bound to a channel

[auvipy]

did you tried master?

[beregond]

I had this problem, installation from master helps - I get error though, but different one:

[2017-04-03 10:05:29,061: CRITICAL/MainProcess] Unrecoverable error: TypeError("'NoneType' object is not callable",)
Traceback (most recent call last):
  File "/usr/local/lib/python3.6/site-packages/celery/worker/worker.py", line 203, in start
    self.blueprint.start(self)
  File "/usr/local/lib/python3.6/site-packages/celery/bootsteps.py", line 119, in start
    step.start(parent)
  File "/usr/local/lib/python3.6/site-packages/celery/bootsteps.py", line 370, in start
    return self.obj.start()
  File "/usr/local/lib/python3.6/site-packages/celery/worker/consumer/consumer.py", line 318, in start
    blueprint.start(self)
  File "/usr/local/lib/python3.6/site-packages/celery/bootsteps.py", line 119, in start
    step.start(parent)
  File "/usr/local/lib/python3.6/site-packages/celery/worker/consumer/consumer.py", line 594, in start
    c.loop(*c.loop_args())
  File "/usr/local/lib/python3.6/site-packages/celery/worker/loops.py", line 47, in asynloop
    consumer.consume()
  File "/usr/local/lib/python3.6/site-packages/kombu/messaging.py", line 477, in consume
    self._basic_consume(T, no_ack=no_ack, nowait=False)
  File "/usr/local/lib/python3.6/site-packages/kombu/messaging.py", line 598, in _basic_consume
    no_ack=no_ack, nowait=nowait)
  File "/usr/local/lib/python3.6/site-packages/kombu/entity.py", line 737, in consume
    arguments=self.consumer_arguments)
  File "/usr/local/lib/python3.6/site-packages/amqp/channel.py", line 1578, in basic_consume
    wait=None if nowait else spec.Basic.ConsumeOk,
  File "/usr/local/lib/python3.6/site-packages/amqp/abstract_channel.py", line 73, in send_method
    return self.wait(wait, returns_tuple=returns_tuple)
  File "/usr/local/lib/python3.6/site-packages/amqp/abstract_channel.py", line 93, in wait
    self.connection.drain_events(timeout=timeout)
  File "/usr/local/lib/python3.6/site-packages/amqp/connection.py", line 464, in drain_events
    return self.blocking_read(timeout)
  File "/usr/local/lib/python3.6/site-packages/amqp/connection.py", line 469, in blocking_read
    return self.on_inbound_frame(frame)
  File "/usr/local/lib/python3.6/site-packages/amqp/method_framing.py", line 88, in on_frame
    callback(channel, msg.frame_method, msg.frame_args, msg)
  File "/usr/local/lib/python3.6/site-packages/amqp/connection.py", line 473, in on_inbound_method
    method_sig, payload, content,
  File "/usr/local/lib/python3.6/site-packages/amqp/abstract_channel.py", line 142, in dispatch_method
    listener(*args)
  File "/usr/local/lib/python3.6/site-packages/amqp/channel.py", line 1613, in _on_basic_deliver
    fun(msg)
  File "/usr/local/lib/python3.6/site-packages/kombu/messaging.py", line 624, in _receive_callback
    return on_m(message) if on_m else self.receive(decoded, message)
  File "/usr/local/lib/python3.6/site-packages/celery/worker/consumer/consumer.py", line 568, in on_task_received
    callbacks,
  File "/usr/local/lib/python3.6/site-packages/celery/worker/strategy.py", line 145, in task_message_handler
    handle(req)
  File "/usr/local/lib/python3.6/site-packages/celery/worker/worker.py", line 221, in _process_task_sem
    return self._quick_acquire(self._process_task, req)
  File "/usr/local/lib/python3.6/site-packages/kombu/async/semaphore.py", line 62, in acquire
    callback(*partial_args, **partial_kwargs)
  File "/usr/local/lib/python3.6/site-packages/celery/worker/worker.py", line 226, in _process_task
    req.execute_using_pool(self.pool)
  File "/usr/local/lib/python3.6/site-packages/celery/worker/request.py", line 531, in execute_using_pool
    correlation_id=task_id,
  File "/usr/local/lib/python3.6/site-packages/celery/concurrency/base.py", line 155, in apply_async
    **options)
  File "/usr/local/lib/python3.6/site-packages/billiard/pool.py", line 1487, in apply_async
    self._quick_put((TASK, (result._job, None, func, args, kwds)))
TypeError: 'NoneType' object is not callable

[M1ha-Shvn]

Same errors on Python 3.6.1:
1) kombu.exceptions.NotBoundError: Can't call method on Exchange not bound to a channel installing via pip
2) TypeError: 'NoneType' object is not callable after installing master

[M1ha-Shvn]

Second error is also seen in python 3.5. Downgrading kombu to 4.0.1 makes celery work correctly.

[auvipy]

pr welcome

[tilmanginzel]

Current master (bf820b2) fixes this issue for me. Tested with Python 3.6.0b2. :+1:

[tilmanginzel]

I apologize if my previous answer is only partly correct.

The main issue for me was, that it was impossible to create more than 3 queues. I got the kombu.exceptions.NotBoundError: Can't call method on Exchange not bound to a channel exception. After using the current master, this exception is now gone and it is possible to create more than 3 queues. So it seemed to be working fine.

But, I just realized that after interrupting the worker and performing a 'Warm shutdown', now it is not possible to restart them and continue the work. I'll receive the TypeError: 'NoneType' object is not callable exception, which is also mentioned in this thread. The only workaround is to purge all messages.

I'm not sure if this is another issue, but it definitely worked before using current master with only 1 queue.

Full stacktrace:

[2017-04-16 16:13:28,154: CRITICAL/MainProcess] Unrecoverable error: TypeError("'NoneType' object is not callable",)
Traceback (most recent call last):
  File "/home/vagrant/venv/lib/python3.6/site-packages/celery/worker/worker.py", line 203, in start
    self.blueprint.start(self)
  File "/home/vagrant/venv/lib/python3.6/site-packages/celery/bootsteps.py", line 119, in start
    step.start(parent)
  File "/home/vagrant/venv/lib/python3.6/site-packages/celery/bootsteps.py", line 370, in start
    return self.obj.start()
  File "/home/vagrant/venv/lib/python3.6/site-packages/celery/worker/consumer/consumer.py", line 318, in start
    blueprint.start(self)
  File "/home/vagrant/venv/lib/python3.6/site-packages/celery/bootsteps.py", line 119, in start
    step.start(parent)
  File "/home/vagrant/venv/lib/python3.6/site-packages/celery/worker/consumer/consumer.py", line 594, in start
    c.loop(*c.loop_args())
  File "/home/vagrant/venv/lib/python3.6/site-packages/celery/worker/loops.py", line 47, in asynloop
    consumer.consume()
  File "/home/vagrant/venv/src/kombu-master/kombu/messaging.py", line 477, in consume
    self._basic_consume(T, no_ack=no_ack, nowait=False)
  File "/home/vagrant/venv/src/kombu-master/kombu/messaging.py", line 598, in _basic_consume
    no_ack=no_ack, nowait=nowait)
  File "/home/vagrant/venv/src/kombu-master/kombu/entity.py", line 737, in consume
    arguments=self.consumer_arguments)
  File "/home/vagrant/venv/lib/python3.6/site-packages/amqp/channel.py", line 1578, in basic_consume
    wait=None if nowait else spec.Basic.ConsumeOk,
  File "/home/vagrant/venv/lib/python3.6/site-packages/amqp/abstract_channel.py", line 73, in send_method
    return self.wait(wait, returns_tuple=returns_tuple)
  File "/home/vagrant/venv/lib/python3.6/site-packages/amqp/abstract_channel.py", line 93, in wait
    self.connection.drain_events(timeout=timeout)
  File "/home/vagrant/venv/lib/python3.6/site-packages/amqp/connection.py", line 464, in drain_events
    return self.blocking_read(timeout)
  File "/home/vagrant/venv/lib/python3.6/site-packages/amqp/connection.py", line 469, in blocking_read
    return self.on_inbound_frame(frame)
  File "/home/vagrant/venv/lib/python3.6/site-packages/amqp/method_framing.py", line 88, in on_frame
    callback(channel, msg.frame_method, msg.frame_args, msg)
  File "/home/vagrant/venv/lib/python3.6/site-packages/amqp/connection.py", line 473, in on_inbound_method
    method_sig, payload, content,
  File "/home/vagrant/venv/lib/python3.6/site-packages/amqp/abstract_channel.py", line 142, in dispatch_method
    listener(*args)
  File "/home/vagrant/venv/lib/python3.6/site-packages/amqp/channel.py", line 1613, in _on_basic_deliver
    fun(msg)
  File "/home/vagrant/venv/src/kombu-master/kombu/messaging.py", line 624, in _receive_callback
    return on_m(message) if on_m else self.receive(decoded, message)
  File "/home/vagrant/venv/lib/python3.6/site-packages/celery/worker/consumer/consumer.py", line 568, in on_task_received
    callbacks,
  File "/home/vagrant/venv/lib/python3.6/site-packages/celery/worker/strategy.py", line 145, in task_message_handler
    handle(req)
  File "/home/vagrant/venv/lib/python3.6/site-packages/celery/worker/worker.py", line 221, in _process_task_sem
    return self._quick_acquire(self._process_task, req)
  File "/home/vagrant/venv/src/kombu-master/kombu/async/semaphore.py", line 62, in acquire
    callback(*partial_args, **partial_kwargs)
  File "/home/vagrant/venv/lib/python3.6/site-packages/celery/worker/worker.py", line 226, in _process_task
    req.execute_using_pool(self.pool)
  File "/home/vagrant/venv/lib/python3.6/site-packages/celery/worker/request.py", line 531, in execute_using_pool
    correlation_id=task_id,
  File "/home/vagrant/venv/lib/python3.6/site-packages/celery/concurrency/base.py", line 155, in apply_async
    **options)
  File "/home/vagrant/venv/lib/python3.6/site-packages/billiard/pool.py", line 1487, in apply_async
    self._quick_put((TASK, (result._job, None, func, args, kwds)))
TypeError: 'NoneType' object is not callable

[ChillarAnand]

@beregond @tilmanginzel TypeError: 'NoneType' object is not callable happens because of https://github.com/celery/celery/issues/3620 which is fixed in celery master.

[tilmanginzel]

Great, thank you! Can confirm, that it works. :)

[beregond]

Thanks, waiting for new version then :)

[adamcharnock]

I've just come across this problem too. With three queues this works just fine, with four I get the error. Python 3.6.1, kombu & celery from their respective master branches.

If it makes a difference, I'm configuring my queues dynamically upon startup in Django's AppConfig:

    def _configure_queues(self):
        task_queues = current_app.conf['task_queues'] or []
        current_app.conf.update(
            task_queues=task_queues + self._get_queues()
        )

Stack trace is identical to what @dgabriele experienced above:

[2017-05-31 20:20:09,077: CRITICAL/MainProcess] Unrecoverable error: NotBoundError("Can't call method on Exchange not bound to a channel",)
Traceback (most recent call last):
  File "/Users/adam/Envs/futurepump-warranty/lib/python3.6/site-packages/celery/worker/worker.py", line 203, in start
    self.blueprint.start(self)
  File "/Users/adam/Envs/futurepump-warranty/lib/python3.6/site-packages/celery/bootsteps.py", line 119, in start
    step.start(parent)
  File "/Users/adam/Envs/futurepump-warranty/lib/python3.6/site-packages/celery/bootsteps.py", line 370, in start
    return self.obj.start()
  File "/Users/adam/Envs/futurepump-warranty/lib/python3.6/site-packages/celery/worker/consumer/consumer.py", line 318, in start
    blueprint.start(self)
  File "/Users/adam/Envs/futurepump-warranty/lib/python3.6/site-packages/celery/bootsteps.py", line 119, in start
    step.start(parent)
  File "/Users/adam/Envs/futurepump-warranty/lib/python3.6/site-packages/celery/worker/consumer/tasks.py", line 37, in start
    c.connection, on_decode_error=c.on_decode_error,
  File "/Users/adam/Envs/futurepump-warranty/lib/python3.6/site-packages/celery/app/amqp.py", line 302, in TaskConsumer
    **kw
  File "/Users/adam/Envs/futurepump-warranty/lib/python3.6/site-packages/kombu/messaging.py", line 386, in __init__
    self.revive(self.channel)
  File "/Users/adam/Envs/futurepump-warranty/lib/python3.6/site-packages/kombu/messaging.py", line 407, in revive
    self.declare()
  File "/Users/adam/Envs/futurepump-warranty/lib/python3.6/site-packages/kombu/messaging.py", line 420, in declare
    queue.declare()
  File "/Users/adam/Envs/futurepump-warranty/lib/python3.6/site-packages/kombu/entity.py", line 604, in declare
    self._create_exchange(nowait=nowait, channel=channel)
  File "/Users/adam/Envs/futurepump-warranty/lib/python3.6/site-packages/kombu/entity.py", line 611, in _create_exchange
    self.exchange.declare(nowait=nowait, channel=channel)
  File "/Users/adam/Envs/futurepump-warranty/lib/python3.6/site-packages/kombu/entity.py", line 182, in declare
    return (channel or self.channel).exchange_declare(
  File "/Users/adam/Envs/futurepump-warranty/lib/python3.6/site-packages/kombu/abstract.py", line 119, in channel
    type(self).__name__))
kombu.exceptions.NotBoundError: Can't call method on Exchange not bound to a channel

Update: I've just confirmed that this problem goes away in Python 3.5.3

[simion]

Same issue, using python 3.6.
Tip: this only happens if you have >= 4 queues.
With 3 queues, works like a charm. Wtf? :)

[adamcharnock]

I think this is going to become a larger problem for me soon. Going back to @auvipy's commit which closed this issue, I'm guessing the problem lies somewhere around here:

class Queue(MaybeChannelBound):
    ...
    def revive(self, channel):
        """Revive consumer after connection loss."""
        self._active_tags.clear()
        channel = self.channel = maybe_channel(channel)
        # modify dict size while iterating over it is not allowed
        for qname, queue in list(items(self._queues)):  # <-- list() was added in 
                                                        # commit linked above
            # name may have changed after declare
            self._queues.pop(qname, None)
            queue = self._queues[queue.name] = queue(self.channel)
            queue.revive(channel)  # <-- Binds queue to channel, which
                                   #     should then bind the exchange

        if self.auto_declare:
            self.declare()  # <-- This call ultimately leads to the 
                            #     "Can't call method on Exchange not bound to a channel"

        if self.prefetch_count is not None:
            self.qos(prefetch_count=self.prefetch_count)

My guesses are therefore:

1. list(items(self._queues)) returns an empty iterable
2. queue.revive(channel) does not bind the exchange

Normally I'd say option 1 is unlikely, but as this is a python 3.5/3.6 quirk then I don't want to discount it.

I need to go now, but I may revisit this later. I'm hoping that pinging @auvipy will also highlight the activity on this closed issue.

[codeadict]

This is weird, will be nice to have a release soon with the bugfix instead of using master

[georgepsarakis]

@adamcharnock can you please inspect what the keys values in the self._queues dictionary are before the loop and when declare is called ?

[alukach]

I'm putting together the Kombu release notes for 4.1.0 and am stating that this issue is resolved by #676. Please correct me ASAP if this is incorrect.

[adamcharnock]

Hi @alukach, this seems to currently be resolved for me. I'm running on a particular commit on the master branch from a while ago.

That being said, I don't actually know what fixed the issue. I recall that I seemed to still encounter the problem when I first upgraded to the linked commit, but in the intervening time since the problem seems to have gone away.

Regardless, being able to use an official release rather than a random chosen commit would be great, so 👍 for that. Fingers crossed #676 sorted it.

[thedrow]

@adamcharnock Please try 4.1.0 which is now released and let us know.

[fjsj]

worked for me in Python 3.6.2 after updating to 4.1.0