Fix repair transmission flagging

CMakeLists.txt

@@ -12,6 +12,17 @@ set(COMMON src/common)
 option(NORM_BUILD_EXAMPLES "Enables building of the examples in /examples." OFF)
 set(NORM_CUSTOM_PROTOLIB_VERSION OFF CACHE STRING "Set a custom protolib version to use, ./protolib to use the local version")
 
+# Build self-checking tests by default only when NORM is the top-level project.
+if(CMAKE_SOURCE_DIR STREQUAL CMAKE_CURRENT_SOURCE_DIR)
+    set(NORM_BUILD_TESTING_DEFAULT ON)
+else()
+    set(NORM_BUILD_TESTING_DEFAULT OFF)
+endif()
+option(NORM_BUILD_TESTING "Enables building NORM self-checking tests." ${NORM_BUILD_TESTING_DEFAULT})
+if(NORM_BUILD_TESTING)
+    enable_testing()
+endif()
+
 include(CheckCXXSymbolExists)
 check_cxx_symbol_exists(dirfd "dirent.h" HAVE_DIRFD)
 if(HAVE_DIRFD)
@@ -224,3 +235,20 @@ if(NORM_BUILD_EXAMPLES)
     endforeach()
 endif()
 
+if(NORM_BUILD_TESTING)
+    # This test exercises internal sender repair state and message generation,
+    # so compile it with the implementation sources instead of depending on
+    # symbols hidden by shared-library builds.
+    add_executable(normRepairFlagTest
+        tests/normRepairFlagTest.cpp
+        ${PLATFORM_SOURCE_FILES}
+        ${COMMON_SOURCE_FILES})
+    target_link_libraries(normRepairFlagTest PRIVATE protokit::protokit ${PLATFORM_LIBS})
+    target_compile_definitions(normRepairFlagTest PRIVATE ${PLATFORM_DEFINITIONS})
+    target_compile_options(normRepairFlagTest PRIVATE ${PLATFORM_FLAGS})
+    target_include_directories(normRepairFlagTest PRIVATE
+        ${CMAKE_CURRENT_BINARY_DIR}
+        ${CMAKE_CURRENT_SOURCE_DIR}/include)
+    add_test(NAME normRepairFlagTest COMMAND normRepairFlagTest)
+    set_tests_properties(normRepairFlagTest PROPERTIES TIMEOUT 15)
+endif()

TODO.TXT

@@ -1,7 +1,5 @@
 TODO:
 
-0) Fix implementation of NORM_DATA_FLAG_REPAIR.
-
 2) Add mechanism to allow window-based control of NORM server
    transmission rate.  
 

VERSION.TXT

@@ -4,6 +4,9 @@
 Version 1.5.10 (in progress)
 =============
     - NormSocket API improvements
+    - Corrected repair flagging for reactive NORM_DATA and NORM_INFO
+      transmissions so receiver synchronization policy can distinguish
+      repair-only traffic from first-pass content.
 
 Version 1.5.9
 =============

include/normObject.h

@@ -89,12 +89,14 @@ class NormObject
             info_ptr = NULL;
             info_len = 0;
             pending_info = false;
+            pending_info_repair = false;
         }
         // This is used to bootstrap reception sender
         // is using FTI_INFO FtiMode
         void SetPendingInfo(bool state, UINT8 fecId)
         {
             pending_info = true;
+            pending_info_repair = false;
             fec_id = fecId;
         }
 
@@ -318,6 +320,7 @@ class NormObject
         UINT16                nparity;
         NormBlockBuffer       block_buffer;
         bool                  pending_info;  // set when we need to send or recv info
+        bool                  pending_info_repair;  // sender: pending INFO is reactive repair
         ProtoSlidingMask      pending_mask;
         bool                  repair_info;   // receiver: set when
         ProtoSlidingMask      repair_mask;

include/normSegment.h

@@ -114,6 +114,7 @@ class NormBlock
             parity_offset = autoParity;  
             flags = 0;
             seg_size_max = 0;
+            TxResetOriginalPass(ndata, autoParity);
             last_nack_time.GetCurrentTime();
         }
         void TxRecover(NormBlockId& blockId, UINT16 ndata, UINT16 nparity)
@@ -126,6 +127,32 @@ class NormBlock
             parity_offset = nparity; // explicit repair mode ???  
             flags = IN_REPAIR;
             seg_size_max = 0;
