pika `pop from an empty queue`

Question

I'm using pika in a kubernetes cluster and consuming messages from a queue, which triggers initiating a function in a new thread. However RabbitMQ seems crash, these are the best logs I've found so far:

2020-12-23 10:39:10,906] WARNING - WRITE indicated on fd=9, but writer callback is None; events=0b100 {/usr/local/lib/python3.9/site-packages/pika/adapters/utils/selector_ioloop_adapter.py:393}
(repeats to a total of n=38 times)
2020-12-23 10:39:10,908] ERROR - _AsyncBaseTransport._produce() failed, aborting connection: error=IndexError('pop from an empty deque'); sock=<socket.socket fd=9, family=AddressFamily.AF_INET, type=SocketKind.SOCK_STREAM, proto=6, laddr=('192.168.100.200',  44892), raddr=('192.168.101.201', 5672)>; Caller's stack:                                                                                 
Traceback (most recent call last):                                                                                                 
File "/usr/local/lib/python3.9/site-packages/pika/adapters/utils/io_services_utils.py", line 1097, in _on_socket_writable        
    self._produce()                                                                                                                
File "/usr/local/lib/python3.9/site-packages/pika/adapters/utils/io_services_utils.py", line 822, in _produce                    
    chunk = self._tx_buffers.popleft()                                                                                             
IndexError: pop from an empty deque                                                                                                
{/usr/local/lib/python3.9/site-packages/pika/adapters/utils/io_services_utils.py:1103}                                            
Traceback (most recent call last):                                                                                                 
File "/usr/local/lib/python3.9/site-packages/pika/adapters/utils/io_services_utils.py", line 1097, in _on_socket_writable        
    self._produce()                                                                                                                
File "/usr/local/lib/python3.9/site-packages/pika/adapters/utils/io_services_utils.py", line 822, in _produce                    
    chunk = self._tx_buffers.popleft()                                                                                             
