Tornado adapter: passive exchange.declare scenario where close hangs

[sjlongland]

Updated bug description:

When a channel is closed due to a passive declaration of an exchange that does not exist, processing of the resulting channel closure hangs.

---

Original description (kept so the context of earlier comments is not lost):

Not sure exactly when it started happening, but lately, we've been getting this message, a lot, when an AMQP client tries to passively declare a number of exchanges that are not present.

Correct behaviour is that the AMQP broker closes the channel with a "not found" exception, and our code retries opening the channel. It seems pika trips up on the latter bit, confusing the AMQP broker which then responds by closing the entire connection.

We're using the Tornado client adapter on Python 2.7.

2018-02-05 01:02:34,929 pika.connection[connection.py:1382] WARNING Disconnected from RabbitMQ at rabbitmq:5672 (504): CHANNEL_ERROR - second 'channel.open' seen
2018-02-05 01:02:35,042 pika.adapters.base_connection[base_connection.py: 212] INFO Connecting to 172.18.0.4:5672
2018-02-05 01:02:35,047 pika.credentials[credentials.py:  71] INFO Erasing stored credential values
2018-02-05 01:02:35,763 pika.adapters.base_connection[base_connection.py: 180] WARNING Socket closed when connection was open
2018-02-05 01:02:35,764 pika.adapters.base_connection[base_connection.py: 308] WARNING Connection is closed but not stopping IOLoop
2018-02-05 01:02:35,764 pika.connection[connection.py:1382] WARNING Disconnected from RabbitMQ at rabbitmq:5672 (504): CHANNEL_ERROR - second 'channel.open' seen
2018-02-05 01:02:38,288 pika.adapters.base_connection[base_connection.py: 212] INFO Connecting to 172.18.0.4:5672
2018-02-05 01:02:38,295 pika.credentials[credentials.py:  71] INFO Erasing stored credential values
2018-02-05 01:02:38,995 pika.adapters.base_connection[base_connection.py: 180] WARNING Socket closed when connection was open
2018-02-05 01:02:38,996 pika.adapters.base_connection[base_connection.py: 308] WARNING Connection is closed but not stopping IOLoop
2018-02-05 01:02:38,997 pika.connection[connection.py:1382] WARNING Disconnected from RabbitMQ at rabbitmq:5672 (504): CHANNEL_ERROR - second 'channel.open' seen

I'll also point out that the "Connection is closed but not stopping IOLoop" is annoying; the IOLoop does not belong to pika, so it isn't pika's place to say whether the IOLoop should be running or stopped.

[sjlongland]

Problem seems to have appeared between 0.9.14 and 0.10.0b2. Our code has not changed between those two releases.

[lukebakken]

If you can provide a self-contained code sample and instructions to reproduce, that would help out. I'm not sure what you mean by "passively declare", for instance.

If you find that WARNING to be annoying, please provide a PR to change or remove it.

[sjlongland]

Okay, I have a simplified test case here:

https://gist.github.com/sjlongland/a8acc05e399ec1700d61fe11aba2f0f4

