Skip to content
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

SystemLayerImplSelect: fix Request/ClearCallbackOnPendingXXX() for… #21135

Merged
merged 1 commit into from
Jul 25, 2022
Merged

SystemLayerImplSelect: fix Request/ClearCallbackOnPendingXXX() for… #21135

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
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
37 changes: 0 additions & 37 deletions src/inet/TCPEndPointImplSockets.cpp
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -132,30 +132,6 @@ CHIP_ERROR TCPEndPointImplSockets::BindImpl(IPAddressType addrType, const IPAddr
}
}

#if CHIP_SYSTEM_CONFIG_USE_DISPATCH
dispatch_queue_t dispatchQueue = static_cast<System::LayerSocketsLoop &>(GetSystemLayer()).GetDispatchQueue();
if (dispatchQueue != nullptr)
{
unsigned long fd = static_cast<unsigned long>(mSocket);

mReadableSource = dispatch_source_create(DISPATCH_SOURCE_TYPE_READ, fd, 0, dispatchQueue);
ReturnErrorCodeIf(mReadableSource == nullptr, CHIP_ERROR_NO_MEMORY);

mWriteableSource = dispatch_source_create(DISPATCH_SOURCE_TYPE_WRITE, fd, 0, dispatchQueue);
ReturnErrorCodeIf(mWriteableSource == nullptr, CHIP_ERROR_NO_MEMORY);

dispatch_source_set_event_handler(mReadableSource, ^{
this->HandlePendingIO(System::SocketEventFlags::kRead);
});

dispatch_source_set_event_handler(mWriteableSource, ^{
this->HandlePendingIO(System::SocketEventFlags::kWrite);
});

dispatch_resume(mReadableSource);
dispatch_resume(mWriteableSource);
}
#endif // CHIP_SYSTEM_CONFIG_USE_DISPATCH
return res;
}

Expand Down Expand Up @@ -648,19 +624,6 @@ void TCPEndPointImplSockets::DoCloseImpl(CHIP_ERROR err, State oldState)
mSocket = kInvalidSocketFd;
}
}

#if CHIP_SYSTEM_CONFIG_USE_DISPATCH
if (mReadableSource)
{
dispatch_source_cancel(mReadableSource);
dispatch_release(mReadableSource);
}
if (mWriteableSource)
{
dispatch_source_cancel(mWriteableSource);
dispatch_release(mWriteableSource);
}
#endif // CHIP_SYSTEM_CONFIG_USE_DISPATCH
}

#if INET_CONFIG_OVERRIDE_SYSTEM_TCP_USER_TIMEOUT
Expand Down
9 changes: 0 additions & 9 deletions src/inet/TCPEndPointImplSockets.h
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -25,10 +25,6 @@
#include <inet/EndPointStateSockets.h>
#include <inet/TCPEndPoint.h>

#if CHIP_SYSTEM_CONFIG_USE_DISPATCH
#include <dispatch/dispatch.h>
#endif

namespace chip {
namespace Inet {

Expand Down Expand Up @@ -74,11 +70,6 @@ class TCPEndPointImplSockets : public TCPEndPoint, public EndPointStateSockets
CHIP_ERROR BindSrcAddrFromIntf(IPAddressType addrType, InterfaceId intfId);
static void HandlePendingIO(System::SocketEvents events, intptr_t data);

#if CHIP_SYSTEM_CONFIG_USE_DISPATCH
dispatch_source_t mReadableSource = nullptr;
dispatch_source_t mWriteableSource = nullptr;
#endif // CHIP_SYSTEM_CONFIG_USE_DISPATCH

#if INET_CONFIG_OVERRIDE_SYSTEM_TCP_USER_TIMEOUT
/// This counts the number of bytes written on the TCP socket since thelast probe into the TCP outqueue was made.
uint32_t mBytesWrittenSinceLastProbe;
Expand Down
24 changes: 0 additions & 24 deletions src/inet/UDPEndPointImplSockets.cpp
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -205,22 +205,6 @@ CHIP_ERROR UDPEndPointImplSockets::BindImpl(IPAddressType addressType, const IPA
}
}

#if CHIP_SYSTEM_CONFIG_USE_DISPATCH
dispatch_queue_t dispatchQueue = static_cast<System::LayerSocketsLoop *>(&GetSystemLayer())->GetDispatchQueue();
if (dispatchQueue != nullptr)
{
unsigned long fd = static_cast<unsigned long>(mSocket);

mReadableSource = dispatch_source_create(DISPATCH_SOURCE_TYPE_READ, fd, 0, dispatchQueue);
ReturnErrorCodeIf(mReadableSource == nullptr, CHIP_ERROR_NO_MEMORY);

dispatch_source_set_event_handler(mReadableSource, ^{
this->HandlePendingIO(System::SocketEventFlags::kRead);
});
dispatch_resume(mReadableSource);
}
#endif // CHIP_SYSTEM_CONFIG_USE_DISPATCH

return CHIP_NO_ERROR;
}

