Skip to content

Commit

Permalink
CommandSender: For batch commands, track requests are responded to (#…
Browse files Browse the repository at this point in the history 
…32105)

* CommandSender: For batch commands, track requests are responded to

* Rename

* Restyled by whitespace

* Restyled by clang-format

* Restyled by gn

* Self review updates

* Quick fix

* Quick fix

* Restyled by whitespace

* More fixes

* Fix CI

* Restyled by whitespace

* Restyled by clang-format

* Restyled by gn

* Apply suggestions from code review

Co-authored-by: Boris Zbarsky <bzbarsky@apple.com>

* Address PR comments

* Fix CI and address some PR comments

* Restyled by clang-format

* Update src/app/CommandSender.cpp

Co-authored-by: Boris Zbarsky <bzbarsky@apple.com>

* Update src/app/CommandSender.cpp

Co-authored-by: Boris Zbarsky <bzbarsky@apple.com>

* Address PR comments

* Address comment

---------

Co-authored-by: Restyled.io <commits@restyled.io>
Co-authored-by: Boris Zbarsky <bzbarsky@apple.com>
  • Loading branch information
@tehampson @restyled-commits
@bzbarsky-apple @pull
3 people authored and pull[bot] committed May 3, 2024
1 parent 59e3cc5 commit 3080793
Show file tree
Hide file tree
Showing 12 changed files with 517 additions and 49 deletions.
3 changes: 3 additions & 0 deletions scripts/tools/check_includes_config.py
View file
Original file line number Diff line number Diff line change
Expand Up @@ -160,6 +160,9 @@
'src/tracing/json/json_tracing.cpp': {'string', 'sstream'},
'src/tracing/json/json_tracing.h': {'fstream', 'unordered_map'},

# Not intended for embedded clients
'src/app/PendingResponseTrackerImpl.h': {'unordered_set'},

