Bugfix: Fix interrupted network read operation

source/core_mqtt.c

@@ -71,17 +71,25 @@ static uint32_t calculateElapsedTime( uint32_t later,
 static MQTTPubAckType_t getAckFromPacketType( uint8_t packetType );
 
 /**
- * @brief Receive bytes into the network buffer, with a timeout.
+ * @brief Receive bytes into the network buffer.
  *
  * @param[in] pContext Initialized MQTT Context.
  * @param[in] bytesToRecv Number of bytes to receive.
- * @param[in] timeoutMs Time remaining to receive the packet.
+ *
+ * @note This operation calls the transport receive function
+ * repeatedly to read bytes from the network until either:
+ * 1. The requested number of bytes @a bytesToRecv are read.
+ *                    OR
+ * 2. No data is received from the network for MQTT_RECV_POLLING_TIMEOUT_MS duration.
+ *
+ *                    OR
+ * 3. There is an error in reading from the network.
+ *
  *
  * @return Number of bytes received, or negative number on network error.
  */
 static int32_t recvExact( const MQTTContext_t * pContext,
-                          size_t bytesToRecv,
-                          uint32_t timeoutMs );
+                          size_t bytesToRecv );
 
 /**
  * @brief Discard a packet from the transport interface.
@@ -683,13 +691,12 @@ static MQTTPubAckType_t getAckFromPacketType( uint8_t packetType )
 /*-----------------------------------------------------------*/
 
 static int32_t recvExact( const MQTTContext_t * pContext,
-                          size_t bytesToRecv,
-                          uint32_t timeoutMs )
+                          size_t bytesToRecv )
 {
     uint8_t * pIndex = NULL;
     size_t bytesRemaining = bytesToRecv;
     int32_t totalBytesRecvd = 0, bytesRecvd;
-    uint32_t entryTimeMs = 0U, elapsedTimeMs = 0U;
+    uint32_t lastDataRecvTimeMs = 0U, timeSinceLastDataWasRecvdMs = 0U;
     TransportRecv_t recvFunc = NULL;
     MQTTGetCurrentTimeFunc_t getTimeStampMs = NULL;
     bool receiveError = false;
@@ -704,7 +711,8 @@ static int32_t recvExact( const MQTTContext_t * pContext,
     recvFunc = pContext->transportInterface.recv;
     getTimeStampMs = pContext->getTime;
 
-    entryTimeMs = getTimeStampMs();
+    /* Part of the MQTT packet has been read before calling this function. */
+    lastDataRecvTimeMs = getTimeStampMs();
 
     while( ( bytesRemaining > 0U ) && ( receiveError == false ) )
     {
@@ -719,8 +727,11 @@ static int32_t recvExact( const MQTTContext_t * pContext,
             totalBytesRecvd = bytesRecvd;
             receiveError = true;
         }
-        else
+        else if( bytesRecvd > 0 )
         {
+            /* Reset the starting time as we have received some data from the network. */
+            lastDataRecvTimeMs = getTimeStampMs();
+
             /* It is a bug in the application's transport receive implementation
              * if more bytes than expected are received. To avoid a possible
              * overflow in converting bytesRemaining from unsigned to signed,
@@ -737,10 +748,14 @@ static int32_t recvExact( const MQTTContext_t * pContext,
                         ( unsigned long ) bytesRemaining,
                         ( long int ) totalBytesRecvd ) );
         }
+        else
+        {
+            /* No bytes were read from the network. */
+            timeSinceLastDataWasRecvdMs = calculateElapsedTime( getTimeStampMs(), lastDataRecvTimeMs );
+        }
 
-        elapsedTimeMs = calculateElapsedTime( getTimeStampMs(), entryTimeMs );
-
-        if( ( bytesRemaining > 0U ) && ( elapsedTimeMs >= timeoutMs ) )
+        if( ( bytesRemaining > 0U ) &&
+            ( timeSinceLastDataWasRecvdMs >= MQTT_RECV_POLLING_TIMEOUT_MS ) )
         {
             LogError( ( "Time expired while receiving packet." ) );
             receiveError = true;
@@ -760,7 +775,6 @@ static MQTTStatus_t discardPacket( const MQTTContext_t * pContext,
     int32_t bytesReceived = 0;
     size_t bytesToReceive = 0U;
     uint32_t totalBytesReceived = 0U, entryTimeMs = 0U, elapsedTimeMs = 0U;
-    uint32_t remainingTimeMs = timeoutMs;
     MQTTGetCurrentTimeFunc_t getTimeStampMs = NULL;
     bool receiveError = false;
 
@@ -779,7 +793,7 @@ static MQTTStatus_t discardPacket( const MQTTContext_t * pContext,
             bytesToReceive = remainingLength - totalBytesReceived;
         }
 
-        bytesReceived = recvExact( pContext, bytesToReceive, remainingTimeMs );
+        bytesReceived = recvExact( pContext, bytesToReceive );
 
         if( bytesReceived != ( int32_t ) bytesToReceive )
         {
@@ -795,12 +809,8 @@ static MQTTStatus_t discardPacket( const MQTTContext_t * pContext,
 
             elapsedTimeMs = calculateElapsedTime( getTimeStampMs(), entryTimeMs );
 
-            /* Update remaining time and check for timeout. */
-            if( elapsedTimeMs < timeoutMs )
-            {
-                remainingTimeMs = timeoutMs - elapsedTimeMs;
-            }
-            else
+            /* Check for timeout. */
+            if( elapsedTimeMs >= timeoutMs )
             {
                 LogError( ( "Time expired while discarding packet." ) );
                 receiveError = true;
@@ -846,7 +856,7 @@ static MQTTStatus_t receivePacket( const MQTTContext_t * pContext,
     else
     {
         bytesToReceive = incomingPacket.remainingLength;
-        bytesReceived = recvExact( pContext, bytesToReceive, remainingTimeMs );
+        bytesReceived = recvExact( pContext, bytesToReceive );
 
         if( bytesReceived == ( int32_t ) bytesToReceive )
         {

source/include/core_mqtt_config_defaults.h

@@ -108,6 +108,27 @@
     #define MQTT_PINGRESP_TIMEOUT_MS    ( 500U )
 #endif
 
+/**
+ * @brief The maximum duration between non-empty network reads while
+ * receiving an MQTT packet via the #MQTT_ProcessLoop or #MQTT_ReceiveLoop
+ * API functions.
+ *
+ * When an incoming MQTT packet is detected, the transport receive function
+ * may be called multiple times until all of the expected number of bytes of the
+ * packet are received. This timeout represents the maximum polling duration that
+ * is allowed without any data reception from the network for the incoming packet.
+ * If the timeout expires, the #MQTT_ProcessLoop and #MQTT_ReceiveLoop functions
+ * return #MQTTRecvFailed.
+ *
+ * <b>Possible values:</b> Any positive integer up to SIZE_MAX. Recommended to
+ * use a small timeout value. <br>
+ * <b>Default value:</b> `10`
+ *
+ */
+#ifndef MQTT_RECV_POLLING_TIMEOUT_MS
+    #define MQTT_RECV_POLLING_TIMEOUT_MS    ( 10U )
+#endif
+
 /**
  * @brief Macro that is called in the MQTT library for logging "Error" level
  * messages.

test/cbmc/include/core_mqtt_config.h

@@ -70,4 +70,18 @@ struct NetworkContext
  */
 #define MQTT_PINGRESP_TIMEOUT_MS                ( 500U )
 
+/**
+ * @brief The maximum duration of receiving no data over network when
+ * attempting to read an incoming MQTT packet by the #MQTT_ProcessLoop or
+ * #MQTT_ReceiveLoop API functions.
+ *
+ * When an incoming MQTT packet is detected, the transport receive function
+ * may be called multiple times until all the expected number of bytes for the
+ * packet are received. This timeout represents the maximum duration of polling
+ * for any data to be received over the network for the incoming.
+ * If the timeout expires, the #MQTT_ProcessLoop or #MQTT_ReceiveLoop functions
+ * return #MQTTRecvFailed.
+ */
+#define MQTT_RECV_POLLING_TIMEOUT_MS            ( 3U )
+
 #endif /* ifndef CORE_MQTT_CONFIG_H_ */

test/unit-test/core_mqtt_utest.c

@@ -132,6 +132,7 @@ typedef struct ProcessLoopReturns
     MQTTStatus_t processLoopStatus;           /**< @brief Return value of the process loop. */
     bool incomingPublish;                     /**< @brief Whether the incoming packet is a publish. */
     MQTTPublishInfo_t * pPubInfo;             /**< @brief Publish information to be returned by the deserializer. */
+    uint32_t timeoutMs;                       /**< @brief The timeout value to call MQTT_ProcessLoop API with. */
 } ProcessLoopReturns_t;
 
 /**
@@ -365,6 +366,20 @@ static int32_t transportRecvOneByte( NetworkContext_t * pNetworkContext,
     return 1;
 }
 
+/**
+ * @brief Mocked transport returning zero bytes to simulate reception
+ * of no data over network.
+ */
+static int32_t transportRecvNoData( NetworkContext_t * pNetworkContext,
+                                    void * pBuffer,
+                                    size_t bytesToRead )
+{
+    ( void ) pNetworkContext;
+    ( void ) pBuffer;
+    ( void ) bytesToRead;
+    return 0;
+}
+
 /**
  * @brief Initialize the transport interface with the mocked functions for
  * send and receive.
@@ -405,6 +420,7 @@ static void resetProcessLoopParams( ProcessLoopReturns_t * pExpectParams )
     pExpectParams->processLoopStatus = MQTTSuccess;
     pExpectParams->incomingPublish = false;
     pExpectParams->pPubInfo = NULL;
+    pExpectParams->timeoutMs = MQTT_NO_TIMEOUT_MS;
 }
 
 /**
@@ -548,7 +564,7 @@ static void expectProcessLoopCalls( MQTTContext_t * const pContext,
     }
 
     /* Expect the above calls when running MQTT_ProcessLoop. */
-    mqttStatus = MQTT_ProcessLoop( pContext, MQTT_NO_TIMEOUT_MS );
+    mqttStatus = MQTT_ProcessLoop( pContext, pExpectParams->timeoutMs );
     TEST_ASSERT_EQUAL( processLoopStatus, mqttStatus );
 
     /* Any final assertions to end the test. */
@@ -853,7 +869,7 @@ void test_MQTT_Connect_partial_receive()
 
     setupTransportInterface( &transport );
     setupNetworkBuffer( &networkBuffer );
-    transport.recv = transportRecvOneByte;
+    transport.recv = transportRecvNoData;
 
     memset( &mqttContext, 0x0, sizeof( mqttContext ) );
     MQTT_Init( &mqttContext, &transport, getTime, eventCallback, &networkBuffer );
@@ -864,14 +880,16 @@ void test_MQTT_Connect_partial_receive()
     incomingPacket.type = MQTT_PACKET_TYPE_CONNACK;
     incomingPacket.remainingLength = 2;
 
-    /* Not enough time to receive entire packet, for branch coverage. This is due
-     * to the fact the mocked receive function reads only one byte at a time. */
-    timeout = 1;
+    /* Timeout in receiving entire packet, for branch coverage. This is due to the fact that the mocked
+     * receive function always returns 0 bytes read. */
     MQTT_GetIncomingPacketTypeAndLength_ExpectAnyArgsAndReturn( MQTTSuccess );
     MQTT_GetIncomingPacketTypeAndLength_ReturnThruPtr_pIncomingPacket( &incomingPacket );
     status = MQTT_Connect( &mqttContext, &connectInfo, NULL, timeout, &sessionPresent );
     TEST_ASSERT_EQUAL_INT( MQTTRecvFailed, status );
 
+    /* Update to use mock receive function that receives one byte at a time for the
+     * rest of the test. */
+    mqttContext.transportInterface.recv = transportRecvOneByte;
     timeout = 10;
 
     /* Not enough space for packet, discard it. */
@@ -1573,6 +1591,53 @@ void test_MQTT_ProcessLoop_handleIncomingPublish_Error_Paths( void )
     TEST_ASSERT_FALSE( isEventCallbackInvoked );
 }
 
+/**
+ * @brief This test checks that the ProcessLoop API function is able to
+ * support receiving an entire incoming MQTT packet over the network when
+ * the transport recv function only reads less than requested bytes at a
+ * time, and the timeout passed to the API is "0ms".
+ */
+void test_MQTT_ProcessLoop_Zero_Duration_And_Partial_Network_Read( void )
+{
+    MQTTStatus_t mqttStatus;
+    MQTTContext_t context;
+    TransportInterface_t transport;
+    MQTTFixedBuffer_t networkBuffer;
+    ProcessLoopReturns_t expectParams = { 0 };
+
+    setupNetworkBuffer( &networkBuffer );
+
+    transport.send = transportSendSuccess;
+
+    /* Set the transport recv function for the test to the mock function that represents
+     * partial read of data from network (i.e. less than requested number of bytes)
+     * at a time. */
+    transport.recv = transportRecvOneByte;
+
+    /* Initialize the context. */
+    mqttStatus = MQTT_Init( &context, &transport, getTime, eventCallback, &networkBuffer );
+    TEST_ASSERT_EQUAL( MQTTSuccess, mqttStatus );
+
+    /* Set flag required for configuring behavior of expectProcessLoopCalls()
+     * helper function. */
+    modifyIncomingPacketStatus = MQTTSuccess;
+
+    /* Test the ProcessLoop() call with zero duration timeout to verify that it
+     * will be able to support reading the packet over network over multiple calls to
+     * the transport receive function. */
+    expectParams.timeoutMs = MQTT_NO_TIMEOUT_MS;
+
+    /* Test with an incoming PUBLISH packet whose payload is read only one byte
+     * per call to the transport recv function. */
+    currentPacketType = MQTT_PACKET_TYPE_PUBLISH;
+    /* Set expected return values during the loop. */
+    resetProcessLoopParams( &expectParams );
+    expectParams.stateAfterDeserialize = MQTTPubAckSend;
+    expectParams.stateAfterSerialize = MQTTPublishDone;
+    expectParams.incomingPublish = true;
+    expectProcessLoopCalls( &context, &expectParams );
+}
+
 /**
  * @brief This test case covers all calls to the private method,
  * handleIncomingAck(...),