"Received a queue.declare-ok response for a mismatching queue (bar instead of ) on channel 1" when channel is shared between threads

[c23omega]

Hello Bunny team,

i faced an issue while trying to use the queue.message_count function.

We use a threaded puma server. In this we store the ruby bunny gem in a instance variable. During initialization we create a bunny connection that we store in a instance variable in a module, then we create to a queue, store this also inside of a instance variable and subscribe to a channel. Later we pick up this instance variable and call the #message_count function. Here is a minimal example that sketches what we are doing:

module MQ
  class Rabbit
    def initialize(name, pw, host, port)
      @conn = Bunny.new("amqp://#{name}:#{pw}@#{host}:#{port}")
      @conn.start
      @chan  = @conn.create_channel
      @exchange = @chan.topic("foo")
    end
    def subscribe(function, queue, channel="#")
      queue_declared = @chan.queue(queue)
      self.class.instance_variable_set(:@queue_declared, queue_declared)
      queue_declared.bind(@exchange, routing_key: "#{channel}").subscribe(manual_ack: true) do |delivery_info, metadata, payload|
        function.call(delivery_info, metadata, payload)
        @chan.ack(delivery_info.delivery_tag)
      end
    end
  end

  class << self
    def message_count
      instance_variable_get(:@queue_declared).message_count
    end
  end
end

Now, our problem is that i sometimes (!) the #message_count function hangs for 30s and we see a timeout the error message:

Traceback (most recent call last):
        2: from (irb):1
        1: from lib/rails_lib/rabbitmq.rb:105:in `message_count'
Timeout::Error (Timeout::Error)

The stdout shows:
Received a queue.declare-ok response for a mismatching queue (bar instead of ) on channel 1 possibly due to a timeout, ignoring it
With wireshark i confirmed that the queue declare request is sent, and the declare ok message is responded fast on the right queue name. It seems that the @pending_queue_declare_name variable gets lost somewhere on the way since this is used in the channel.rb#handle_method (L1666).

In some cases it works without error as it should, and in some it doesn't. My suspicion is that there is a thread safety problem somewhere. If the #meassage_count call is made by the thread that created the connection it works, if there is another thread calling it it fails. So i guess there is somewhere a problem with thread safety here.

What would you recommend as a workaround, or do you have any suggestions how to fix this problem?

[michaelklishin]

You are sharing a channel between threads, which the docs very explicitly tell you not to do. Consider what would happen if you run Bunny::Channel#queue_declare or similar synchronous (has a response) protocol method from two threads. One of them will be overwrite the pending state ivar of the other.

Message count is taken from a Bunny::Channel#queue_declare response. So, don't share channels between threads. How to best
do that with your threaded server, I don't know, but a shared class variable without any synchronisation is not it.

[c23omega]

ok then we have a error of usage here, thanks for the hint! Picking now up on what you said and upon the doc, would the problem be fixed if we create a new Bunny::Channel per active thread?

In general, I'm wondering, where are the threads switched? When i do Bunny::Queue#message_count a queue is declared, and then directly the response is on the same thread returned (Thread.current). Where is the issue there why this gets lost?

[michaelklishin]

Since you cannot share channels between threads, you need one per thread, and open a new one should it be closed because of a channel exception. In most cases either catching exceptions or using Bunny::Channel#open? would be enough. Opening a new channel for every operation would be extremely wasteful and create a very high channel churn scenario.

[michaelklishin]

@c23omega you cannot share an object between threads and assume (without any synchronisation, which Bunny does not do and neither does your code) that their state won't get corrupted. This is concurrent programming 101.

For every declared queue, Bunny::Channel keeps an ivar with the name of the queue declared by the most recent queue.declare. It can be an empty string for server-named queues. When it receives a queue.declare-ok response frame it gets and compares them.
With concurrent access, one thread can overwrite the value of that ivar to the name of a later declared queue.

The threads are "switched" (interleaved) when the runtime decides to do it.

[c23omega]

Hello @michaelklishin,

Thank you very much again for your responses. I was now during the last hours trying different things out. Beginning form using the ruby concurrency package until adding a connection pool for bunny. None of these approaches was fruitful, all led to the threading problem.

I see the problem that the connection is created in one thread, and as we remain in this thread we are fine, but once we use that connection from another it stops to work. But, shouldn't this be solved by encapsulating the declared queue in a TVar (or another of the concurrent_rubys (https://ruby-concurrency.github.io/concurrent-ruby/1.1.10/index.html) classes). ? All my trials failed right now. Is this even an approach that can be fruitful? If so, do you see any issues in my approach?

What i've tried with the connection_pool was

pool = ConnectionPool.new do
  @conn.create_channel
end
self.class.instance_variable_set(:@pool, pool)

since the connection pool is usually a global variable and can be used globally according to the docs. Do you see any issues here?

Do you have any suggestions for me how i can solve that problem?

[michaelklishin]

Opening a channel in every thread and storing it in a thread-local variable is often enough. I'm not familiar with ConnectionPool to tell, I don't know how it would handle exceptions, for example.

[c23omega]

Finally i figured out how to circumvent the problem. This approach works right now quite fine for me:

module MQ
  class Rabbit
    def initialize(name, pw, host, port)
      @conn = Bunny.new("amqp://#{name}:#{pw}@#{host}:#{port}")
      @conn.start
      self.class.instance_variable_set(:@conn, @conn)
      @chan  = @conn.create_channel
      Thread.current[:mq_channel] = @chan
      @exchange = @chan.topic("foo")
    end
    def subscribe(function, queue, channel="#")
      queue_declared = @chan.queue(queue)
      self.class.instance_variable_set(:@queue_declared, queue_declared)
      queue_declared.bind(@exchange, routing_key: "#{channel}").subscribe(manual_ack: true) do |delivery_info, metadata, payload|
        function.call(delivery_info, metadata, payload)
        @chan.ack(delivery_info.delivery_tag)
      end
    end
  end

  class << self
    def message_count(queue_name)
      Thread.current[:mq_channel] = instance_variable_get(:@conn).create_channel unless Thread.current[:mq_channel]
      Thread.current[:mq_channel].queue_declare(queue_name, { passive: true }).message_count
    end
  end
end

First it didn't work smooth. I established the connection via a rails initializer, and then i could not confirm that it works, it always gave me a timeout. After disabling the spring preloader it worked quite fine, so it does now also in the prod environment.

@michaelklishin thank you for your advice!