# Not intended for embedded clients
'src/lib/support/jsontlv/JsonToTlv.cpp': {'sstream'},
'src/lib/support/jsontlv/JsonToTlv.h': {'string'},
Expand Down
3 changes: 3 additions & 0 deletions src/app/BUILD.gn
View file
Original file line number Diff line number Diff line change
Expand Up @@ -231,6 +231,9 @@ static_library("app") {
"FailSafeContext.cpp",
"FailSafeContext.h",
"OTAUserConsentCommon.h",
"PendingResponseTracker.h",
"PendingResponseTrackerImpl.cpp",
"PendingResponseTrackerImpl.h",
"ReadHandler.cpp",
"SafeAttributePersistenceProvider.h",
"TimedRequest.cpp",
Expand Down
85 changes: 70 additions & 15 deletions src/app/CommandSender.cpp
View file
Original file line number Diff line number Diff line change
Expand Up @@ -75,6 +75,9 @@ CommandSender::CommandSender(ExtendableCallback * apExtendableCallback, Messagin
mTimedRequest(aIsTimedRequest), mUseExtendableCallback(true)
{
assertChipStackLockedByCurrentThread();
#if CHIP_CONFIG_COMMAND_SENDER_BUILTIN_SUPPORT_FOR_BATCHED_COMMANDS
mpPendingResponseTracker = &mNonTestPendingResponseTracker;
#endif // CHIP_CONFIG_COMMAND_SENDER_BUILTIN_SUPPORT_FOR_BATCHED_COMMANDS
}

CommandSender::~CommandSender()
Expand Down Expand Up @@ -222,9 +225,9 @@ CHIP_ERROR CommandSender::OnMessageReceived(Messaging::ExchangeContext * apExcha

if (aPayloadHeader.HasMessageType(MsgType::InvokeCommandResponse))
{
mInvokeResponseMessageCount++;
err = ProcessInvokeResponse(std::move(aPayload), moreChunkedMessages);
SuccessOrExit(err);
mInvokeResponseMessageCount++;
if (moreChunkedMessages)
{
StatusResponse::Send(Status::Success, apExchangeContext, /*aExpectResponse = */ true);
Expand Down Expand Up @@ -258,6 +261,10 @@ CHIP_ERROR CommandSender::OnMessageReceived(Messaging::ExchangeContext * apExcha

if (mState != State::AwaitingResponse)
{
if (err == CHIP_NO_ERROR)
{
FlushNoCommandResponse();
}
Close();
}
// Else we got a response to a Timed Request and just sent the invoke.
Expand Down Expand Up @@ -331,12 +338,25 @@ void CommandSender::OnResponseTimeout(Messaging::ExchangeContext * apExchangeCon
Close();
}

void CommandSender::FlushNoCommandResponse()
{
if (mpPendingResponseTracker && mUseExtendableCallback && mCallbackHandle.extendableCallback)
{
Optional<uint16_t> commandRef = mpPendingResponseTracker->PopPendingResponse();
while (commandRef.HasValue())
{
NoResponseData noResponseData = { commandRef.Value() };
mCallbackHandle.extendableCallback->OnNoResponse(this, noResponseData);
commandRef = mpPendingResponseTracker->PopPendingResponse();
}
}
}

void CommandSender::Close()
{
mSuppressResponse = false;
mTimedRequest = false;
MoveToState(State::AwaitingDestruction);

OnDoneCallback();
}

Expand All @@ -350,10 +370,10 @@ CHIP_ERROR CommandSender::ProcessInvokeResponseIB(InvokeResponseIB::Parser & aIn
StatusIB statusIB;

{
bool commandRefExpected = (mFinishedCommandCount > 1);
bool hasDataResponse = false;
bool hasDataResponse = false;
TLV::TLVReader commandDataReader;
Optional<uint16_t> commandRef;
bool commandRefExpected = mpPendingResponseTracker && (mpPendingResponseTracker->Count() > 1);

CommandStatusIB::Parser commandStatus;
err = aInvokeResponse.GetStatus(&commandStatus);
Expand Down Expand Up @@ -409,6 +429,27 @@ CHIP_ERROR CommandSender::ProcessInvokeResponseIB(InvokeResponseIB::Parser & aIn
}
ReturnErrorOnFailure(err);

if (commandRef.HasValue() && mpPendingResponseTracker != nullptr)
{
err = mpPendingResponseTracker->Remove(commandRef.Value());
if (err != CHIP_NO_ERROR)
{
// This can happen for two reasons:
// 1. The current InvokeResponse is a duplicate (based on its commandRef).
// 2. The current InvokeResponse is for a request we never sent (based on its commandRef).
// Used when logging errors related to server violating spec.
[[maybe_unused]] ScopedNodeId remoteScopedNode;
if (mExchangeCtx.Get()->HasSessionHandle())
{
remoteScopedNode = mExchangeCtx.Get()->GetSessionHandle()->GetPeer();
}
ChipLogError(DataManagement,
"Received Unexpected Response from remote node " ChipLogFormatScopedNodeId ", commandRef=%u",
ChipLogValueScopedNodeId(remoteScopedNode), commandRef.Value());
return err;
}
}

// When using ExtendableCallbacks, we are adhering to a different API contract where path
// specific errors are sent to the OnResponse callback. For more information on the history
// of this issue please see https://github.com/project-chip/connectedhomeip/issues/30991
Expand All @@ -430,17 +471,19 @@ CHIP_ERROR CommandSender::ProcessInvokeResponseIB(InvokeResponseIB::Parser & aIn

CHIP_ERROR CommandSender::SetCommandSenderConfig(CommandSender::ConfigParameters & aConfigParams)
{
#if CHIP_CONFIG_SENDING_BATCH_COMMANDS_ENABLED
VerifyOrReturnError(mState == State::Idle, CHIP_ERROR_INCORRECT_STATE);
VerifyOrReturnError(aConfigParams.remoteMaxPathsPerInvoke > 0, CHIP_ERROR_INVALID_ARGUMENT);
if (mpPendingResponseTracker != nullptr)
{

mRemoteMaxPathsPerInvoke = aConfigParams.remoteMaxPathsPerInvoke;
mBatchCommandsEnabled = (aConfigParams.remoteMaxPathsPerInvoke > 1);
return CHIP_NO_ERROR;
#else
VerifyOrReturnError(aConfigParams.remoteMaxPathsPerInvoke == 1, CHIP_ERROR_UNSUPPORTED_CHIP_FEATURE);
mRemoteMaxPathsPerInvoke = aConfigParams.remoteMaxPathsPerInvoke;
mBatchCommandsEnabled = (aConfigParams.remoteMaxPathsPerInvoke > 1);
}
else
{
VerifyOrReturnError(aConfigParams.remoteMaxPathsPerInvoke == 1, CHIP_ERROR_UNSUPPORTED_CHIP_FEATURE);
}
return CHIP_NO_ERROR;
#endif
}

CHIP_ERROR CommandSender::PrepareCommand(const CommandPathParams & aCommandPathParams,
Expand All @@ -453,12 +496,19 @@ CHIP_ERROR CommandSender::PrepareCommand(const CommandPathParams & aCommandPathP
//
bool canAddAnotherCommand = (mState == State::AddedCommand && mBatchCommandsEnabled && mUseExtendableCallback);
VerifyOrReturnError(mState == State::Idle || canAddAnotherCommand, CHIP_ERROR_INCORRECT_STATE);
VerifyOrReturnError(mFinishedCommandCount < mRemoteMaxPathsPerInvoke, CHIP_ERROR_MAXIMUM_PATHS_PER_INVOKE_EXCEEDED);

if (mpPendingResponseTracker != nullptr)
{
size_t pendingCount = mpPendingResponseTracker->Count();
VerifyOrReturnError(pendingCount < mRemoteMaxPathsPerInvoke, CHIP_ERROR_MAXIMUM_PATHS_PER_INVOKE_EXCEEDED);
}

if (mBatchCommandsEnabled)
{
VerifyOrReturnError(mpPendingResponseTracker != nullptr, CHIP_ERROR_INCORRECT_STATE);
VerifyOrReturnError(aPrepareCommandParams.commandRef.HasValue(), CHIP_ERROR_INVALID_ARGUMENT);
VerifyOrReturnError(aPrepareCommandParams.commandRef.Value() == mFinishedCommandCount, CHIP_ERROR_INVALID_ARGUMENT);
uint16_t commandRef = aPrepareCommandParams.commandRef.Value();
VerifyOrReturnError(!mpPendingResponseTracker->IsTracked(commandRef), CHIP_ERROR_INVALID_ARGUMENT);
}

InvokeRequests::Builder & invokeRequests = mInvokeRequestBuilder.GetInvokeRequests();
Expand All @@ -482,8 +532,10 @@ CHIP_ERROR CommandSender::FinishCommand(FinishCommandParameters & aFinishCommand
{
if (mBatchCommandsEnabled)
{
VerifyOrReturnError(mpPendingResponseTracker != nullptr, CHIP_ERROR_INCORRECT_STATE);
VerifyOrReturnError(aFinishCommandParams.commandRef.HasValue(), CHIP_ERROR_INVALID_ARGUMENT);
VerifyOrReturnError(aFinishCommandParams.commandRef.Value() == mFinishedCommandCount, CHIP_ERROR_INVALID_ARGUMENT);
uint16_t commandRef = aFinishCommandParams.commandRef.Value();
VerifyOrReturnError(!mpPendingResponseTracker->IsTracked(commandRef), CHIP_ERROR_INVALID_ARGUMENT);
}

return FinishCommandInternal(aFinishCommandParams);
Expand Down Expand Up @@ -511,7 +563,10 @@ CHIP_ERROR CommandSender::FinishCommandInternal(FinishCommandParameters & aFinis

MoveToState(State::AddedCommand);

mFinishedCommandCount++;
if (mpPendingResponseTracker && aFinishCommandParams.commandRef.HasValue())
{
mpPendingResponseTracker->Add(aFinishCommandParams.commandRef.Value());
}

if (aFinishCommandParams.timedInvokeTimeoutMs.HasValue())
{
Expand Down
63 changes: 50 additions & 13 deletions src/app/CommandSender.h
View file
Original file line number Diff line number Diff line change
Expand Up @@ -32,6 +32,7 @@
#include <app/MessageDef/InvokeRequestMessage.h>
#include <app/MessageDef/InvokeResponseMessage.h>
#include <app/MessageDef/StatusIB.h>
#include <app/PendingResponseTrackerImpl.h>
#include <app/data-model/Encode.h>
#include <lib/core/CHIPCore.h>
#include <lib/core/Optional.h>
Expand Down Expand Up @@ -75,6 +76,16 @@ class CommandSender final : public Messaging::ExchangeDelegate
Optional<uint16_t> commandRef;
};

// CommandSender::ExtendableCallback::OnNoResponse is public SDK API, so we cannot break
// source compatibility for it. To allow for additional values to be added at a future
// time without constantly changing the function's declaration parameter list, we are
// defining the struct NoResponseData and adding that to the parameter list to allow for
// future extendability.
struct NoResponseData
{
uint16_t commandRef;
};

// CommandSender::ExtendableCallback::OnError is public SDK API, so we cannot break source
// compatibility for it. To allow for additional values to be added at a future time
// without constantly changing the function's declaration parameter list, we are
Expand Down Expand Up @@ -127,9 +138,21 @@ class CommandSender final : public Messaging::ExchangeDelegate
* @param[in] apCommandSender The command sender object that initiated the command transaction.
* @param[in] aResponseData Information pertaining to the response.
*/
;
virtual void OnResponse(CommandSender * commandSender, const ResponseData & aResponseData) {}

/**
* Called for each request that failed to receive a response after the server indicates completion of all requests.
*
* This callback may be omitted if clients have alternative ways to track non-responses.
*
* The CommandSender object MUST continue to exist after this call is completed. The application shall wait until it
* receives an OnDone call to destroy the object.
*
* @param apCommandSender The CommandSender object that initiated the transaction.
* @param aNoResponseData Details about the request without a response.
*/
virtual void OnNoResponse(CommandSender * commandSender, const NoResponseData & aNoResponseData) {}

/**
* OnError will be called when a non-path-specific error occurs *after* a successful call to SendCommandRequest().
*
Expand Down Expand Up @@ -229,12 +252,6 @@ class CommandSender final : public Messaging::ExchangeDelegate
// early in PrepareCommand, even though it's not used until FinishCommand. This proactive
// validation helps prevent unnecessary writing an InvokeRequest into the packet that later
// needs to be undone.
//
// Currently, provided commandRefs for the first request must start at 0 and increment by one
// for each subsequent request. This requirement can be relaxed in the future if a compelling
// need arises.
// TODO(#30453): After introducing Request/Response tracking, remove statement above about
// this currently enforced requirement on commandRefs.
Optional<uint16_t> commandRef;
// If the InvokeRequest needs to be in a state with a started data TLV struct container
bool startDataStruct = false;
Expand Down Expand Up @@ -278,6 +295,10 @@ class CommandSender final : public Messaging::ExchangeDelegate
bool endDataStruct = false;
};

class TestOnlyMarker
{
};

/*
* Constructor.
*
Expand All @@ -287,12 +308,20 @@ class CommandSender final : public Messaging::ExchangeDelegate
*/
CommandSender(Callback * apCallback, Messaging::ExchangeManager * apExchangeMgr, bool aIsTimedRequest = false,
bool aSuppressResponse = false);
CommandSender(ExtendableCallback * apCallback, Messaging::ExchangeManager * apExchangeMgr, bool aIsTimedRequest = false,
bool aSuppressResponse = false);
CommandSender(std::nullptr_t, Messaging::ExchangeManager * apExchangeMgr, bool aIsTimedRequest = false,
bool aSuppressResponse = false) :
CommandSender(static_cast<Callback *>(nullptr), apExchangeMgr, aIsTimedRequest, aSuppressResponse)
{}
CommandSender(ExtendableCallback * apCallback, Messaging::ExchangeManager * apExchangeMgr, bool aIsTimedRequest = false,
bool aSuppressResponse = false);
// TODO(#32138): After there is a macro that is always defined for all unit tests, the constructor with
// TestOnlyMarker should only be compiled if that macro is defined.
CommandSender(TestOnlyMarker aTestMarker, ExtendableCallback * apCallback, Messaging::ExchangeManager * apExchangeMgr,
PendingResponseTracker * apPendingResponseTracker, bool aIsTimedRequest = false, bool aSuppressResponse = false) :
CommandSender(apCallback, apExchangeMgr, aIsTimedRequest, aSuppressResponse)
{
mpPendingResponseTracker = apPendingResponseTracker;
}
~CommandSender();

/**
Expand All @@ -307,11 +336,14 @@ class CommandSender final : public Messaging::ExchangeDelegate
* based on how many paths the remote peer claims to support.
*
* @return CHIP_ERROR_INCORRECT_STATE
* If device has previously called `PrepareCommand`.
* If device has previously called `PrepareCommand`.
* @return CHIP_ERROR_INVALID_ARGUMENT
* Invalid argument value.
* Invalid argument value.
* @return CHIP_ERROR_UNSUPPORTED_CHIP_FEATURE
* Device has not enabled CHIP_CONFIG_SENDING_BATCH_COMMANDS_ENABLED.
* Device has not enabled batch command support. To enable:
* 1. Enable the CHIP_CONFIG_COMMAND_SENDER_BUILTIN_SUPPORT_FOR_BATCHED_COMMANDS
* configuration option.
* 2. Ensure you provide ExtendableCallback.
*/
CHIP_ERROR SetCommandSenderConfig(ConfigParameters & aConfigParams);

Expand Down Expand Up @@ -497,6 +529,7 @@ class CommandSender final : public Messaging::ExchangeDelegate
System::PacketBufferHandle && aPayload) override;
void OnResponseTimeout(Messaging::ExchangeContext * apExchangeContext) override;

void FlushNoCommandResponse();
//
// Called internally to signal the completion of all work on this object, gracefully close the
// exchange (by calling into the base class) and finally, signal to the application that it's
Expand Down Expand Up @@ -580,8 +613,12 @@ class CommandSender final : public Messaging::ExchangeDelegate

chip::System::PacketBufferTLVWriter mCommandMessageWriter;

#if CHIP_CONFIG_COMMAND_SENDER_BUILTIN_SUPPORT_FOR_BATCHED_COMMANDS
PendingResponseTrackerImpl mNonTestPendingResponseTracker;
#endif // CHIP_CONFIG_COMMAND_SENDER_BUILTIN_SUPPORT_FOR_BATCHED_COMMANDS
PendingResponseTracker * mpPendingResponseTracker = nullptr;

uint16_t mInvokeResponseMessageCount = 0;
uint16_t mFinishedCommandCount = 0;
uint16_t mRemoteMaxPathsPerInvoke = 1;

State mState = State::Idle;
Expand Down
Loading

0 comments on commit 3080793

Please sign in to comment.