Skip to content

Separate the locking strategy of each asynchronous socket into separate send and receive paths #421

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
mattrm456 wants to merge 2 commits into
base: main
Choose a base branch
from
Open

Separate the locking strategy of each asynchronous socket into separate send and receive paths #421

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
2 commits
Select commit Hold shift + click to select a range
e4b2d5f
Interim checkpoint: use separate send/receive mutexes in stream sockets
mattrm456 Sep 1, 2026
75a2aa6
Take only the send mutex for encrypted sends and merge the state and …
mattrm456 Oct 2, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
83 changes: 51 additions & 32 deletions groups/ntc/ntcd/ntcd_encryption.cpp
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -1366,13 +1366,6 @@ ntsa::Error Encryption::processIncomingHello()

d_shutdownState.close();
d_handshakeState = e_FAILED;
{
ntccfg::UnLockGuard guard(&d_mutex);
d_handshakeCallback(ntsa::Error(ntsa::Error::e_NOT_AUTHORIZED),
bsl::shared_ptr<ntci::EncryptionCertificate>(),
"Not authorized");
}
d_handshakeCallback = ntci::Encryption::HandshakeCallback();
}
else {
error = this->enqueueOutgoingAccept(true);
Expand All @@ -1385,11 +1378,6 @@ ntsa::Error Encryption::processIncomingHello()
if (d_role == ntca::EncryptionRole::e_CLIENT) {
BSLS_ASSERT(d_remoteCertificate_sp);
d_handshakeState = e_ESTABLISHED;
{
ntccfg::UnLockGuard guard(&d_mutex);
d_handshakeCallback(ntsa::Error(), d_remoteCertificate_sp, "");
}
d_handshakeCallback = ntci::Encryption::HandshakeCallback();
}
else {
error = this->enqueueOutgoingHello();
Expand Down Expand Up @@ -1438,23 +1426,11 @@ ntsa::Error Encryption::processIncomingAccept()
if (d_role == ntca::EncryptionRole::e_SERVER) {
BSLS_ASSERT(d_remoteCertificate_sp);
d_handshakeState = e_ESTABLISHED;
{
ntccfg::UnLockGuard guard(&d_mutex);
d_handshakeCallback(ntsa::Error(), d_remoteCertificate_sp, "");
}
d_handshakeCallback = ntci::Encryption::HandshakeCallback();
}
}
else {
d_shutdownState.close();
d_handshakeState = e_FAILED;
{
ntccfg::UnLockGuard guard(&d_mutex);
d_handshakeCallback(ntsa::Error(ntsa::Error::e_NOT_AUTHORIZED),
bsl::shared_ptr<ntci::EncryptionCertificate>(),
"Not authorized");
}
d_handshakeCallback = ntci::Encryption::HandshakeCallback();
}

return ntsa::Error();
Expand Down Expand Up @@ -1518,7 +1494,7 @@ ntsa::Error Encryption::processIncomingGoodbye()
return ntsa::Error();
}

ntsa::Error Encryption::process()
ntsa::Error Encryption::processData()
{
ntsa::Error error;

Expand Down Expand Up @@ -1568,6 +1544,49 @@ ntsa::Error Encryption::process()
return ntsa::Error();
}

ntsa::Error Encryption::process(LockGuard* lock)
{
ntsa::Error error = this->processData();

// Announce the outcome of the handshake only after all complete incoming
// frames have been consumed, so that a concurrent caller cannot observe
// (and re-process) a partially-processed frame while the mutex is
// released. The handshake callback is reset when the outcome is
// announced, so the outcome is announced at most once.

if (d_handshakeCallback && (d_handshakeState == e_ESTABLISHED ||
d_handshakeState == e_FAILED))
{
ntci::Encryption::HandshakeCallback handshakeCallback =
d_handshakeCallback;

d_handshakeCallback = ntci::Encryption::HandshakeCallback();

const bool established = d_handshakeState == e_ESTABLISHED;

bsl::shared_ptr<ntci::EncryptionCertificate> remoteCertificate;
if (established) {
remoteCertificate = d_remoteCertificate_sp;
}

// Do not re-acquire the mutex after the handshake callback returns:
// the handshake callback may destroy this object.

lock->release()->unlock();

if (established) {
handshakeCallback(ntsa::Error(), remoteCertificate, "");
}
else {
handshakeCallback(ntsa::Error(ntsa::Error::e_NOT_AUTHORIZED),
remoteCertificate,
"Not authorized");
}
}

return error;
}

Encryption::Encryption(
ntca::EncryptionRole::Value role,
const bsl::shared_ptr<ntcd::EncryptionCertificate>& certificate,
Expand Down Expand Up @@ -1627,7 +1646,7 @@ ntsa::Error Encryption::initiateHandshake(const HandshakeCallback& callback)
d_handshakeState = e_HELLO_SENT;
}

return this->process();
return this->process(&guard);
}

ntsa::Error Encryption::pushIncomingCipherText(const bdlbb::Blob& input)
Expand All @@ -1640,7 +1659,7 @@ ntsa::Error Encryption::pushIncomingCipherText(const bdlbb::Blob& input)

bdlbb::BlobUtil::append(d_incomingCipherText_sp.get(), input);

return this->process();
return this->process(&guard);
}

ntsa::Error Encryption::pushIncomingCipherText(const ntsa::Data& input)
Expand All @@ -1653,7 +1672,7 @@ ntsa::Error Encryption::pushIncomingCipherText(const ntsa::Data& input)

ntsa::DataUtil::append(d_incomingCipherText_sp.get(), input);

return this->process();
return this->process(&guard);
}

