WS11: Make the Event Hubs clients thread safe

[j7nw4r]

Summary

The Event Hubs C++ clients are not thread safe. The C++ guidelines make thread safety mandatory for a service client. Three data races are in the producer send path. No public header gives a thread safety guarantee. The AMQP layer below the clients does no serialization.

The .NET library tells callers to share one client across threads. A C++ caller who does the same gets undefined behavior today.

This workstream makes every public member of the clients thread safe. #7255 (WS3) covers the different lifecycle races in ProducerClient::Close and Processor::m_isRunning. The two workstreams do not overlap. Both must close before GA.

Severity tags follow the key in #7252. The code anchors refer to commit 2de7c583e. Line numbers can move. The symbol names are the stable reference.

Motivation

The requirement is mandatory and specific to C++. docs/cpp/introduction.md line 335 has the id cpp-design-client-methods-thread-safety. It says that all public members of the client type must be safe to call from more than one thread at the same time. Line 86 has the id cpp-service-client-immutable. It also makes the service client classes thread safe.

The requirement applies to each member. A design that makes only Send thread safe does not obey it. A design that tells the caller to serialize does not obey it. The language-independent guidelines say nothing about client thread safety. Quote the C++ page.

The .NET library is the reference bar. It already has this contract. Its README says that all client instance methods are thread safe and independent of each other. It also says that each client type is safe to cache and to use as a singleton for the life of the application.

The C++ library says nothing. A search of sdk/eventhubs for "thread safe", "thread-safe" and "thread safety" finds one file. That file is a third-party header in the stress test (test/eventhubs-stress-test/src/inc/argagg.hpp:612).

Three defects stop the guarantee today.

ProducerClient::GetSender reads the sender map with no lock (src/producer_client.cpp:259-263). EnsureSender changes the same std::map under m_sendersLock (src/producer_client.cpp:225-257). Two threads that send to different partitions for the first time cause a data race. One thread reads the map. The other thread writes it. Send calls GetSender at src/producer_client.cpp:111. CreateBatch calls it at src/producer_client.cpp:90. Every send uses this path. GetSession takes m_sessionsLock for the same pattern, so GetSender looks like an error.

EnsureSender keeps m_sendersLock across GetSession(...).CreateMessageSender(...) and sender.Open(context) (src/producer_client.cpp:229-257). Open sends data to the service and waits for the answer. The first send to one partition must wait for the link setup of every other partition.

EventDataBatch::ToAmqpMessage() is const (src/event_data_batch.cpp:22). m_rwMutex is not mutable (inc/azure/messaging/eventhubs/event_data_batch.hpp:56). A const method cannot lock a mutex that is not mutable. TryAddAmqpMessage takes that lock at src/event_data_batch.cpp:68. ProducerClient::Send calls ToAmqpMessage() first (src/producer_client.cpp:102). A TryAdd from a second thread thus races the serialization.

The transport does not solve this. AMQP 1.0 does not permit two messages to interleave their frames on one link. The unit of mutual exclusion for a shared sender link is therefore the full delivery. azure-core-amqp does almost no locking. src/impl/uamqp/amqp/message_sender.cpp has no mutex. One global polling thread drives every protocol object (azure-core-amqp/src/common/global_state.cpp:109). The objects register with that thread through AddPollable. MessageSenderImpl::Open calls AddPollable (src/impl/uamqp/amqp/message_sender.cpp:290,340). The note at azure-core-amqp/src/common/global_state.cpp:159 says that the caller must not keep a connection lock or a link lock across that call. The polling thread takes the connection lock, and the two threads then deadlock.

Qpid Proton C++ uses the same design. Its protocol objects are not thread safe. A work queue for each connection does the serialization. The Event Hubs layer must do the serialization itself.

Proposal

Correct the three races first. Each one is small, and each one can merge on its own. Then change the locking design, which breaks the ABI. Then add the contract and the tests.

• Lock the sender map read in ProducerClient::GetSender #7285 Lock the sender map read in ProducerClient::GetSender. [GA blocker]
• Move the Event Hubs link and session creation outside the client locks #7286 Move the link and session creation outside the client locks. [GA blocker]
• Lock the batch read in EventDataBatch::ToAmqpMessage #7287 Lock the batch read in EventDataBatch::ToAmqpMessage, and make a wrong parallel call fail immediately. [GA blocker]
• Add two-tier client locking and move the mutable state behind a pimpl #7288 Add the two-tier locking design, and move the mutable client state behind a pimpl. [GA blocker]
• Document the thread safety contract on every public Event Hubs type #7289 Document the thread safety contract on every public type. [GA quality bar]
• Add a thread sanitizer CI lane and Event Hubs thread safety tests #7290 Add a thread sanitizer CI lane and thread safety tests. [GA quality bar]

The two-tier design is the target.

Tier 1 protects the client registries. Replace m_sendersLock, m_receiversLock and m_sessionsLock with a std::shared_mutex. Take a shared lock to read the map. Take a unique lock to write the map. Do not keep either lock across a network operation.

Tier 2 protects the wire. Give each partition entry its own mutex in the map value. Keep that mutex for the full delivery, because AMQP does not permit interleaving on one link. Parallel links on different partitions then give the throughput. The .NET library gets its throughput from the same place.

azure-core-amqp can need single-threaded access, as Qpid Proton does. A work queue for each connection then replaces the mutex for each partition. The public contract stays the same.

Settle the ABI question now. New mutex members change the size and the layout of the public client classes. The library is on the 1.0.0-beta line, so the break costs nothing today. After GA the break becomes impossible. Move the mutable state behind a pimpl in the same change. Later locking corrections then keep the ABI.

Validation

• The thread sanitizer reports no race for parallel sends to different partitions on one shared ProducerClient.
• The thread sanitizer reports no race for parallel sends to the same partition.
• A TryAdd during a Send on the same batch fails each time. The batch stays correct.
• Each public client header gives its thread safety guarantee.
• Each model type gives its restriction.
• The GA API review quotes cpp-design-client-methods-thread-safety. The review passes.