IndexError: pop from an empty deque                                                                                                
2020-12-23 10:39:10,908] INFO - _AsyncTransportBase._initate_abort(): Initiating abrupt asynchronous transport shutdown: state=1; error=IndexError('pop from an empty deque'); <socket.socket fd=9, family=AddressFamily.AF_INET, type=SocketKind.SOCK_STREAM, proto=6, laddr=('192.168.100.200', 44892), raddr=('192.168.101.201', 5672)> {/usr/local/lib/python3.9/site-packages/pika/adapters/utils/io_services_utils.py:904}                                                                                                               
2020-12-23 10:39:10,908] INFO - Deactivating transport: state=1; <socket.socket fd=9, family=AddressFamily.AF_INET, type=SocketKind.SOCK_STREAM, proto=6, laddr=('192.168.100.200', 44892), raddr=('192.168.101.201', 5672)> {/usr/local/lib/python3.9/site-packages/pika/adapters/utils/io_services_utils.py:869}
2020-12-23 10:39:10,909] ERROR - connection_lost: StreamLostError: ("Stream connection lost: IndexError('pop from an empty deque')",) {/usr/local/lib/python3.9/site-packages/pika/adapters/base_connection.py:428}                                                   
2020-12-23 10:39:10,909] INFO - AMQP stack terminated, failed to connect, or aborted: opened=True, error-arg=StreamLostError: ("Stream connection lost: IndexError('pop from an empty deque')",); pending-error=None {/usr/local/lib/python3.9/site-packages/pika/connection.py:1996}
2020-12-23 10:39:10,909] INFO - Stack terminated due to StreamLostError: ("Stream connection lost: IndexError('pop from an empty deque')",) {/usr/local/lib/python3.9/site-packages/pika/connection.py:2065}                                                          
2020-12-23 10:39:10,909] INFO - Closing transport socket and unlinking: state=2; <socket.socket fd=9, family=AddressFamily.AF_INET, type=SocketKind.SOCK_STREAM, proto=6, laddr=('192.168.100.200', 44892), raddr=('192.168.101.201', 5672)> {/usr/local/lib/python3.9/site-packages/pika/adapters/utils/io_services_utils.py:882}      
2020-12-23 10:39:10,909] ERROR - Unexpected connection close detected: StreamLostError: ("Stream connection lost: IndexError('pop from an empty deque')",) {/usr/local/lib/python3.9/site-packages/pika/adapters/blocking_connection.py:520} 
2020-12-23 10:39:31,416] INFO - Pika version 1.1.0 connecting to ('192.168.101.201', 5672) {/usr/local/lib/python3.9/site-packages/pika/adapters/utils/connection_workflow.py:179}                  
2020-12-23 10:39:31,417] INFO - Socket connected: <socket.socket fd=9, family=AddressFamily.AF_INET, type=SocketKind.SOCK_STREAM, proto=6, laddr=('192.168.100.200', 47142), raddr=('192.168.101.201', 5672)> {/usr/local/lib/python3.9/site-packages/pika/adapters/utils/io_services_utils.py:345}                                                                                                         
2020-12-23 10:39:31,418] INFO - Streaming transport linked up: (<pika.adapters.utils.io_services_utils._AsyncPlaintextTransport object at 0x7f81b3099a60>, _StreamingProtocolShim: <SelectConnection PROTOCOL transport=<pika.adapters.utils.io_services_utils._AsyncPlaintextTransport object at 0x7f81b3099a60> params=<ConnectionParameters host=rabbitmq-0.rabbitmq.testing.svc.cluster.local port=5672 virtual_host=/ ssl=False>>). {/usr/local/lib/python3.9/site-packages/pika/adapters/utils/connection_workflow.py:428}
2020-12-23 10:39:31,421] INFO - AMQPConnector - reporting success: <SelectConnection OPEN transport=<pika.adapters.utils.io_services_utils._AsyncPlaintextTransport object at 0x7f81b3099a60> params=<ConnectionParameters host=rabbitmq-0.rabbitmq.testing.svc.cluster.local port=5672 virtual_host=/ ssl=False>> {/usr/local/lib/python3.9/site-packages/pika/adapters/utils/connection_workflow.py:293}
2020-12-23 10:39:31,421] INFO - AMQPConnectionWorkflow - reporting success: <SelectConnection OPEN transport=<pika.adapters.utils.io_services_utils._AsyncPlaintextTransport object at 0x7f81b3099a60> params=<ConnectionParameters host=rabbitmq-0.rabbitmq.testing.svc.cluster.local port=5672 virtual_host=/ ssl=False>> {/usr/local/lib/python3.9/site-packages/pika/adapters/utils/connection_workflow.py:725}                                                                                                                          
2020-12-23 10:39:31,421] INFO - Connection workflow succeeded: <SelectConnection OPEN transport=<pika.adapters.utils.io_services_utils._AsyncPlaintextTransport object at 0x7f81b3099a60> params=<ConnectionParameters host=rabbitmq-0.rabbitmq.testing.svc.cluster.local port=5672 virtual_host=/ ssl=False>> {/usr/local/lib/python3.9/site-packages/pika/adapters/blocking_connection.py:452} 
2020-12-23 10:39:31,422] INFO - Created channel=1 {/usr/local/lib/python3.9/site-packages/pika/adapters/blocking_connection.py:1247
}

My consumer has the following definition:

def publish_message(channel, message):
    channel.basic_publish(exchange='',
                        routing_key='my_queue',
                        body=message)


def connect_to_mq():
    credentials = pika.PlainCredentials(rabbit_user, rabbit_password)
    parameters = pika.ConnectionParameters(rabbit_host, rabbit_port, '/', credentials)
    connection = pika.BlockingConnection(parameters=parameters)
    channel = connection.channel()
    channel.queue_declare(queue='my_queue')
    return connection, channel
    
    
def on_message(channel, method_frame, header_frame, body):
    message = body.decode('utf-8')
    if message == 'do_work':
        thread = threading.Thread(target=start_processing, args=(channel,))
        thread.start()
        publish_message(channel, 'initiated thread')
    
    
def start_processing(channel):
    publish_message(channel, 'starting...')
    time.sleep(240)
    publish_message(channel, 'processing complete!')


def main():
    connection, channel = connect_to_mq()
    channel.basic_consume(queue='my_queue',
                        auto_ack=True,
                        on_message_callback=on_message)

    channel.start_consuming()

Is there anything inherently wrong with my implementation and strategy for handling messages and workloads in separate threads that is causing this to happen?

Answer 1

Pika isn't thread safe by default. You should ideally keep one connection per thread.

There are a bunch of example implementation here, and I have a thread safe rpc example here that you look at as well, but I would recommend using one of their reference implementations for threading.