+            original_tx_next = original_tx_end = ndata;
+        }
+        void TxResetOriginalPass(UINT16 ndata, UINT16 autoParity)
+        {
+            ASSERT((ndata + autoParity) <= size);
+            original_tx_next = 0;
+            original_tx_end = ndata + autoParity;
+        }
+        void TxCompleteOriginalPass()
+        {
+            original_tx_next = original_tx_end;
+        }
+        bool TxIsOriginalPending(NormSegmentId segmentId) const
+        {
+            return ((segmentId >= original_tx_next) &&
+                    (segmentId < original_tx_end));
+        }
+        void TxAdvanceOriginal(NormSegmentId segmentId)
+        {
+            if (TxIsOriginalPending(segmentId))
+            {
+                // Sender selection is ordered, so first-pass symbols cannot
+                // skip ahead of this frontier.
+                ASSERT(segmentId == original_tx_next);
+                original_tx_next = segmentId + 1;
+            }
         }
         bool TxReset(UINT16 ndata, UINT16 nparity, UINT16 autoParity, 
                      UINT16 segmentSize);
@@ -164,6 +191,7 @@ class NormBlock
             parity_count = 0;
             parity_offset = 0;
             flags = 0;
+            original_tx_next = original_tx_end = 0;
         }
         // Note: This invalidates the repair_mask state.
         bool IsRepairPending(UINT16 ndata, UINT16 nparity); 
@@ -279,6 +307,8 @@ class NormBlock
         UINT16       parity_count;  // how many fresh parity we are currently planning to send
         UINT16       parity_offset; // offset from where our fresh parity will be sent
         UINT16       seg_size_max;
+        NormSegmentId original_tx_next; // first unsent symbol in the original pass
+        NormSegmentId original_tx_end;  // source data plus proactive parity
 
         ProtoBitmask pending_mask;
         ProtoBitmask repair_mask;

src/common/normObject.cpp

@@ -12,7 +12,7 @@ NormObject::NormObject(NormObject::Type      theType,
                        class NormSenderNode* theSender,
                        const NormObjectId&   transportId)
  : type(theType), session(theSession), sender(theSender), reference_count(1),
-   transport_id(transportId), segment_size(0), pending_info(false), repair_info(false),
+   transport_id(transportId), segment_size(0), pending_info(false), pending_info_repair(false), repair_info(false),
    current_block_id(0), next_segment_id(0), 
    max_pending_block(0), max_pending_segment(0),
    info_ptr(NULL), info_len(0), first_pass(true), accepted(false), notify_on_update(true),
@@ -82,6 +82,7 @@ bool NormObject::Open(const NormObjectSize& objectSize,
                       UINT16                numData,
                       UINT16                numParity)
 {
+    pending_info_repair = false;
     // Note "objectSize" represents actual total object size for
     // DATA or FILE objects, buffer size for STREAM objects
     // In either case, we need our sliding bit masks to be of 
@@ -301,19 +302,16 @@ bool NormObject::HandleInfoRequest(bool holdoff)
     bool increasedRepair = false;
     if (info_ptr || (NormSession::FTI_INFO == session.SenderFtiMode()))
     {
-        if (!repair_info)
+        if (!repair_info && !pending_info)
         {
             increasedRepair = true;
             if (holdoff)
             {
-                if (pending_info)
-                    increasedRepair = false;
-                else
-                    pending_info = true;
+                pending_info = true;
+                pending_info_repair = true;
             }
             else
             {
-                pending_info = true;  // does this really need to be done?
                 repair_info = true;
             }
         }   
@@ -392,7 +390,9 @@ bool NormObject::TxReset(NormBlockId firstBlock, bool requeue)
     {
         increasedRepair = true;
         pending_info = true;
+        pending_info_repair = !requeue;
     }
+    if (requeue) pending_info_repair = false;
     repair_info = false;
     repair_mask.Reset(firstBlock.GetValue());
     repair_mask.Xor(pending_mask);
@@ -415,7 +415,12 @@ bool NormObject::TxReset(NormBlockId firstBlock, bool requeue)
                                               nparity, 
                                               session.SenderAutoParity(), 
                                               segment_size);
-            if (requeue) block->ClearFlag(NormBlock::IN_REPAIR);  // since we're requeuing
+            if (requeue)
+            {
+                block->ClearFlag(NormBlock::IN_REPAIR);  // since we're requeuing
+                block->TxResetOriginalPass(GetBlockSize(blockId),
+                                           session.SenderAutoParity());
+            }
         }
     }
     if (requeue) 