Basically, pypi releases 0.9.14 works on Python 2.7 only (not sure if it's Python 3 compatible, this could be a known issue); 0.10.0 works on both, 0.11.2 fails on both. I'm not seeing the connection getting closed due to double-ups on channel.open like with my other code (which is proprietary sadly, otherwise I'd happily share it), but it shouldn't hang like this either.

In our code base, it could also be a race condition, which is a hard thing to reproduce. This isn't the exact problem I saw in the original report, but on the other hand, when someone says close, they mean it.

[lukebakken]

@sjlongland I can reproduce this issue using your code and only 3 channels. Everything exits as expected when 2 channels are used.

[lukebakken]

@sjlongland could you please edit your original description as it not the issue that your test code demonstrates? Another option would be to open a new GitHub issue for the second 'channel.open' seen issue if you can provide code that reproduces it.

[sjlongland]

Yeah, my other code base actually opens one channel per queue / exchange, an idea I got from QAMQP. It's a device driver which creates an exchange per device… and in the situation where we had the problem, there's nearly 200 devices.

[lukebakken]

This is a tricky one, but I do see some cases with three channels where it runs successfully, and cases where it doesn't, with packet captures for each. If you're interested, I forked your test code here and addressed one bug in it (self._all_channels_opened never reset to False) and _stop_test won't allow itself to runmultple times. Still, I see the issue on occasion.

[lukebakken]

@sjlongland if you'd like to test out my fix in the pika-945 branch, I would appreciate it.

[sjlongland]

On 15/02/18 02:48, Luke Bakken wrote: @sjlongland <https://github.com/sjlongland> if you'd like to test out my fix in the |pika-945| branch <https://github.com/pika/pika/tree/pika-945>, I would appreciate it.

Yep, I'll see if I can get that bundled into a Debian package and try it out on the production instance affected. It'll either work fine, or fail (safely), and in the latter case roll-back is easy; we just install the python-pika package from the official Debian repositories. -- Stuart Longland (aka Redhatter, VK4MSL) I haven't lost my mind... ...it's backed up on a tape somewhere.

[sjlongland]

I'm just trying that branch now… seems I get a 404 message looking for it. I'll try the master branch instead and see how I go.

[sjlongland]

The build breaks:

  File "/usr/lib/python2.7/dist-packages/mm_da/amqp/amqpconnection.py", line 6, in <module>
    from pika.adapters import TornadoConnection
  File "/usr/lib/python2.7/dist-packages/pika/__init__.py", line 9, in <module>
    from pika.connection import ConnectionParameters
  File "/usr/lib/python2.7/dist-packages/pika/connection.py", line 22, in <module>
    from pika import callback as pika_callback
  File "/usr/lib/python2.7/dist-packages/pika/callback.py", line 8, in <module>
    from pika import frame
  File "/usr/lib/python2.7/dist-packages/pika/frame.py", line 7, in <module>
    from pika import spec
  File "/usr/lib/python2.7/dist-packages/pika/spec.py", line 16, in <module>
    from pika import data
  File "/usr/lib/python2.7/dist-packages/pika/data.py", line 10, in <module>
    from pika.compat import PY2, PY3
ImportError: No module named compat

[lukebakken]

That's strange. What commands are you running?

[sjlongland]

Well, the offending line appears to be this:

from pika.adapters import TornadoConnection

It doesn't get a chance to run anything else as it dies right there.

[lukebakken]

@sjlongland since I am not seeing this error, nor do we see it on the build servers, I can't diagnose unless I know exactly what commands you're running in your environment. If you can provide those, great, otherwise there should be a beta release of 1.0.0 this week. Thanks!

[sjlongland]

Okay, we actually build pika out of a forked repository on Github. For this issue, I'm testing against https://github.com/vrtsystems/pika/tree/bugfix/WH-378-close-hang-fix where I have merged the latest master branch.

The diff between this branch; and current pika master is not huge:

$ git log --oneline upstream/master -1
0e13941 (pika/master, origin/upstream/master, upstream/master) Merge pull request #947 from vitaly-krugl/fix-multiple-timers-same-deadline
$ git log --oneline HEAD -3
f127c53 (HEAD -> bugfix/WH-378-close-hang-fix, origin/bugfix/WH-378-close-hang-fix) WH-388: changelog: Version bump to 0.99.0
cc93128 WH-388: Merge branch 'upstream/master' into bugfix/WH-378-close-hang-fix
0e13941 (pika/master, origin/upstream/master, upstream/master) Merge pull request #947 from vitaly-krugl/fix-multiple-timers-same-deadline
$ git diff upstream/master..HEAD
diff --git a/debian/changelog b/debian/changelog
new file mode 100644
index 0000000..aa591e0
--- /dev/null
+++ b/debian/changelog
@@ -0,0 +1,16 @@
+pika (0.99.0) unstable; urgency=low
+  * Pull upstream release 1.0.0b1 to address issues with handling channel.close.
+ -- Stuart Longland <stuartl@vrt.com.au>  Fri, 16 Feb 2018 15:22:01 +1000
+
+pika (0.10.0-1) unstable; urgency=low
+
+  * Weakly reference Connection from Channel objects to prevent garbage
+    collection cycles.
+
+ -- Stuart Longland <stuartl@vrt.com.au>  Wed, 30 Nov 2016 09:20:12 +1000
+
+pika (0.10.0) unstable; urgency=low
+
+  * source package automatically created by stdeb 0.8.5
+
+ -- Stuart Longland <stuartl@vrt.com.au>  Thu, 17 Nov 2016 22:59:17 +0000
diff --git a/debian/compat b/debian/compat
new file mode 100644
index 0000000..ec63514
--- /dev/null
+++ b/debian/compat
@@ -0,0 +1 @@
+9
diff --git a/debian/control b/debian/control
new file mode 100644
index 0000000..aa10310
--- /dev/null
+++ b/debian/control
@@ -0,0 +1,36 @@
+Source: pika
+Maintainer: Gavin M. Roy <gavinmroy@gmail.com>
+Section: python
+Priority: optional
+Build-Depends: python-setuptools (>= 0.6b3), python-all (>= 2.6.6-3), debhelper (>= 7)
+Standards-Version: 3.9.1
+
+
+
+Package: python-pika
+Architecture: all
+Depends: ${misc:Depends}, ${python:Depends}
+Description: Pika Python AMQP Client Library
+ Pika, an AMQP 0-9-1 client library for Python
+ =============================================
+ .
+ |Version| |Downloads| |Status| |Coverage| |License|
+ .
+ Introduction
+ -------------
+ Pika is a pure-Python implementation of the AMQP 0-9-1 protocol that tries
+ to stay fairly independent of the underlying network support library.
+ .
+ - Python 2.6+ and 3.3+ are supported.
+ .
+ - Since threads aren't appropriate to every situation, it doesn't
+   require threads. It takes care not to forbid them, either. The same
+   goes for greenlets, callbacks, continuations and generators. It is
+   not necessarily thread-safe however, and your mileage will vary.
+ .
+ - People may be using direct sockets, plain old `select()`,
+   or any of the wide variety of ways of getting network events to and from a
+   python application. Pika tries to stay compatible with all of these, and to
+
+
+
diff --git a/debian/rules b/debian/rules
new file mode 100755
index 0000000..dbaa2c9
--- /dev/null
+++ b/debian/rules
@@ -0,0 +1,31 @@
+#!/usr/bin/make -f
+
+# This file was automatically generated by stdeb 0.8.5 at
+# Thu, 17 Nov 2016 22:59:17 +0000
+
+%:
+	dh $@ --with python2 --buildsystem=python_distutils
+
+
+override_dh_auto_clean:
+	python setup.py clean -a
+	find . -name \*.pyc -exec rm {} \;
+
+
+
+override_dh_auto_build:
+	python setup.py build --force
+
+
+
+override_dh_auto_install:
+	python setup.py install --force --root=debian/python-pika --no-compile -O0 --install-layout=deb  
+
+
+
+override_dh_python2:
+	dh_python2 --no-guessing-versions
+
+
+
+
diff --git a/debian/source/format b/debian/source/format
new file mode 100644
index 0000000..163aaf8
--- /dev/null
+++ b/debian/source/format
@@ -0,0 +1 @@
+3.0 (quilt)
diff --git a/debian/source/options b/debian/source/options
new file mode 100644
index 0000000..bcc4bbb
--- /dev/null
+++ b/debian/source/options
@@ -0,0 +1 @@
+extend-diff-ignore="\.egg-info$"
\ No newline at end of file
diff --git a/pika/channel.py b/pika/channel.py
index 7fc501a..63ed468 100644
--- a/pika/channel.py
+++ b/pika/channel.py
@@ -8,6 +8,7 @@ implementing the methods and behaviors for an AMQP Channel.
 import collections
 import logging
 import uuid
+import weakref
 
 import pika.frame as frame
 import pika.exceptions as exceptions
@@ -60,8 +61,7 @@ class Channel(object):
         self._validate_rpc_completion_callback(on_open_callback)
 
         self.channel_number = channel_number
-        self.callbacks = connection.callbacks
-        self.connection = connection
+        self._connection = weakref.ref(connection)
 
         # Initially, flow is assumed to be active
         self.flow_active = True
@@ -90,6 +90,27 @@ class Channel(object):
         # opaque cookie value set by wrapper layer (e.g., BlockingConnection)
         # via _set_cookie
         self._cookie = None
+        LOGGER.debug('Creating channel %d', self.channel_number)
+
+    def __repr__(self):
+        return '<%s.%s #%d>' % (
+                self.__class__.__module__,
+                self.__class__.__name__,
+                self.channel_number
+        )
+
+    def __del__(self):
+        LOGGER.debug('Destroying channel %d', self.channel_number)
+
+    @property
+    def connection(self):
+        """Return the connection this channel belongs to."""
+        return self._connection()
+
+    @property
+    def callbacks(self):
+        """Return the callback manager for this channel."""
+        return self.connection.callbacks
 
     def __int__(self):
         """Return the channel object as its channel number
@@ -977,6 +998,7 @@ class Channel(object):
 
     def _cleanup(self):
         """Remove all consumers and any callbacks for the channel."""
+        LOGGER.debug('Cleaning up channel %d', self.channel_number)
         self.callbacks.process(self.channel_number,
                                self._ON_CHANNEL_CLEANUP_CB_KEY, self,
                                self)

The other changes there concern a memory leak we had with pika, in that it wouldn't clean up the channels properly after they were closed.

The above branch gets built as a Debian package dpkg-buildpackage -us -uc -b then the resulting package deployed. When my Python code runs, it runs the following Python statement:

from pika.adapters import TornadoConnection

That's just importing the Pika library, nothing else. That's where it fails.

[lukebakken]

@sjlongland - could you add pika.compat to this line, re-build your package, and let me know if that resolves the issue? Thanks

[sjlongland]

No worries, just doing a build now.

[sjlongland]

Okay, looks like there's now an issue to address regarding the parameters we supply to pika:

2018-02-26 13:52:16,996 mm_da.amqp.amqpconnection.AMQPConnection[amqpconnection.py: 409] ERROR FAILED CONNECTION
Traceback (most recent call last):
  File "/usr/lib/python2.7/dist-packages/mm_da/amqp/amqpconnection.py", line 390, in _on_connect
    backpressure_detection=None)
  File "/usr/lib/python2.7/dist-packages/pika/connection.py", line 647, in __init__
    self.backpressure_detection = backpressure_detection
  File "/usr/lib/python2.7/dist-packages/pika/connection.py", line 209, in backpressure_detection
    'but got %r' % (value,))
TypeError: backpressure_detection must be a bool, but got None

But otherwise, we're over the hump caused by setup.py not including pika.compat. :-)

[sjlongland]

I'll chase the above bug down in the morning, but I think now any further problems are small API changes that needs to be taken into account in my code.

[lukebakken]

Thanks for letting me know, I will address the setup.py issue.

[sjlongland]

Okay, so further testing today (on a dev instance, not production this time), it seems we're not out of the woods:

2018-02-27 00:06:07,627 tornado.application[ioloop.py: 638] ERROR Exception in callback <functools.partial object at 0x7fe3ed84c2b8>
Traceback (most recent call last):
  File "/usr/lib/python2.7/dist-packages/tornado/ioloop.py", line 605, in _run_callback
    ret = callback()
  File "/usr/lib/python2.7/dist-packages/tornado/stack_context.py", line 277, in null_wrapper
    return fn(*args, **kwargs)
  File "/usr/lib/python2.7/dist-packages/mm_da/amqp/amqpexchange.py", line 238, in declare
    self._channel_state_machine.declare_ev()
  File "/usr/lib/python2.7/dist-packages/fysom/__init__.py", line 303, in fn
    self.transition()
  File "/usr/lib/python2.7/dist-packages/fysom/__init__.py", line 298, in _tran
    self._after_event(e)
  File "/usr/lib/python2.7/dist-packages/fysom/__init__.py", line 329, in _after_event
    return getattr(self, fnname)(e)
  File "/usr/lib/python2.7/dist-packages/fysom/__init__.py", line 89, in _callback
    return func(obj, *args, **kwargs)
  File "/usr/lib/python2.7/dist-packages/mm_da/amqp/amqpexchange.py", line 58, in _on_declare
    arguments=self._exchange_args)
TypeError: exchange_declare() got multiple values for keyword argument 'passive'
2018-02-27 00:06:07,628 tornado.application[ioloop.py: 638] ERROR Exception in callback <functools.partial object at 0x7fe3ed84c3c0>
Traceback (most recent call last):
  File "/usr/lib/python2.7/dist-packages/tornado/ioloop.py", line 605, in _run_callback
    ret = callback()
  File "/usr/lib/python2.7/dist-packages/tornado/stack_context.py", line 277, in null_wrapper
    return fn(*args, **kwargs)
  File "/usr/lib/python2.7/dist-packages/mm_da/amqp/amqpqueue.py", line 177, in declare
    self._channel_state_machine.declare_ev()
  File "/usr/lib/python2.7/dist-packages/fysom/__init__.py", line 303, in fn
    self.transition()
  File "/usr/lib/python2.7/dist-packages/fysom/__init__.py", line 298, in _tran
    self._after_event(e)
  File "/usr/lib/python2.7/dist-packages/fysom/__init__.py", line 329, in _after_event
    return getattr(self, fnname)(e)
  File "/usr/lib/python2.7/dist-packages/fysom/__init__.py", line 89, in _callback
    return func(obj, *args, **kwargs)
  File "/usr/lib/python2.7/dist-packages/mm_da/amqp/amqpqueue.py", line 55, in _on_declare
    arguments=self._queue_args)
TypeError: queue_declare() got multiple values for keyword argument 'passive'

It seems declaring queues and exchanges is now broken because Python is confusing arguments.

amqpqueue.py:

    def _on_declare(self, event):
        self._log.debug('Declaring exchange %s', self._name)
        self._channel.exchange_declare(
                # callback
                self._channel_state_machine.declared_ev,
                # exchange name
                self._name,
                # exchange type
                self._type,
                # other args
                passive=self._passive,
                durable=self._durable,
                auto_delete=self._auto_delete,
                internal=self._internal,
                arguments=self._exchange_args)

and amqpexchange.py:

    def _on_declare(self, event):
        self._log.debug('Declaring exchange %s', self._name)
        self._channel.exchange_declare(
                # callback
                self._channel_state_machine.declared_ev,
                # exchange name
                self._name,
                # exchange type
                self._type,
                # other args
                passive=self._passive,
                durable=self._durable,
                auto_delete=self._auto_delete,
                internal=self._internal,
                arguments=self._exchange_args)

I get the same thing if all arguments are keyword arguments (which is what I originally had).

[sjlongland]

Okay, found that one easy enough… callback goes to the end. Final trip-up was no_ack:

  File "/usr/lib/python2.7/dist-packages/mm_da/amqp/amqpqueue.py", line 148, in _on_consume
    **consumer_opts)
TypeError: basic_consume() got an unexpected keyword argument 'no_ack'

It might be nice to have no_ack also be accepted in newer pika with perhaps a warning emitted. I'm thinking something along the lines of:

import warnings
# …
class Channel(…):
    …
    def basic_consume(self, … auto_ack=False, **kwargs):
        if 'no_ack' in kwargs:
            warnings.warn('basic_consume no_ack is now called auto_ack')
            auto_ack = kwargs.pop('no_ack')
        …

Then pika is backward compatible, otherwise this is a breaking change.

[lukebakken]

We had a discussion about no_ack in issue #951 and decided that since we're working on 1.0.0 we would not be concerned about backwards compatibility. If you'd like to chime in, see issue #977.

[lukebakken]

PS @sjlongland thank you for all the testing you have been doing.

[sjlongland]

No problems there… it's a package that my workplace depends rather heavily on, so it's in my interest to get things right. :-)