Replace the EndPoint::PrepareIO() functions with stored state

#### Problem Currently, the socket-based networking implementation calls `PrepareIO()` on each endpoint to determine file descriptor sets for `select()`. This usually does redundant work (since the conditions only change on endpoint state changes) and is not usable by event-based interfaces (e.g. epoll/kqueue and libraries like libevent/libev/libuv) – see issue project-chip#5556 _rework system layer event loop_. #### Summary of Changes - Maintain the required flags in an endpoint member, `mRequestIO`, in place of recalculating them in `PrepareIO()`. (The points where `mRequestIO` is changed are the points where an event-based interface would be invoked.) - For UDP and Raw endpoints, pass callbacks to `Listen()` rather than setting members separately, to avoid any possiblity of races in a less synchronous environment. Part of issue project-chip#5556 _rework system layer event loop_.
kpschoedel · Apr 16, 2021 · 7eb4869 · 7eb4869
1 parent 2b6cc31
commit 7eb4869
Show file tree

Hide file tree

Showing 16 changed files with 111 additions and 89 deletions.
diff --git a/src/inet/EndPointBasis.cpp b/src/inet/EndPointBasis.cpp
@@ -41,6 +41,7 @@ void EndPointBasis::InitEndPointBasis(InetLayer & aInetLayer, void * aAppState)
 #if CHIP_SYSTEM_CONFIG_USE_SOCKETS
     mSocket = INET_INVALID_SOCKET_FD;
     mPendingIO.Clear();
+    mRequestIO.Clear();
 #endif // CHIP_SYSTEM_CONFIG_USE_SOCKETS
 }
 

diff --git a/src/inet/EndPointBasis.h b/src/inet/EndPointBasis.h
@@ -95,7 +95,8 @@ class DLL_EXPORT EndPointBasis : public InetLayerBasis
 #if CHIP_SYSTEM_CONFIG_USE_SOCKETS
     int mSocket;             /**< Encapsulated socket descriptor. */
     IPAddressType mAddrType; /**< Protocol family, i.e. IPv4 or IPv6. */
-    SocketEvents mPendingIO; /**< Socket event masks */
+    SocketEvents mPendingIO; /**< Socket event masks (read/write/error) currently available */
+    SocketEvents mRequestIO; /**< Socket event masks (read/write) to wait for */
 #endif                       // CHIP_SYSTEM_CONFIG_USE_SOCKETS
 
 #if CHIP_SYSTEM_CONFIG_USE_LWIP

diff --git a/src/inet/IPEndPointBasis.cpp b/src/inet/IPEndPointBasis.cpp
@@ -1078,16 +1078,6 @@ INET_ERROR IPEndPointBasis::GetSocket(IPAddressType aAddressType, int aType, int
     return INET_NO_ERROR;
 }
 