Expand Down Expand Up @@ -431,14 +415,6 @@ void UDPEndPointImplSockets::CloseImpl()
close(mSocket);
mSocket = kInvalidSocketFd;
}

#if CHIP_SYSTEM_CONFIG_USE_DISPATCH
if (mReadableSource)
{
dispatch_source_cancel(mReadableSource);
dispatch_release(mReadableSource);
}
#endif // CHIP_SYSTEM_CONFIG_USE_DISPATCH
}

void UDPEndPointImplSockets::Free()
Expand Down
8 changes: 0 additions & 8 deletions src/inet/UDPEndPointImplSockets.h
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -26,10 +26,6 @@
#include <inet/EndPointStateSockets.h>
#include <inet/UDPEndPoint.h>

#if CHIP_SYSTEM_CONFIG_USE_DISPATCH
#include <dispatch/dispatch.h>
#endif

namespace chip {
namespace Inet {

Expand Down Expand Up @@ -65,10 +61,6 @@ class UDPEndPointImplSockets : public UDPEndPoint, public EndPointStateSockets
InterfaceId mBoundIntfId;
uint16_t mBoundPort;

#if CHIP_SYSTEM_CONFIG_USE_DISPATCH
dispatch_source_t mReadableSource = nullptr;
#endif // CHIP_SYSTEM_CONFIG_USE_DISPATCH

#if CHIP_SYSTEM_CONFIG_USE_PLATFORM_MULTICAST_API
public:
using MulticastGroupHandler = CHIP_ERROR (*)(InterfaceId, const IPAddress &);
Expand Down
83 changes: 83 additions & 0 deletions src/system/SystemLayerImplSelect.cpp
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -246,6 +246,30 @@ CHIP_ERROR LayerImplSelect::RequestCallbackOnPendingRead(SocketWatchToken token)
VerifyOrReturnError(watch != nullptr, CHIP_ERROR_INVALID_ARGUMENT);

watch->mPendingIO.Set(SocketEventFlags::kRead);

#if CHIP_SYSTEM_CONFIG_USE_DISPATCH
dispatch_queue_t dispatchQueue = GetDispatchQueue();
if (watch->mWrSource)
{
dispatch_resume(watch->mRdSource);
}
else
{
if (dispatchQueue)
{
watch->mRdSource =
dispatch_source_create(DISPATCH_SOURCE_TYPE_READ, static_cast<uintptr_t>(watch->mFD), 0, dispatchQueue);
ReturnErrorCodeIf(watch->mRdSource == nullptr, CHIP_ERROR_NO_MEMORY);
dispatch_source_set_event_handler(watch->mRdSource, ^{
SocketEvents events;
events.Set(SocketEventFlags::kRead);
watch->mCallback(events, watch->mCallbackData);
});
dispatch_activate(watch->mRdSource);
}
}
#endif // CHIP_SYSTEM_CONFIG_USE_DISPATCH

return CHIP_NO_ERROR;
}

Expand All @@ -255,6 +279,30 @@ CHIP_ERROR LayerImplSelect::RequestCallbackOnPendingWrite(SocketWatchToken token
VerifyOrReturnError(watch != nullptr, CHIP_ERROR_INVALID_ARGUMENT);

watch->mPendingIO.Set(SocketEventFlags::kWrite);

#if CHIP_SYSTEM_CONFIG_USE_DISPATCH
if (watch->mWrSource)
{
dispatch_resume(watch->mWrSource);
Copy link
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

I don't think this is right... For example, TCP will call RequestCallbackOnPendingWrite for every single packet, without doing any ClearCallbackOnPendingWrite in between. So we will have multiple dispatch_resume calls when the dispatch source is already resumed. You're not allowed to do that, per the API documentation.

I have not carefully checked for other failure modes in TCP and UDP here, but fundamentally assuming that their calls to RequestCallback* and ClearCallback* will be ordered in a sane way is not reasonable: those calls have an API designed around a single boolean flag, not a counter. So we need to keep track with a boolean flag, and ensure our counter is only ever 0 or 1.

Copy link
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

I agree, I will update the change accordingly.

}
else
{
dispatch_queue_t dispatchQueue = GetDispatchQueue();
if (dispatchQueue)
{
watch->mWrSource =
dispatch_source_create(DISPATCH_SOURCE_TYPE_WRITE, static_cast<uintptr_t>(watch->mFD), 0, dispatchQueue);
ReturnErrorCodeIf(watch->mWrSource == nullptr, CHIP_ERROR_NO_MEMORY);
dispatch_source_set_event_handler(watch->mWrSource, ^{
SocketEvents events;
events.Set(SocketEventFlags::kWrite);
watch->mCallback(events, watch->mCallbackData);
});
dispatch_activate(watch->mWrSource);
}
}
#endif // CHIP_SYSTEM_CONFIG_USE_DISPATCH

return CHIP_NO_ERROR;
}

Expand All @@ -263,6 +311,14 @@ CHIP_ERROR LayerImplSelect::ClearCallbackOnPendingRead(SocketWatchToken token)
SocketWatch * watch = reinterpret_cast<SocketWatch *>(token);
VerifyOrReturnError(watch != nullptr, CHIP_ERROR_INVALID_ARGUMENT);
watch->mPendingIO.Clear(SocketEventFlags::kRead);

#if CHIP_SYSTEM_CONFIG_USE_DISPATCH
if (watch->mRdSource)
{
dispatch_suspend(watch->mRdSource);
}
#endif // CHIP_SYSTEM_CONFIG_USE_DISPATCH

return CHIP_NO_ERROR;
}

