From 9d76424c3c24f0f7b2a47434ca2786c9db90ac5a Mon Sep 17 00:00:00 2001 From: Sam Kumar Date: Tue, 5 Jul 2022 00:34:59 -0700 Subject: [PATCH] [cli-tcp] use TCP Circular Send Buffer in TCP CLI tool and test (#7867) --- src/cli/cli_tcp.cpp | 232 +++++++++++++++++++++++-------- src/cli/cli_tcp.hpp | 40 ++++-- tests/scripts/expect/cli-tcp.exp | 4 +- 3 files changed, 204 insertions(+), 72 deletions(-) diff --git a/src/cli/cli_tcp.cpp b/src/cli/cli_tcp.cpp index 7c242d047..79ece00d7 100644 --- a/src/cli/cli_tcp.cpp +++ b/src/cli/cli_tcp.cpp @@ -56,8 +56,9 @@ TcpExample::TcpExample(Output &aOutput) , mInitialized(false) , mEndpointConnected(false) , mSendBusy(false) + , mUseCircularSendBuffer(true) , mBenchmarkBytesTotal(0) - , mBenchmarkLinksLeft(0) + , mBenchmarkBytesUnsent(0) { } @@ -82,30 +83,59 @@ otError TcpExample::ProcessInit(Arg aArgs[]) if (aArgs[0].IsEmpty()) { - receiveBufferSize = sizeof(mReceiveBuffer); + mUseCircularSendBuffer = true; + receiveBufferSize = sizeof(mReceiveBufferBytes); } else { - uint32_t windowSize; + if (aArgs[0] == "circular") + { + mUseCircularSendBuffer = true; + } + else if (aArgs[0] == "linked") + { + mUseCircularSendBuffer = false; + } + else + { + ExitNow(error = OT_ERROR_INVALID_ARGS); + } - SuccessOrExit(error = aArgs[0].ParseAsUint32(windowSize)); - VerifyOrExit(aArgs[1].IsEmpty(), error = OT_ERROR_INVALID_ARGS); + if (aArgs[1].IsEmpty()) + { + receiveBufferSize = sizeof(mReceiveBufferBytes); + } + else + { + uint32_t windowSize; - receiveBufferSize = windowSize + ((windowSize + 7) >> 3); - VerifyOrExit(receiveBufferSize <= sizeof(mReceiveBuffer) && receiveBufferSize != 0, - error = OT_ERROR_INVALID_ARGS); + SuccessOrExit(error = aArgs[1].ParseAsUint32(windowSize)); + + receiveBufferSize = windowSize + ((windowSize + 7) >> 3); + VerifyOrExit(receiveBufferSize <= sizeof(mReceiveBufferBytes) && receiveBufferSize != 0, + error = OT_ERROR_INVALID_ARGS); + } } + otTcpCircularSendBufferInitialize(&mSendBuffer, mSendBufferBytes, sizeof(mSendBufferBytes)); + { otTcpEndpointInitializeArgs endpointArgs; memset(&endpointArgs, 0x00, sizeof(endpointArgs)); - endpointArgs.mEstablishedCallback = HandleTcpEstablishedCallback; - endpointArgs.mSendDoneCallback = HandleTcpSendDoneCallback; + endpointArgs.mEstablishedCallback = HandleTcpEstablishedCallback; + if (mUseCircularSendBuffer) + { + endpointArgs.mForwardProgressCallback = HandleTcpForwardProgressCallback; + } + else + { + endpointArgs.mSendDoneCallback = HandleTcpSendDoneCallback; + } endpointArgs.mReceiveAvailableCallback = HandleTcpReceiveAvailableCallback; endpointArgs.mDisconnectedCallback = HandleTcpDisconnectedCallback; endpointArgs.mContext = this; - endpointArgs.mReceiveBuffer = mReceiveBuffer; + endpointArgs.mReceiveBuffer = mReceiveBufferBytes; endpointArgs.mReceiveBufferSize = receiveBufferSize; SuccessOrExit(error = otTcpEndpointInitialize(GetInstancePtr(), &mEndpoint, &endpointArgs)); @@ -137,6 +167,7 @@ otError TcpExample::ProcessDeinit(Arg aArgs[]) { otError error = OT_ERROR_NONE; otError endpointError; + otError bufferError; otError listenerError; VerifyOrExit(aArgs[0].IsEmpty(), error = OT_ERROR_INVALID_ARGS); @@ -145,10 +176,14 @@ otError TcpExample::ProcessDeinit(Arg aArgs[]) endpointError = otTcpEndpointDeinitialize(&mEndpoint); mSendBusy = false; + otTcpCircularSendBufferForceDiscardAll(&mSendBuffer); + bufferError = otTcpCircularSendBufferDeinitialize(&mSendBuffer); + listenerError = otTcpListenerDeinitialize(&mListener); mInitialized = false; SuccessOrExit(error = endpointError); + SuccessOrExit(error = bufferError); SuccessOrExit(error = listenerError); exit: @@ -203,18 +238,28 @@ otError TcpExample::ProcessSend(Arg aArgs[]) otError error; VerifyOrExit(mInitialized, error = OT_ERROR_INVALID_STATE); - VerifyOrExit(!mSendBusy, error = OT_ERROR_BUSY); VerifyOrExit(mBenchmarkBytesTotal == 0, error = OT_ERROR_BUSY); - - mSendLink.mNext = nullptr; - mSendLink.mData = mSendBuffer; VerifyOrExit(!aArgs[0].IsEmpty(), error = OT_ERROR_INVALID_ARGS); - mSendLink.mLength = OT_MIN(aArgs[0].GetLength(), sizeof(mSendBuffer)); - memcpy(mSendBuffer, aArgs[0].GetCString(), mSendLink.mLength); VerifyOrExit(aArgs[1].IsEmpty(), error = OT_ERROR_INVALID_ARGS); - SuccessOrExit(error = otTcpSendByReference(&mEndpoint, &mSendLink, 0)); - mSendBusy = true; + if (mUseCircularSendBuffer) + { + size_t written; + SuccessOrExit(error = otTcpCircularSendBufferWrite(&mEndpoint, &mSendBuffer, aArgs[0].GetCString(), + aArgs[0].GetLength(), &written, 0)); + } + else + { + VerifyOrExit(!mSendBusy, error = OT_ERROR_BUSY); + + mSendLink.mNext = nullptr; + mSendLink.mData = mSendBufferBytes; + mSendLink.mLength = OT_MIN(aArgs[0].GetLength(), sizeof(mSendBufferBytes)); + memcpy(mSendBufferBytes, aArgs[0].GetCString(), mSendLink.mLength); + + SuccessOrExit(error = otTcpSendByReference(&mEndpoint, &mSendLink, 0)); + mSendBusy = true; + } exit: return error; @@ -222,8 +267,7 @@ exit: otError TcpExample::ProcessBenchmark(Arg aArgs[]) { - otError error = OT_ERROR_NONE; - uint32_t toSendOut; + otError error = OT_ERROR_NONE; VerifyOrExit(!mSendBusy, error = OT_ERROR_BUSY); VerifyOrExit(mBenchmarkBytesTotal == 0, error = OT_ERROR_BUSY); @@ -239,30 +283,37 @@ otError TcpExample::ProcessBenchmark(Arg aArgs[]) } VerifyOrExit(aArgs[1].IsEmpty(), error = OT_ERROR_INVALID_ARGS); - memset(mSendBuffer, 'a', sizeof(mSendBuffer)); + mBenchmarkStart = TimerMilli::GetNow(); + mBenchmarkBytesUnsent = mBenchmarkBytesTotal; - mBenchmarkLinksLeft = (mBenchmarkBytesTotal + sizeof(mSendBuffer) - 1) / sizeof(mSendBuffer); - toSendOut = OT_MIN(OT_ARRAY_LENGTH(mBenchmarkLinks), mBenchmarkLinksLeft); - mBenchmarkStart = TimerMilli::GetNow(); - for (uint32_t i = 0; i != toSendOut; i++) + if (mUseCircularSendBuffer) { - mBenchmarkLinks[i].mNext = nullptr; - mBenchmarkLinks[i].mData = mSendBuffer; - mBenchmarkLinks[i].mLength = sizeof(mSendBuffer); - if (i == 0 && mBenchmarkBytesTotal % sizeof(mSendBuffer) != 0) + SuccessOrExit(error = ContinueBenchmarkCircularSend()); + } + else + { + uint32_t benchmarkLinksLeft = (mBenchmarkBytesTotal + sizeof(mSendBufferBytes) - 1) / sizeof(mSendBufferBytes); + uint32_t toSendOut = OT_MIN(OT_ARRAY_LENGTH(mBenchmarkLinks), benchmarkLinksLeft); + + /* We could also point the linked buffers directly to sBenchmarkData. */ + memset(mSendBufferBytes, 'a', sizeof(mSendBufferBytes)); + + for (uint32_t i = 0; i != toSendOut; i++) { - mBenchmarkLinks[i].mLength = mBenchmarkBytesTotal % sizeof(mSendBuffer); + mBenchmarkLinks[i].mNext = nullptr; + mBenchmarkLinks[i].mData = mSendBufferBytes; + mBenchmarkLinks[i].mLength = sizeof(mSendBufferBytes); + if (i == 0 && mBenchmarkBytesTotal % sizeof(mSendBufferBytes) != 0) + { + mBenchmarkLinks[i].mLength = mBenchmarkBytesTotal % sizeof(mSendBufferBytes); + } + error = otTcpSendByReference(&mEndpoint, &mBenchmarkLinks[i], + i == toSendOut - 1 ? 0 : OT_TCP_SEND_MORE_TO_COME); + VerifyOrExit(error == OT_ERROR_NONE, mBenchmarkBytesTotal = 0); } - SuccessOrExit(error = otTcpSendByReference(&mEndpoint, &mBenchmarkLinks[i], - i == toSendOut - 1 ? 0 : OT_TCP_SEND_MORE_TO_COME)); } exit: - if (error != OT_ERROR_NONE) - { - mBenchmarkBytesTotal = 0; - mBenchmarkLinksLeft = 0; - } return error; } @@ -350,6 +401,12 @@ void TcpExample::HandleTcpSendDoneCallback(otTcpEndpoint *aEndpoint, otLinkedBuf static_cast(otTcpEndpointGetContext(aEndpoint))->HandleTcpSendDone(aEndpoint, aData); } +void TcpExample::HandleTcpForwardProgressCallback(otTcpEndpoint *aEndpoint, size_t aInSendBuffer, size_t aBacklog) +{ + static_cast(otTcpEndpointGetContext(aEndpoint)) + ->HandleTcpForwardProgress(aEndpoint, aInSendBuffer, aBacklog); +} + void TcpExample::HandleTcpReceiveAvailableCallback(otTcpEndpoint *aEndpoint, size_t aBytesAvailable, bool aEndOfStream, @@ -388,6 +445,7 @@ void TcpExample::HandleTcpEstablished(otTcpEndpoint *aEndpoint) void TcpExample::HandleTcpSendDone(otTcpEndpoint *aEndpoint, otLinkedBuffer *aData) { OT_UNUSED_VARIABLE(aEndpoint); + OT_ASSERT(!mUseCircularSendBuffer); // this callback is not used when using the circular send buffer if (mBenchmarkBytesTotal == 0) { @@ -402,25 +460,44 @@ void TcpExample::HandleTcpSendDone(otTcpEndpoint *aEndpoint, otLinkedBuffer *aDa else { OT_ASSERT(aData != &mSendLink); - mBenchmarkLinksLeft--; - if (mBenchmarkLinksLeft >= OT_ARRAY_LENGTH(mBenchmarkLinks)) + OT_ASSERT(mBenchmarkBytesUnsent >= aData->mLength); + mBenchmarkBytesUnsent -= aData->mLength; // could be less than sizeof(mSendBufferBytes) for the first link + if (mBenchmarkBytesUnsent >= OT_ARRAY_LENGTH(mBenchmarkLinks) * sizeof(mSendBufferBytes)) { - aData->mLength = sizeof(mSendBuffer); + aData->mLength = sizeof(mSendBufferBytes); if (otTcpSendByReference(&mEndpoint, aData, 0) != OT_ERROR_NONE) { OutputLine("TCP Benchmark Failed"); mBenchmarkBytesTotal = 0; } } - else if (mBenchmarkLinksLeft == 0) + else if (mBenchmarkBytesUnsent == 0) { - uint32_t milliseconds = TimerMilli::GetNow() - mBenchmarkStart; - uint32_t thousandTimesGoodput = (1000 * (mBenchmarkBytesTotal << 3) + (milliseconds >> 1)) / milliseconds; + CompleteBenchmark(); + } + } +} - OutputLine("TCP Benchmark Complete: Transferred %u bytes in %u milliseconds", - static_cast(mBenchmarkBytesTotal), static_cast(milliseconds)); - OutputLine("TCP Goodput: %u.%03u kb/s", thousandTimesGoodput / 1000, thousandTimesGoodput % 1000); - mBenchmarkBytesTotal = 0; +void TcpExample::HandleTcpForwardProgress(otTcpEndpoint *aEndpoint, size_t aInSendBuffer, size_t aBacklog) +{ + OT_UNUSED_VARIABLE(aEndpoint); + OT_UNUSED_VARIABLE(aBacklog); + OT_ASSERT(mUseCircularSendBuffer); // this callback is only used when using the circular send buffer + + otTcpCircularSendBufferHandleForwardProgress(&mSendBuffer, aInSendBuffer); + + /* Handle case where we're in a benchmark. */ + if (mBenchmarkBytesTotal != 0) + { + if (mBenchmarkBytesUnsent != 0) + { + /* Continue sending out data if there's data we haven't sent. */ + IgnoreError(ContinueBenchmarkCircularSend()); + } + else if (aInSendBuffer == 0) + { + /* Handle case where all data is sent out and the send buffer has drained. */ + CompleteBenchmark(); } } } @@ -441,7 +518,7 @@ void TcpExample::HandleTcpReceiveAvailable(otTcpEndpoint *aEndpoint, IgnoreError(otTcpReceiveByReference(aEndpoint, &data)); for (; data != nullptr; data = data->mNext) { - OutputLine("TCP: Received %u bytes: %.*s", static_cast(data->mLength), data->mLength, + OutputLine("TCP: Received %u bytes: %.*s", data->mLength, data->mLength, reinterpret_cast(data->mData)); totalReceived += data->mLength; } @@ -482,17 +559,18 @@ void TcpExample::HandleTcpDisconnected(otTcpEndpoint *aEndpoint, otTcpDisconnect mSendBusy = false; // Mark the benchmark as inactive if the connection was disconnected. - if (mBenchmarkBytesTotal != 0) - { - mBenchmarkBytesTotal = 0; - mBenchmarkLinksLeft = 0; - } + mBenchmarkBytesTotal = 0; + mBenchmarkBytesUnsent = 0; + + otTcpCircularSendBufferForceDiscardAll(&mSendBuffer); } otTcpIncomingConnectionAction TcpExample::HandleTcpAcceptReady(otTcpListener * aListener, const otSockAddr *aPeer, otTcpEndpoint ** aAcceptInto) { + otTcpIncomingConnectionAction action; + OT_UNUSED_VARIABLE(aListener); if (mEndpointConnected) @@ -501,11 +579,14 @@ otTcpIncomingConnectionAction TcpExample::HandleTcpAcceptReady(otTcpListener * OutputSockAddr(*aPeer); OutputLine(" (active socket is busy)"); - return OT_TCP_INCOMING_CONNECTION_ACTION_DEFER; + ExitNow(action = OT_TCP_INCOMING_CONNECTION_ACTION_DEFER); } *aAcceptInto = &mEndpoint; - return OT_TCP_INCOMING_CONNECTION_ACTION_ACCEPT; + action = OT_TCP_INCOMING_CONNECTION_ACTION_ACCEPT; + +exit: + return action; } void TcpExample::HandleTcpAcceptDone(otTcpListener *aListener, otTcpEndpoint *aEndpoint, const otSockAddr *aPeer) @@ -517,6 +598,45 @@ void TcpExample::HandleTcpAcceptDone(otTcpListener *aListener, otTcpEndpoint *aE OutputSockAddrLine(*aPeer); } +otError TcpExample::ContinueBenchmarkCircularSend(void) +{ + otError error = OT_ERROR_NONE; + size_t freeSpace; + + while (mBenchmarkBytesUnsent != 0 && (freeSpace = otTcpCircularSendBufferGetFreeSpace(&mSendBuffer)) != 0) + { + size_t toSendThisIteration = OT_MIN(mBenchmarkBytesUnsent, sBenchmarkDataLength); + uint32_t flag = (toSendThisIteration < freeSpace && toSendThisIteration < mBenchmarkBytesUnsent) + ? OT_TCP_CIRCULAR_SEND_BUFFER_WRITE_MORE_TO_COME + : 0; + size_t written; + + SuccessOrExit(error = otTcpCircularSendBufferWrite(&mEndpoint, &mSendBuffer, sBenchmarkData, + toSendThisIteration, &written, flag)); + mBenchmarkBytesUnsent -= written; + } + +exit: + if (error != OT_ERROR_NONE) + { + OutputLine("TCP Benchmark Failed"); + mBenchmarkBytesTotal = 0; + mBenchmarkBytesUnsent = 0; + } + + return error; +} + +void TcpExample::CompleteBenchmark(void) +{ + uint32_t milliseconds = TimerMilli::GetNow() - mBenchmarkStart; + uint32_t thousandTimesGoodput = (1000 * (mBenchmarkBytesTotal << 3) + (milliseconds >> 1)) / milliseconds; + + OutputLine("TCP Benchmark Complete: Transferred %u bytes in %u milliseconds", mBenchmarkBytesTotal, milliseconds); + OutputLine("TCP Goodput: %u.%03u kb/s", thousandTimesGoodput / 1000, thousandTimesGoodput % 1000); + mBenchmarkBytesTotal = 0; +} + } // namespace Cli } // namespace ot diff --git a/src/cli/cli_tcp.hpp b/src/cli/cli_tcp.hpp index 4bebc9632..d450a37d0 100644 --- a/src/cli/cli_tcp.hpp +++ b/src/cli/cli_tcp.hpp @@ -37,6 +37,7 @@ #include "openthread-core-config.h" #include +#include #include "cli/cli_config.h" #include "cli/cli_output.hpp" @@ -85,8 +86,12 @@ private: otError ProcessListen(Arg aArgs[]); otError ProcessStopListening(Arg aArgs[]); + otError ContinueBenchmarkCircularSend(void); + void CompleteBenchmark(void); + static void HandleTcpEstablishedCallback(otTcpEndpoint *aEndpoint); static void HandleTcpSendDoneCallback(otTcpEndpoint *aEndpoint, otLinkedBuffer *aData); + static void HandleTcpForwardProgressCallback(otTcpEndpoint *aEndpoint, size_t aInSendBuffer, size_t aBacklog); static void HandleTcpReceiveAvailableCallback(otTcpEndpoint *aEndpoint, size_t aBytesAvailable, bool aEndOfStream, @@ -99,13 +104,14 @@ private: otTcpEndpoint * aEndpoint, const otSockAddr *aPeer); - void HandleTcpEstablished(otTcpEndpoint *aEndpoint); - void HandleTcpSendDone(otTcpEndpoint *aEndpoint, otLinkedBuffer *aData); - void HandleTcpReceiveAvailable(otTcpEndpoint *aEndpoint, - size_t aBytesAvailable, - bool aEndOfStream, - size_t aBytesRemaining); - void HandleTcpDisconnected(otTcpEndpoint *aEndpoint, otTcpDisconnectedReason aReason); + void HandleTcpEstablished(otTcpEndpoint *aEndpoint); + void HandleTcpSendDone(otTcpEndpoint *aEndpoint, otLinkedBuffer *aData); + void HandleTcpForwardProgress(otTcpEndpoint *aEndpoint, size_t aInSendBuffer, size_t aBacklog); + void HandleTcpReceiveAvailable(otTcpEndpoint *aEndpoint, + size_t aBytesAvailable, + bool aEndOfStream, + size_t aBytesRemaining); + void HandleTcpDisconnected(otTcpEndpoint *aEndpoint, otTcpDisconnectedReason aReason); otTcpIncomingConnectionAction HandleTcpAcceptReady(otTcpListener * aListener, const otSockAddr *aPeer, otTcpEndpoint ** aAcceptInto); @@ -133,15 +139,21 @@ private: bool mInitialized; bool mEndpointConnected; bool mSendBusy; + bool mUseCircularSendBuffer; - otLinkedBuffer mSendLink; - uint8_t mSendBuffer[OPENTHREAD_CONFIG_CLI_MAX_LINE_LENGTH]; - uint8_t mReceiveBuffer[OPENTHREAD_CONFIG_CLI_TCP_RECEIVE_BUFFER_SIZE]; + otTcpCircularSendBuffer mSendBuffer; + otLinkedBuffer mSendLink; + uint8_t mSendBufferBytes[OPENTHREAD_CONFIG_CLI_TCP_RECEIVE_BUFFER_SIZE]; + uint8_t mReceiveBufferBytes[OPENTHREAD_CONFIG_CLI_TCP_RECEIVE_BUFFER_SIZE]; - otLinkedBuffer mBenchmarkLinks[(sizeof(mReceiveBuffer) + sizeof(mSendBuffer) - 1) / sizeof(mSendBuffer)]; - uint32_t mBenchmarkBytesTotal; - uint32_t mBenchmarkLinksLeft; - TimeMilli mBenchmarkStart; + otLinkedBuffer + mBenchmarkLinks[(sizeof(mReceiveBufferBytes) + sizeof(mSendBufferBytes) - 1) / sizeof(mSendBufferBytes)]; + uint32_t mBenchmarkBytesTotal; + uint32_t mBenchmarkBytesUnsent; + TimeMilli mBenchmarkStart; + + static constexpr const char * sBenchmarkData = "aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa"; + static constexpr const size_t sBenchmarkDataLength = 52; }; } // namespace Cli diff --git a/tests/scripts/expect/cli-tcp.exp b/tests/scripts/expect/cli-tcp.exp index ce641a9d9..46fa325c1 100755 --- a/tests/scripts/expect/cli-tcp.exp +++ b/tests/scripts/expect/cli-tcp.exp @@ -33,11 +33,11 @@ source "tests/scripts/expect/_multinode.exp" setup_two_nodes switch_node 1 -send "tcp init\n" +send "tcp init circular\n" expect_line "Done" switch_node 2 -send "tcp init\n" +send "tcp init linked\n" expect_line "Done" set addr_2 [get_ipaddr mleid] send "tcp listen :: 30000\n"