ntsa::Error Encryption::pushOutgoingPlainText(const bdlbb::Blob& input)
Expand All @@ -1666,7 +1685,7 @@ ntsa::Error Encryption::pushOutgoingPlainText(const bdlbb::Blob& input)

bdlbb::BlobUtil::append(d_outgoingPlainText_sp.get(), input);

return this->process();
return this->process(&guard);
}

ntsa::Error Encryption::pushOutgoingPlainText(const ntsa::Data& input)
Expand All @@ -1679,7 +1698,7 @@ ntsa::Error Encryption::pushOutgoingPlainText(const ntsa::Data& input)

ntsa::DataUtil::append(d_outgoingPlainText_sp.get(), input);

return this->process();
return this->process(&guard);
}

ntsa::Error Encryption::popIncomingPlainText(bdlbb::Blob* output)
Expand All @@ -1693,7 +1712,7 @@ ntsa::Error Encryption::popIncomingPlainText(bdlbb::Blob* output)
d_incomingPlainText_sp->length());
}

return this->process();
return this->process(&guard);
}

ntsa::Error Encryption::popOutgoingCipherText(bdlbb::Blob* output)
Expand All @@ -1707,7 +1726,7 @@ ntsa::Error Encryption::popOutgoingCipherText(bdlbb::Blob* output)
d_outgoingCipherText_sp->length());
}

return this->process();
return this->process(&guard);
}

ntsa::Error Encryption::shutdown()
Expand Down
12 changes: 11 additions & 1 deletion groups/ntc/ntcd/ntcd_encryption.h
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -682,7 +682,17 @@ class Encryption : public ntci::Encryption

/// Process the available data through the encryption state machine.
/// Return the error.
ntsa::Error process();
ntsa::Error processData();

/// Process the available data through the encryption state machine.
/// If the handshake has completed or failed and that outcome has not
/// yet been announced, release the specified 'lock', unlock the mutex,
/// and invoke the handshake callback. Return the error. Note that
/// the handshake callback is invoked only after all complete incoming
/// frames have been consumed, and the mutex is not re-acquired after the
/// handshake callback returns, so the handshake callback may safely
/// destroy this object.
ntsa::Error process(LockGuard* lock);

public:
/// Create a new encryption operating in the specified 'role' from an
Expand Down
33 changes: 26 additions & 7 deletions groups/ntc/ntci/ntci_encryption.h
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -50,6 +50,16 @@ namespace ntci {
/// @par Thread Safety
/// This class is thread safe.
///
/// @par Handshake Callback Invocation
/// Implementations must invoke the handshake callback only from within a call
/// to 'initiateHandshake' or 'pushIncomingCipherText', i.e., only from within
/// the call that supplies the input that determines the outcome of the
/// handshake. Implementations must not invoke the handshake callback from
/// within any other function, in particular 'pushOutgoingPlainText',
/// 'popIncomingPlainText', 'popOutgoingCipherText', or 'shutdown'. Users of
/// this class rely on this guarantee to call those other functions without
/// being prepared to process the completion or failure of the handshake.
///
/// @ingroup module_ntci_encryption
class Encryption
{
Expand All @@ -58,7 +68,9 @@ class Encryption
/// a handshake is complete. The first parameter indicates the error,
/// if any. The second parameter indicates the certificate of the peer,
/// if any. The third parameters indicates the cause of the handshake
/// error, if an error is indicated.
/// error, if an error is indicated. Note that this function is invoked
/// only from within a call to 'initiateHandshake' or
/// 'pushIncomingCipherText'.
typedef NTCCFG_FUNCTION(
const ntsa::Error& error,
const bsl::shared_ptr<ntci::EncryptionCertificate>& certificate,
Expand All @@ -68,30 +80,37 @@ class Encryption
virtual ~Encryption();

/// Initiate the handshake to begin the session. Invoke the specified
/// 'callback' when the handshake completes. Return the error.
/// 'callback' when the handshake completes, either from within this
/// call or from within a subsequent call to 'pushIncomingCipherText'.
/// Return the error.
virtual ntsa::Error initiateHandshake(const HandshakeCallback& callback);

/// Initiate the handshake to begin the session according to the specified
/// 'upgradeOptions'. Invoke the specified 'callback' when the handshake
/// completes. Return the error.
/// completes, either from within this call or from within a subsequent
/// call to 'pushIncomingCipherText'. Return the error.
virtual ntsa::Error initiateHandshake(
const ntca::UpgradeOptions& upgradeOptions,
const HandshakeCallback& callback);

/// Add the specified 'input' containing ciphertext read from the peer.
/// Return the error.
/// Invoke the handshake callback if 'input' completes the handshake or
/// causes it to fail. Return the error.
virtual ntsa::Error pushIncomingCipherText(const bdlbb::Blob& input);

/// Add the specified 'input' containing ciphertext read from the peer.
/// Return the error.
/// Invoke the handshake callback if 'input' completes the handshake or
/// causes it to fail. Return the error.
virtual ntsa::Error pushIncomingCipherText(const ntsa::Data& input);

/// Add the specified 'input' containing plaintext to be sent to the
/// peer. Return the error.
/// peer. Return the error. Note that this function never invokes the
/// handshake callback.
virtual ntsa::Error pushOutgoingPlainText(const bdlbb::Blob& input);

/// Add the specified 'input' containing plaintext to be sent to the
/// peer. Return the error.
/// peer. Return the error. Note that this function never invokes the
/// handshake callback.
virtual ntsa::Error pushOutgoingPlainText(const ntsa::Data& input);

/// Pop plaintext read from the peer and append it to the specified
Expand Down
Loading
Loading