[py-amqp 2.0] Race condition on drain_events & channel pending list

[kbespalov]

I'm trying to implement a channel multiplexing over connection with one listening thread which just call the drain_events in while loop. Other threads creates it's own channels and use them.

But some channel actions in the threads (e.g channel opening or publishing confirmation) are stucks on the channel.wait() method, because drain_events() steals a response from the broker and does not dispatch the frame to waiting promise.

As I understand the problem is the race condition between adding the new pending task in the channel.wait() method and calls drain_events in the listening thread. Take a look on the next case:

(1) Thread A: run drain_events() in while loop
(2) Thread B: send(Channel.Open)
(3) Thread A: receive OpenOk and trying to dispatch the frame. But in this time penging list of the channel is empty
(4) Thread B: append the new promise to pending list and goto infinity loop:

while not p.ready:
      self.connection.drain_events(timeout=timeout)

So, I think the problem can be solved if a promise will be added in the list before calling of _frame_writer.send() method.

[ask]

The Connection is not designed to be thread-safe at all, I'd expect there to be more issues even if you fix this one.

It should be thread-safe for greenlets etc, I wonder if this could affect green threads.