Expand All @@ -271,6 +327,14 @@ CHIP_ERROR LayerImplSelect::ClearCallbackOnPendingWrite(SocketWatchToken token)
SocketWatch * watch = reinterpret_cast<SocketWatch *>(token);
VerifyOrReturnError(watch != nullptr, CHIP_ERROR_INVALID_ARGUMENT);
watch->mPendingIO.Clear(SocketEventFlags::kWrite);

#if CHIP_SYSTEM_CONFIG_USE_DISPATCH
if (watch->mWrSource)
{
dispatch_suspend(watch->mWrSource);
}
#endif // CHIP_SYSTEM_CONFIG_USE_DISPATCH

return CHIP_NO_ERROR;
}

Expand All @@ -282,10 +346,25 @@ CHIP_ERROR LayerImplSelect::StopWatchingSocket(SocketWatchToken * tokenInOut)
VerifyOrReturnError(watch != nullptr, CHIP_ERROR_INVALID_ARGUMENT);
VerifyOrReturnError(watch->mFD >= 0, CHIP_ERROR_INCORRECT_STATE);

#if CHIP_SYSTEM_CONFIG_USE_DISPATCH
if (watch->mRdSource)
{
dispatch_cancel(watch->mRdSource);
dispatch_release(watch->mRdSource);
}
if (watch->mWrSource)
{
dispatch_cancel(watch->mWrSource);
dispatch_release(watch->mWrSource);
}
#endif

watch->Clear();

#if !CHIP_SYSTEM_CONFIG_USE_DISPATCH
// Wake the thread calling select so that it stops selecting on the socket.
Signal();
#endif

return CHIP_NO_ERROR;
}
Expand Down Expand Up @@ -426,6 +505,10 @@ void LayerImplSelect::SocketWatch::Clear()
mPendingIO.ClearAll();
mCallback = nullptr;
mCallbackData = 0;
#if CHIP_SYSTEM_CONFIG_USE_DISPATCH
mRdSource = nullptr;
mWrSource = nullptr;
#endif
}

} // namespace System
Expand Down
4 changes: 4 additions & 0 deletions src/system/SystemLayerImplSelect.h
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -90,6 +90,10 @@ class LayerImplSelect : public LayerSocketsLoop
int mFD;
SocketEvents mPendingIO;
SocketWatchCallback mCallback;
#if CHIP_SYSTEM_CONFIG_USE_DISPATCH
dispatch_source_t mRdSource;
dispatch_source_t mWrSource;
#endif
intptr_t mCallbackData;
};
SocketWatch mSocketWatchPool[kSocketWatchMax];
Expand Down