@@ -472,10 +477,11 @@ bool NormObject::TxUpdateBlock(NormBlock*       theBlock,
 bool NormObject::ActivateRepairs()
 {
     bool repairsActivated = false;
-    // Activate repair of info if applicable (TBD - how to flag info message as repair???)
+    // Activate repair of info if applicable.
     if (repair_info)
     {
         pending_info = true;
+        pending_info_repair = true;
         repair_info = false;
         repairsActivated = true;
     }
@@ -1857,10 +1863,12 @@ bool NormObject::NextSenderMsg(NormObjectMsg* msg)
     }
     if (pending_info)
     {
-        // (TBD) set REPAIR_FLAG for retransmitted info
+        if (pending_info_repair)
+            msg->SetFlag(NormObjectMsg::FLAG_REPAIR);
         NormInfoMsg* infoMsg = static_cast<NormInfoMsg*>(msg);
         infoMsg->SetInfo(info_ptr, info_len);
         pending_info = false;
+        pending_info_repair = false;
         return true;
     }
     // This block gets the next pending block/segment
@@ -1930,7 +1938,10 @@ bool NormObject::NextSenderMsg(NormObjectMsg* msg)
            block->TxInit(blockId, numData, session.SenderAutoParity());  
            //if (blockId < max_pending_block) 
            if (Compare(blockId, max_pending_block) < 0)
+           {
                block->SetFlag(NormBlock::IN_REPAIR);
+               block->TxCompleteOriginalPass();
+           }
            while (!block_buffer.Insert(block))
            {
                //ASSERT(STREAM == type);
@@ -2013,15 +2024,19 @@ bool NormObject::NextSenderMsg(NormObjectMsg* msg)
                 {
                     PLOG(PL_ERROR, "NormObject::NextSenderMsg() node>%lu Warning! can't repair old stream segment\n", 
                                     (unsigned long)LocalNodeId());
-                    block->UnsetPending(segmentId); 
+                    block->UnsetPending(segmentId);
+                    block->TxAdvanceOriginal(segmentId);
                     if (!block->IsPending())
                     {
                         // End of old block reached
                         block->ResetParityCount(nparity);
                         pending_mask.Unset(blockId.GetValue()); 
                         // for EMCON sending, mark NORM_INFO for re-transmission, if applicable
                         if (session.SndrEmcon() && (HaveInfo() || (NormSession::FTI_INFO == session.SenderFtiMode())))
+                        {
+                            if (!pending_info) pending_info_repair = false;
                             pending_info = true;
+                        }
                     }
                     block = NULL;
                     continue;  //return NextSenderMsg(msg);
@@ -2072,9 +2087,11 @@ bool NormObject::NextSenderMsg(NormObjectMsg* msg)
 #endif // if/else SIMULATE
         }
     }  // end while (NULL == block)
-    block->UnsetPending(segmentId); 
-    //if (block->InRepair()) 
-    //    data->SetFlag(NormObjectMsg::FLAG_REPAIR);
+    const bool originalTransmission = block->TxIsOriginalPending(segmentId);
+    block->UnsetPending(segmentId);
+    block->TxAdvanceOriginal(segmentId);
+    if (block->InRepair() && !originalTransmission)
+        data->SetFlag(NormObjectMsg::FLAG_REPAIR);
     data->SetFecPayloadId(fec_id, blockId.GetValue(), segmentId, numData, fec_m);
     if (!block->IsPending()) 
     {
@@ -2083,7 +2100,10 @@ bool NormObject::NextSenderMsg(NormObjectMsg* msg)
         pending_mask.Unset(blockId.GetValue()); 
         // for EMCON sending, mark NORM_INFO for re-transmission, if applicable
         if (session.SndrEmcon() && (HaveInfo() || (NormSession::FTI_INFO == session.SenderFtiMode())))
+        {
+            if (!pending_info) pending_info_repair = false;
             pending_info = true;
+        }
         // Advance sender use of "max_pending_block" so we always
         // know when a block should be flagged as IN_REPAIR
         if (blockId == max_pending_block)

src/common/normSegment.cpp

@@ -90,7 +90,8 @@ char* NormSegmentPool::Get()
 // NormBlock Implementation
 
 NormBlock::NormBlock()
- : size(0), segment_table(NULL), erasure_count(0), parity_count(0), next(NULL)
+ : size(0), segment_table(NULL), erasure_count(0), parity_count(0),
+   original_tx_next(0), original_tx_end(0), next(NULL)
 {
 }     
 
@@ -125,6 +126,7 @@ bool NormBlock::Init(UINT16 totalSize)
     parity_count = 0;
     parity_offset = 0;
     seg_size_max = 0;
+    original_tx_next = original_tx_end = 0;
     return true;
 }  // end NormBlock::Init()
 
@@ -144,6 +146,7 @@ void NormBlock::Destroy()
         segment_table = (char**)NULL;
     }
     erasure_count = parity_count = size = 0;
+    original_tx_next = original_tx_end = 0;
 }  // end NormBlock::Destroy()
 
 void NormBlock::EmptyToPool(NormSegmentPool& segmentPool)
@@ -335,6 +338,7 @@ bool NormBlock::TxUpdate(NormSegmentId nextId, NormSegmentId lastId,
             } 
         }   
     }
+    if (increasedRepair) SetFlag(IN_REPAIR);
     return increasedRepair;
 }  // end NormBlock::TxUpdate()