-SocketEvents IPEndPointBasis::PrepareIO()
-{
-    SocketEvents res;
-
-    if (mState == kState_Listening && OnMessageReceived != nullptr)
-        res.SetRead();
-
-    return res;
-}
-
 void IPEndPointBasis::HandlePendingIO(uint16_t aPort)
 {
     INET_ERROR lStatus = INET_NO_ERROR;

diff --git a/src/inet/IPEndPointBasis.h b/src/inet/IPEndPointBasis.h
@@ -93,9 +93,6 @@ class DLL_EXPORT IPEndPointBasis : public EndPointBasis
     typedef void (*OnMessageReceivedFunct)(IPEndPointBasis * endPoint, chip::System::PacketBufferHandle msg,
                                            const IPPacketInfo * pktInfo);
 
-    /** The endpoint's message reception event handling function delegate. */
-    OnMessageReceivedFunct OnMessageReceived;
-
     /**
      * @brief   Type of reception error event handling function.
      *
@@ -109,16 +106,19 @@ class DLL_EXPORT IPEndPointBasis : public EndPointBasis
      */
     typedef void (*OnReceiveErrorFunct)(IPEndPointBasis * endPoint, INET_ERROR err, const IPPacketInfo * pktInfo);
 
-    /** The endpoint's receive error event handling function delegate. */
-    OnReceiveErrorFunct OnReceiveError;
-
     INET_ERROR SetMulticastLoopback(IPVersion aIPVersion, bool aLoopback);
     INET_ERROR JoinMulticastGroup(InterfaceId aInterfaceId, const IPAddress & aAddress);
     INET_ERROR LeaveMulticastGroup(InterfaceId aInterfaceId, const IPAddress & aAddress);
 
 protected:
     void Init(InetLayer * aInetLayer);
 
+    /** The endpoint's message reception event handling function delegate. */
+    OnMessageReceivedFunct OnMessageReceived;
+
+    /** The endpoint's receive error event handling function delegate. */
+    OnReceiveErrorFunct OnReceiveError;
+
 #if CHIP_SYSTEM_CONFIG_USE_LWIP
 public:
     static struct netif * FindNetifFromInterfaceId(InterfaceId aInterfaceId);
@@ -138,7 +138,7 @@ class DLL_EXPORT IPEndPointBasis : public EndPointBasis
     INET_ERROR BindInterface(IPAddressType aAddressType, InterfaceId aInterfaceId);
     INET_ERROR SendMsg(const IPPacketInfo * aPktInfo, chip::System::PacketBufferHandle aBuffer, uint16_t aSendFlags);
     INET_ERROR GetSocket(IPAddressType aAddressType, int aType, int aProtocol);
-    SocketEvents PrepareIO();
+    SocketEvents PrepareIO() const { return mRequestIO; }
     void HandlePendingIO(uint16_t aPort);
 #endif // CHIP_SYSTEM_CONFIG_USE_SOCKETS
 

diff --git a/src/inet/InetLayer.cpp b/src/inet/InetLayer.cpp
@@ -544,9 +544,13 @@ INET_ERROR InetLayer::GetLinkLocalAddr(InterfaceId link, IPAddress * llAddr)
  *
  *  @param[in]      ipProto        A protocol within the IP family (e.g., ICMPv4 or ICMPv6).
  *
- *  @param[in,out]  retEndPoint    A pointer to a pointer of the RawEndPoint object that is
- *                                 a return parameter upon completion of the object creation.
- *                                 *retEndPoint is NULL if creation fails.
+ *  @param[in]      onMessageReceived     The endpoint's message reception event handling function delegate.
+ *
+ *  @param[in]      OnReceiveError        The endpoint's receive error event handling function delegate.
+ *
+ *  @param[in,out]  retEndPoint           A pointer to a pointer of the RawEndPoint object that is
+ *                                        a return parameter upon completion of the object creation.
+ *                                        *retEndPoint is NULL if creation fails.
  *
  *  @retval  #INET_ERROR_INCORRECT_STATE  If the InetLayer object is not initialized.
  *  @retval  #INET_ERROR_NO_ENDPOINTS     If the InetLayer RawEndPoint pool is full and no new
@@ -628,9 +632,9 @@ INET_ERROR InetLayer::NewTCPEndPoint(TCPEndPoint ** retEndPoint)
  *    This function gets a free UDPEndPoint object from a pre-allocated pool
  *    and also calls the explicit initializer on the new object.
  *
- *  @param[in,out]  retEndPoint    A pointer to a pointer of the UDPEndPoint object that is
- *                                 a return parameter upon completion of the object creation.
- *                                 *retEndPoint is NULL if creation fails.
+ *  @param[in,out]  retEndPoint           A pointer to a pointer of the UDPEndPoint object that is
+ *                                        a return parameter upon completion of the object creation.
+ *                                        *retEndPoint is NULL if creation fails.
  *
  *  @retval  #INET_ERROR_INCORRECT_STATE  If the InetLayer object is not initialized.
  *  @retval  #INET_ERROR_NO_ENDPOINTS     If the InetLayer UDPEndPoint pool is full and no new

diff --git a/src/inet/InetLayerBasis.h b/src/inet/InetLayerBasis.h
@@ -104,7 +104,7 @@ inline void InetLayerBasis::InitInetLayerBasis(InetLayer & aInetLayer, void * aA
 class SocketEvents
 {
 public:
-    enum
+    enum : uint8_t
     {
         kRead  = 0x01, /**< Bit flag indicating if there is a read event on a socket. */
         kWrite = 0x02, /**< Bit flag indicating if there is a write event on a socket. */

diff --git a/src/inet/RawEndPoint.cpp b/src/inet/RawEndPoint.cpp
@@ -363,6 +363,10 @@ INET_ERROR RawEndPoint::BindIPv6LinkLocal(InterfaceId intfId, const IPAddress &
 /**
  * @brief   Prepare the endpoint to receive ICMP messages.
  *
+ * @param[in]  onMessageReceived   The endpoint's message reception event handling function delegate.
+ * @param[in]  onReceiveError      The endpoint's receive error event handling function delegate.
+ * @param[in]  appState            Application state pointer.
+ *
  * @retval  INET_NO_ERROR   always returned.
  *
  * @details
@@ -374,7 +378,8 @@ INET_ERROR RawEndPoint::BindIPv6LinkLocal(InterfaceId intfId, const IPAddress &
  *  On LwIP, this method must not be called with the LwIP stack lock
  *  already acquired
  */
-INET_ERROR RawEndPoint::Listen()
+INET_ERROR RawEndPoint::Listen(IPEndPointBasis::OnMessageReceivedFunct onMessageReceived,
+                               IPEndPointBasis::OnReceiveErrorFunct onReceiveError, void * appState)
 {
     INET_ERROR res = INET_NO_ERROR;
 
@@ -394,6 +399,10 @@ INET_ERROR RawEndPoint::Listen()
         goto exit;
     }
 
+    OnMessageReceived = onMessageReceived;
+    OnReceiveError    = onReceiveError;
+    AppState          = appState;
+
 #if CHIP_SYSTEM_CONFIG_USE_LWIP
 
     // Lock LwIP stack
@@ -423,6 +432,11 @@ INET_ERROR RawEndPoint::Listen()
     if (res == INET_NO_ERROR)
     {
         mState = kState_Listening;
+
+#if CHIP_SYSTEM_CONFIG_USE_SOCKETS
+        // Wait for ability to read on this endpoint.
+        mRequestIO.SetRead();
+#endif // CHIP_SYSTEM_CONFIG_USE_SOCKETS
     }
 
 exit:
@@ -478,6 +492,9 @@ void RawEndPoint::Close()
         // Clear any results from select() that indicate pending I/O for the socket.
         mPendingIO.Clear();
 
+        // Do not wait for I/O on this endpoint.
+        mRequestIO.Clear();
+
 #endif // CHIP_SYSTEM_CONFIG_USE_SOCKETS
 
         mState = kState_Closed;
@@ -1049,11 +1066,6 @@ INET_ERROR RawEndPoint::GetSocket(IPAddressType aAddressType)
     return (lRetval);
 }
 
-SocketEvents RawEndPoint::PrepareIO()
-{
-    return (IPEndPointBasis::PrepareIO());
-}
-
 void RawEndPoint::HandlePendingIO()
 {
     if (mState == kState_Listening && OnMessageReceived != nullptr && mPendingIO.IsReadable())

diff --git a/src/inet/RawEndPoint.h b/src/inet/RawEndPoint.h
@@ -74,7 +74,8 @@ class DLL_EXPORT RawEndPoint : public IPEndPointBasis
     INET_ERROR BindIPv6LinkLocal(InterfaceId intfId, const IPAddress & addr);
     INET_ERROR BindInterface(IPAddressType addrType, InterfaceId intfId);
     InterfaceId GetBoundInterface();
-    INET_ERROR Listen();
+    INET_ERROR Listen(IPEndPointBasis::OnMessageReceivedFunct onMessageReceived,
+                      IPEndPointBasis::OnReceiveErrorFunct onReceiveError, void * appState = nullptr);
     INET_ERROR SendTo(const IPAddress & addr, chip::System::PacketBufferHandle && msg, uint16_t sendFlags = 0);
     INET_ERROR SendTo(const IPAddress & addr, InterfaceId intfId, chip::System::PacketBufferHandle && msg, uint16_t sendFlags = 0);
     INET_ERROR SendMsg(const IPPacketInfo * pktInfo, chip::System::PacketBufferHandle msg, uint16_t sendFlags = 0);
@@ -107,7 +108,6 @@ class DLL_EXPORT RawEndPoint : public IPEndPointBasis
 
 #if CHIP_SYSTEM_CONFIG_USE_SOCKETS
     INET_ERROR GetSocket(IPAddressType addrType);
-    SocketEvents PrepareIO();
     void HandlePendingIO();
 #endif // CHIP_SYSTEM_CONFIG_USE_SOCKETS
 };

diff --git a/src/inet/TCPEndPoint.cpp b/src/inet/TCPEndPoint.cpp
@@ -245,7 +245,9 @@ INET_ERROR TCPEndPoint::Bind(IPAddressType addrType, const IPAddress & addr, uin
 #endif // CHIP_SYSTEM_CONFIG_USE_SOCKETS
 
     if (res == INET_NO_ERROR)
+    {
         State = kState_Bound;
+    }
 
     return res;
 }
@@ -254,10 +256,6 @@ INET_ERROR TCPEndPoint::Listen(uint16_t backlog)
 {
     INET_ERROR res = INET_NO_ERROR;
 
-#if CHIP_SYSTEM_CONFIG_USE_SOCKETS
-    chip::System::Layer & lSystemLayer = SystemLayer();
-#endif // CHIP_SYSTEM_CONFIG_USE_SOCKETS
-
     if (State != kState_Bound)
         return INET_ERROR_INCORRECT_STATE;
 
@@ -278,8 +276,11 @@ INET_ERROR TCPEndPoint::Listen(uint16_t backlog)
     if (listen(mSocket, backlog) != 0)
         res = chip::System::MapErrorPOSIX(errno);
 
+    // Wait for ability to read on this endpoint.
+    mRequestIO.SetRead();
+
     // Wake the thread calling select so that it recognizes the new socket.
-    lSystemLayer.WakeSelect();
+    SystemLayer().WakeSelect();
 
 #endif // CHIP_SYSTEM_CONFIG_USE_SOCKETS
 
@@ -493,11 +494,17 @@ INET_ERROR TCPEndPoint::Connect(const IPAddress & addr, uint16_t port, Interface
     if (conRes == 0)
     {
         State = kState_Connected;
+        // Wait for ability to read on this endpoint.
+        mRequestIO.SetRead();
         if (OnConnectComplete != nullptr)
             OnConnectComplete(this, INET_NO_ERROR);
     }
     else
+    {
         State = kState_Connecting;
+        // Wait for ability to write on this endpoint.
+        mRequestIO.SetWrite();
+    }
 
     // Wake the thread calling select so that it recognizes the new socket.
     lSystemLayer.WakeSelect();
@@ -699,9 +706,17 @@ INET_ERROR TCPEndPoint::Send(System::PacketBufferHandle data, bool push)
     }
 
     if (mSendQueue.IsNull())
+    {
         mSendQueue = std::move(data);
+#if CHIP_SYSTEM_CONFIG_USE_SOCKETS
+        // Wait for ability to write on this endpoint.
+        mRequestIO.SetWrite();
+#endif // CHIP_SYSTEM_CONFIG_USE_SOCKETS
+    }
     else
+    {
         mSendQueue->AddToEnd(std::move(data));
+    }
 
 #if CHIP_SYSTEM_CONFIG_USE_LWIP
 
@@ -1318,9 +1333,18 @@ INET_ERROR TCPEndPoint::DriveSending()
         MarkActive();
 
         if (lenSent < bufLen)
+        {
             mSendQueue->ConsumeHead(lenSent);
+        }
         else
+        {
             mSendQueue.FreeHead();
+            if (mSendQueue.IsNull())
+            {
+                // Do not wait for ability to write on this endpoint.
+                mRequestIO.ClearWrite();
+            }
+        }
 
         if (OnDataSent != nullptr)
             OnDataSent(this, lenSent);
@@ -1418,6 +1442,13 @@ void TCPEndPoint::HandleConnectComplete(INET_ERROR err)
         MarkActive();
 
         State = kState_Connected;
+
+#if CHIP_SYSTEM_CONFIG_USE_SOCKETS
+        // Wait for ability to read or write on this endpoint.
+        mRequestIO.SetRead();
+        mRequestIO.SetWrite();
+#endif // CHIP_SYSTEM_CONFIG_USE_SOCKETS
+
         if (OnConnectComplete != nullptr)
             OnConnectComplete(this, INET_NO_ERROR);
     }
@@ -1547,6 +1578,7 @@ INET_ERROR TCPEndPoint::DoClose(INET_ERROR err, bool suppressCallback)
             if (close(mSocket) != 0 && err == INET_NO_ERROR)
                 err = chip::System::MapErrorPOSIX(errno);
             mSocket = INET_INVALID_SOCKET_FD;
+            mRequestIO.Clear();
 
             // Wake the thread calling select so that it recognizes the socket is closed.
             lSystemLayer.WakeSelect();
@@ -2335,26 +2367,6 @@ INET_ERROR TCPEndPoint::GetSocket(IPAddressType addrType)
     return INET_NO_ERROR;
 }
 
-SocketEvents TCPEndPoint::PrepareIO()
-{
-    SocketEvents ioType;
-
-    // If initiating a new connection...
-    // OR if connected and there is data to be sent...
-    // THEN arrange for the kernel to alert us when the socket is ready to be written.
-    if (State == kState_Connecting || (IsConnected() && !mSendQueue.IsNull()))
-        ioType.SetWrite();
-
-    // If listening for incoming connections and the app is ready to receive a connection...
-    // OR if in a state where receiving is allowed, and the app is ready to receive data...
-    // THEN arrange for the kernel to alert us when the socket is ready to be read.
-    if ((State == kState_Listening && OnConnectionReceived != nullptr) ||
-        ((State == kState_Connected || State == kState_SendShutdown) && ReceiveEnabled && OnDataReceived != nullptr))
-        ioType.SetRead();
-
-    return ioType;
-}
-
 void TCPEndPoint::HandlePendingIO()
 {
     // Prevent the end point from being freed while in the middle of a callback.
@@ -2504,7 +2516,8 @@ void TCPEndPoint::ReceiveData()
                 State = kState_ReceiveShutdown;
             else
                 State = kState_Closing;
-
+            // Do not wait for ability to read on this endpoint.
+            mRequestIO.ClearRead();
             // Call the app's OnPeerClose.
             if (OnPeerClose != nullptr)
                 OnPeerClose(this);
@@ -2601,6 +2614,9 @@ void TCPEndPoint::HandleIncomingConnection()
 #endif // !INET_CONFIG_ENABLE_IPV4
         conEP->Retain();
 
+        // Wait for ability to read on this endpoint.
+        conEP->mRequestIO.SetRead();
+
         // Call the app's callback function.
         OnConnectionReceived(this, conEP, peerAddr, peerPort);
     }

diff --git a/src/inet/TCPEndPoint.h b/src/inet/TCPEndPoint.h
@@ -673,7 +673,7 @@ class DLL_EXPORT TCPEndPoint : public EndPointBasis
 
 #if CHIP_SYSTEM_CONFIG_USE_SOCKETS
     INET_ERROR GetSocket(IPAddressType addrType);
-    SocketEvents PrepareIO();
+    SocketEvents PrepareIO() const { return mRequestIO; }
     void HandlePendingIO();
     void ReceiveData();
     void HandleIncomingConnection();