A Question: What happens when we never close the AMPQ connection?

[cabal-daniel]

Consider the following code

from azure.identity import DefaultAzureCredential
from azure.servicebus import ServiceBusClient, ServiceBusMessage
from datetime import datetime, timedelta

client = ServiceBusClient(
    fully_qualified_namespace='foobar',
    credential=DefaultAzureCredential(),
)

start = datetime.now()
sender_context = client.get_queue_sender(queue_name='hello')
sender = sender_context.__enter__()
checkpoint_1 = datetime.now()
sender.send_messages(ServiceBusMessage('hi'))
checkpoint_2 = datetime.now()
sender = sender_context.__exit__()
checkpoint_3 = datetime.now()

print(f'initiating took {(checkpoint_1 - start) / timedelta(milliseconds=1)}ms')
print(f'sending took {(checkpoint_2 - checkpoint_1) / timedelta(milliseconds=1)}ms')
print(f'closing took {(checkpoint_3 - checkpoint_2) / timedelta(milliseconds=1)}ms')

The results are

initiating took 1255.869ms
sending took 77.747ms
closing took 804.139ms

If a process never calls sender_context.__exit__() because the process gets SIGTERM-ed, would that cause problems?

[github-actions[bot]]

Thank you for your feedback. Tagging and routing to the team member best able to assist.

[kashifkhan]

Hi @cabal-daniel,

our typical usage pattern is to use the context manager ( with ) or simply managing the opening / closing of the client by yourself, so the code ends up like this.

It shouldn't impact anything but you will have sockets left open, other pieces of code that might be running. In general it is a good idea to call close just in case.

from azure.identity import DefaultAzureCredential
from azure.servicebus import ServiceBusClient, ServiceBusMessage
from datetime import datetime, timedelta

client = ServiceBusClient(
    fully_qualified_namespace='foobar',
    credential=DefaultAzureCredential(),
)

start = datetime.now()
sender_context = client.get_queue_sender(queue_name='hello')
checkpoint_1 = datetime.now()
sender.send_messages(ServiceBusMessage('hi')) # this will open up the connection if not open
checkpoint_2 = datetime.now()
sender = sender_context.close()
checkpoint_3 = datetime.now()

[github-actions[bot]]

Hi @cabal-daniel. Thank you for opening this issue and giving us the opportunity to assist. We believe that this has been addressed. If you feel that further discussion is needed, please add a comment with the text "/unresolve" to remove the "issue-addressed" label and continue the conversation.

[cabal-daniel]

@kashifkhan the problem is that opening the connection and closing the connection takes a considerable amount of time

initiating took 1255.869ms
sending took 77.747ms
closing took 804.139ms

For our pods, it seems easier to just keep the connection open and detect if the connection is closed on the other end then re-open so sending messages takes 77ms rather than 3000ms if we're using SB to build a DAG.

[kashifkhan]

@cabal-daniel thank you for that context. There are some things to note here

• Would you be able to try using the exact credential for your environment. DefaultAzureCredential can add some overhead as it tries to determine which credential to use. That itself should save you some time on connects
• An AMQP connection is pretty expensive, so a first connection will always take time. After that it should be quite speedy. The clients are designed to open a connection when a desired operation is called. In the example above, the first send_messages will take some time as it has to open a connection, while subsequent calls will be fast as the connection is already open.
• Calling __exit__ is the same as sender.close(), so there is no short circuit here.
• There is also a keep alive running in the background and if the service drops the connection due to timeouts, the client will reconnect and be ready.
• If the socket dies or something in the client goes wrong, unless it's something the client doesnt handle then it has to be re-initiated.

[cabal-daniel]

@kashifkhan thank you for this helpful information.

Could you comment on the following type of architecture for distributed compute?

1. Start worker
2. Open all the queue senders / AMPQ connections
3. Worker does it's job for a long time
4. SIGTERM
5. Use the atexit library to close all the connections

Between steps 2 to 5, the workers could be alive for days or weeks.

[kashifkhan]

@cabal-daniel

In general the strategy looks fine, just wanted to add a few notes:

• Simply calling get_queue_sender/get_topic_sender wont open the connection. An operation such as send_messages will do that and then subsequent calls will be faster as they will benefit from an open connection.
• Switching from DefaultAzureCredential to the exact credential you are looking for such as ManagedIdentityCredential etc will speed up first connection time.
• As long as there is a consistent flow of messages the clients will work without having to do anything special. Keep alive running is also key
• The clients are designed to run for a long time. It will retry a lot of failures etc. on your behalf. If something goes wrong that it can't handle, an outside process goes wrong, errors that are not retryable etc. etc. then they have to recreated or that scenario has to be handled per the application logic.
• Calling close on the senders/receivers will also close down the connection for you

[cabal-daniel]

Got it -- thanks @kashifkhan ! That makes sense.