Fix segmentation fault when sending large messages after socket is completely lost

[yunhaoling]

The PR is to address EventHub issue regarding to sending large messages triggering segmentation fault after socket is complete lost: https://github.com/Azure/azure-sdk-for-python/issues/14543, https://github.com/Azure/azure-sdk-for-python/issues/13739

ROOT CAUSE:

• ASYNC_OPERATION_HANDLE result gets freed twice leading to segmentation fault in link_transfer_async in link.c when sending large message composed of multiple transfer frames in socket lost case
  • ASYNC_OPERATION_HANDLE is a wrapper of delivery item, delivery denotes the operation of sending a frame
  • the FIRST time it's freed is when the connection state is changed from OPENED to END due to socket completely lost ("Failure: sending socket failed 10054.")
  • when the connection state changes, the endpoint's state change callback will be called
  • the callback path is connection_set_state -> session->on_connection_state_changed -> link->on_session_state_changed
  • finally, in link.c on_session_state_changed is called to remove all the deliveries when session is in an error status (caused by connection error status).
  • the SECOND time freeing the delivery handle in link_transfer_async triggers the segmentation fault when session_send_transfer results in SESSION_SEND_TRANSFER_ERROR.

HOW TO FIX:

• in Python layer, call connection.dowork() before sending message
  • connection.dowork() could detect socket error status and set session/link status to the correct error status via _set_state chain.
• The caller in Python -- SendClient/MessageSender, will be able to detect error first instead of still trying to send messages on a broken socket connection which leads to the segmentation fault.

---------------------------------------- Advanced Topic ----------------------------------------

WHY SMALL MESSAGE COULD SURVIVE let's understand the send logic first:

• the sending execution path in C module is: link_transfer_async -> session_send_transfer -> connection_encode_frame -> amqp_frame_codec_encode_frame -> frame_codec_encode_frame -> connection::on_bytes_encoded -> xio_send() -> saslclientio_send_async -> tlsio_schannel_send -> socketio_send
  • the bald path will
  • run only one time if the message is composed a single transfer frame
  • loop for payload_size/available_frame_size times to send all the transfer frames that composes the large message
• the FIRST transfer frame encoding would succeed (no matter it's a message of small/big size) even when the socket is lost, because there is no real operation happening on the socket yet, so the errored status of the socket io is not reflected, see frame_codec_encode_frame in frame_codec.c. there is no return value from on_bytes_encoded.
• The socket error happens when real sending takes place and could be reflected after socket send executed.
• And for message composed of a single transfer, the aforementioned path is only executed once, the delivery instance doesn't get a chance to freed twice because after link_transfer_async is called and succeeded, the execution of the program is given back to the python layer -- to connection.do_work(), and the socket error will be detected and raised.
• it's when sending the second transfer frame of a message of large size triggering the segmentation fault (freed twice)

    # in SendClient._client_run, small message won't trigger seg fault
    def _client_run(self):
        # code...
        self._pending_messages = self._filter_pending() # this method will call C module to send messages, small message could pass
        # code...
        self._connection.work()  # connection error raised

---------------------------------------- Code Snippets to Reproduce ----------------------------------------

import uamqp
from uamqp import authentication
from datetime import datetime
import time
import logging
logging.basicConfig(level=logging.INFO, format='%(asctime)s %(name)-12s %(levelname)-8s %(message)s')

live_eventhub_config = {...}

uri = "sb://{}/{}".format(live_eventhub_config['hostname'], live_eventhub_config['event_hub'])
sas_auth = authentication.SASTokenAuth.from_shared_access_key(
    uri, live_eventhub_config['key_name'], live_eventhub_config['access_key'])

target = "amqps://{}/{}".format(live_eventhub_config['hostname'], live_eventhub_config['event_hub'])

send_client = uamqp.SendClient(target, auth=sas_auth, debug=True)
send_client.open()
while not send_client.client_ready():
    send_client.do_work()
print(datetime.now(), "send client is opened")

print(datetime.now(), 'start sleep')
time.sleep(350)
# sleep until the underlying socket io is completely lost
# On windows, the socket io reports "Failure: sending socket failed 10054."
# On linux, the socket io reports "sending socket failed. errno=104 (Connection reset by peer)."
print(datetime.now(), 'end sleep')

# big message will be split into multiple amqp frames which goes into an execution path
# different than a small message (which is composed of just one frame)
# see code: https://github.com/Azure/azure-uamqp-c/blob/master/src/session.c#L1532-L1676
message = uamqp.Message(
    b't'*1024*700
)

# seg fault happens
send_client.send_message(message)

send_client.close()
print(datetime.now(), "send client is closed")

[yunhaoling]

macos would trigger connection ERROR state instead of connection CLOSE on windows/linux (this is due to the different tlsio implementation in c)after socket gets completely lost while our current implementation doesn't handle connection ERROR state properly, so adding one line code change into connect.py to handle ERROR state.