support partial rateless decode with re-NACK + enhance mock

examples/normRatelessTest.cpp

@@ -1,40 +1,128 @@
 #include <normApi.h>
 #include <normEncoder.h>
+
 #include <stdio.h>
 #include <string.h>
 #include <stdlib.h>
+#include <time.h>
+#include <sys/select.h>
+#include <sys/time.h>
 
-class MockRatelessEncoder : public NormEncoder {
+// FEC ID for the (partially-specified) rateless code family.
-public:
+static const UINT8 RATELESS_FEC_ID = 131;
-    virtual bool Init(unsigned int numData, unsigned int numParity, UINT16 vectorSize) {
+
-        ndata = numData;
+// Deterministic permutation of [0, n) shared by encoder and decoder.
-        vec_size = vectorSize;
+// The seed depends only on "n", so both sides produce the identical ordering.
-        return true;
+static void BuildRatelessPermutation(unsigned int* perm, unsigned int n)
+{
+    for (unsigned int i = 0; i < n; i++) perm[i] = i;
+    unsigned int state = 0x9e3779b9u ^ n;
+    for (unsigned int i = n; i > 1; i--)
+    {
+        state = state * 1103515245u + 12345u;
+        unsigned int j = (state >> 8) % i;
+        unsigned int t = perm[i - 1]; perm[i - 1] = perm[j]; perm[j] = t;
     }
-    virtual void Destroy() {}
+}
-    virtual void Encode(unsigned int segmentId, const char* dataVector, char** parityVectorList) {}
-    virtual bool IsRateless() const { return true; }
 
-private:
+class MockRatelessEncoder : public NormEncoder
-    unsigned int ndata;
+{
-    UINT16 vec_size;
+    public:
+        MockRatelessEncoder() : ndata(0), vec_size(0), perm(NULL) {}
+        virtual ~MockRatelessEncoder() { Destroy(); }
+
+        virtual bool Init(unsigned int numData, unsigned int /*numParity*/, UINT16 vectorSize)
+        {
+            Destroy();
+            ndata = numData;
+            vec_size = vectorSize;
+            perm = new unsigned int[ndata ? ndata : 1];
+            return (NULL != perm);
+        }
+        virtual void Destroy() { delete[] perm; perm = NULL; }
+
+        // Block-oriented encode is unused by this rateless surrogate; repair
+        // symbols are synthesized on demand in EncodeParity() below.
+        virtual void Encode(unsigned int, const char*, char**) {}
+
+        // Generate the "parityId"-th repair symbol as a copy of a source symbol
+        // selected by the shared permutation.
+        virtual void EncodeParity(unsigned int parityId, const char** sourceVectorList,
+                                  unsigned int numData, char* parityVector)
+        {
+            if (0 == numData) return;
+            BuildRatelessPermutation(perm, numData);
+            unsigned int k = perm[parityId % numData];
+            if ((k < numData) && (NULL != sourceVectorList[k]))
+                memcpy(parityVector, sourceVectorList[k], vec_size);
+        }
+
+        virtual bool IsRateless() const { return true; }
+
+    private:
+        unsigned int  ndata;
+        UINT16        vec_size;
+        unsigned int* perm;  // scratch buffer sized to the block size
 };
 
-class MockRatelessDecoder : public NormDecoder {
+class MockRatelessDecoder : public NormDecoder
-public:
+{
-    virtual bool Init(unsigned int numData, unsigned int numParity, UINT16 vectorSize) {
+    public:
-        ndata = numData;
+        MockRatelessDecoder() : ndata(0), nparity(0), vec_size(0), perm(NULL) {}
-        vec_size = vectorSize;
+        virtual ~MockRatelessDecoder() { Destroy(); }
-        return true;
-    }
-    virtual void Destroy() {}
-    virtual int Decode(char** vectorList, unsigned int numData, unsigned int erasureCount, unsigned int* erasureLocs) {
-        return erasureCount;
-    }
 
-private:
+        virtual bool Init(unsigned int numData, unsigned int numParity, UINT16 vectorSize)
-    unsigned int ndata;
+        {
-    UINT16 vec_size;
+            Destroy();
+            ndata = numData;
+            nparity = numParity;
+            vec_size = vectorSize;
+            perm = new unsigned int[ndata ? ndata : 1];
+            return (NULL != perm);
+        }
+        virtual void Destroy() { delete[] perm; perm = NULL; }
+
+        // Fill erased source symbols using received repair symbols. Returns the
+        // number of source erasures that could NOT be recovered (0 == success),
+        // which NORM uses to decide whether to NACK for more repair symbols.
+        virtual int Decode(char** vectorList, unsigned int numData,
+                           unsigned int erasureCount, unsigned int* erasureLocs)
+        {
+            unsigned int total = numData + nparity;
+            bool* erased = new bool[total];
+            for (unsigned int i = 0; i < total; i++) erased[i] = false;
+            for (unsigned int i = 0; i < erasureCount; i++)
+                if (erasureLocs[i] < total) erased[erasureLocs[i]] = true;
+
+            unsigned int sourceErasures = 0;
+            for (unsigned int i = 0; i < numData; i++)
+                if (erased[i]) sourceErasures++;
+
+            BuildRatelessPermutation(perm, numData);
+
+            unsigned int recovered = 0;
+            for (unsigned int p = 0; (p < nparity) && (recovered < sourceErasures); p++)
+            {
+                unsigned int pos = numData + p;
+                if (erased[pos] || (NULL == vectorList[pos])) continue;  // repair symbol not received
+                unsigned int k = perm[p % numData];  // source symbol this repair carries
+                if ((k < numData) && erased[k] && (NULL != vectorList[k]))
+                {
+                    memcpy(vectorList[k], vectorList[pos], vec_size);
+                    erased[k] = false;
+                    recovered++;
+                }
+            }
+            unsigned int stillMissing = sourceErasures - recovered;
+            delete[] erased;
+            return (int)stillMissing;
+        }
+
+    private:
+        unsigned int  ndata;
+        unsigned int  nparity;
+        UINT16        vec_size;
+        unsigned int* perm;
 };
 
 extern "C" {
@@ -42,47 +130,126 @@ extern "C" {
     NormDecoder* CreateMockDecoder() { return new MockRatelessDecoder(); }
 }
 
-int main(int argc, char* argv[]) {
+int main(int /*argc*/, char* /*argv*/[])
+{
     NormInstanceHandle instance = NormCreateInstance();
-    if (!instance) return -1;
+    if (NORM_INSTANCE_INVALID == instance) return -1;
-    
+
-    NormSessionHandle session = NormCreateSession(instance, "127.0.0.1", 6003, 1);
+    if (getenv("NORM_DEBUG") != NULL) NormSetDebugLevel((unsigned int)atoi(getenv("NORM_DEBUG")));
-    
+    srand((unsigned int)time(NULL));
-    // Register Rateless Coder (FEC ID 131)
+
-    if (!NormRegisterFecCoder(session, 131, CreateMockEncoder, CreateMockDecoder, true)) {
+    // Use two sessions with DISTINCT node ids on the same multicast group so the
-        fprintf(stderr, "Failed to register custom FEC codec.\n");
+    // receiver treats the sender as a genuine remote peer.  (A single session
+    // acting as both sender and receiver would hear its own NACKs and suppress
+    // subsequent ones, stalling multi-round rateless repair.)
+    const char* groupAddr = "224.1.2.3";
+    const UINT16 groupPort = 6003;
+
+    NormSessionHandle txSession = NormCreateSession(instance, groupAddr, groupPort, 1);
+    NormSessionHandle rxSession = NormCreateSession(instance, groupAddr, groupPort, 2);
+    if ((NORM_SESSION_INVALID == txSession) || (NORM_SESSION_INVALID == rxSession)) return -1;
+
+    // Register the mock rateless codec on both the sender and receiver sessions.
+    if (!NormRegisterFecCoder(txSession, RATELESS_FEC_ID, CreateMockEncoder, CreateMockDecoder, true) ||
+        !NormRegisterFecCoder(rxSession, RATELESS_FEC_ID, CreateMockEncoder, CreateMockDecoder, true))
+    {
+        fprintf(stderr, "normRatelessTest error: failed to register rateless FEC codec\n");
         return -1;
     }
-    
+
-    NormSetRxPortReuse(session, true);
+    NormSetRxPortReuse(txSession, true);
-    NormSetMulticastLoopback(session, true);
+    NormSetRxPortReuse(rxSession, true);
-    
+    NormSetMulticastLoopback(txSession, true);  // let our data reach the rx session on this host
-    // Start Sender using FEC 131
+    NormSetMulticastLoopback(rxSession, true);  // let our NACKs reach the tx session on this host
-    if (!NormStartSender(session, 1, 1024*1024, 1024, 64, 0, 131)) {
+    NormSetTxRate(txSession, 10.0e+06);  // 10 Mbps for a quick loopback run
-        fprintf(stderr, "Failed to start sender.\n");
+
+    // Induce loss on the receiver so it must repair via rateless parity.
+    double lossPct = (getenv("NORM_RX_LOSS") != NULL) ? atof(getenv("NORM_RX_LOSS")) : 10.0;
+    NormSetRxLoss(rxSession, lossPct);
+
+    if (!NormStartReceiver(rxSession, 1024 * 1024))
+    {
+        fprintf(stderr, "normRatelessTest error: NormStartReceiver() failed\n");
         return -1;
     }
-    
+
-    printf("Sender started with mock rateless codec. Transmitting...\n");
+    // 16 source symbols/block, up to 64 repair symbols/block, 512-byte segments.
-    
+    if (!NormStartSender(txSession, 1, 1024 * 1024, 512, 16, 64, RATELESS_FEC_ID))
-    NormObjectHandle obj = NormDataEnqueue(session, "testdata", 8);
+    {
-    
+        fprintf(stderr, "normRatelessTest error: NormStartSender() failed\n");
-    if (obj) {
+        return -1;
-        bool running = true;
+    }
-        while (running) {
+
-            NormEvent event;
+    // Build a known payload spanning several blocks (incl. a short final block).
-            if (NormGetNextEvent(instance, &event)) {
+    const UINT32 dataLen = 20000;
-                if (event.type == NORM_TX_FLUSH_COMPLETED) {
+    char* txData = new char[dataLen];
-                    printf("Transmission flushed completely.\n");
+    for (UINT32 i = 0; i < dataLen; i++)
+        txData[i] = (char)((i * 31 + 7) & 0xff);
+
+    NormObjectHandle txObj = NormDataEnqueue(txSession, txData, dataLen);
+    if (NORM_OBJECT_INVALID == txObj)
+    {
+        fprintf(stderr, "normRatelessTest error: NormDataEnqueue() failed\n");
+        return -1;
+    }
+    printf("normRatelessTest: sending %u bytes via mock rateless codec (10%% loss)...\n", dataLen);
+
+    int result = -1;
+    // NormGetNextEvent() blocks, so drive it via select() on the NORM descriptor
+    // to enforce a wall-clock watchdog (the demo must never hang).
+    NormDescriptor normFd = NormGetDescriptor(instance);
+    time_t deadline = time(NULL) + 30;
+    bool running = true;
+    while (running && (time(NULL) <= deadline))
+    {
+        fd_set fdSet;
+        FD_ZERO(&fdSet);
+        FD_SET(normFd, &fdSet);
+        struct timeval timeout;
+        timeout.tv_sec = 1;
+        timeout.tv_usec = 0;
+        int n = select((int)normFd + 1, &fdSet, NULL, NULL, &timeout);
+        if (n <= 0) continue;  // timeout tick (re-check deadline) or interrupted
+
+        NormEvent event;
+        while (running && NormGetNextEvent(instance, &event, false))
+        {
+            switch (event.type)
+            {
+                case NORM_RX_OBJECT_COMPLETED:
+                {
+                    unsigned int rxLen = (unsigned int)NormObjectGetSize(event.object);
+                    const char* rxData = NormDataAccessData(event.object);
+                    if ((rxLen == dataLen) && (NULL != rxData) && (0 == memcmp(rxData, txData, dataLen)))
+                    {
+                        printf("normRatelessTest: received %u bytes, contents verified. PASS\n", rxLen);
+                        result = 0;
+                    }
+                    else
+                    {
+                        fprintf(stderr, "normRatelessTest: received object mismatch (len %u vs %u). FAIL\n",
+                                rxLen, dataLen);
+                    }
                     running = false;
+                    break;
                 }
+                case NORM_RX_OBJECT_ABORTED:
+                    fprintf(stderr, "normRatelessTest error: NORM_RX_OBJECT_ABORTED\n");
+                    running = false;
+                    break;
+                default:
+                    break;
             }
         }
     }
-    
+    if (running)
-    NormStopSender(session);
+        fprintf(stderr, "normRatelessTest error: timed out waiting for object reception\n");
+
+    NormStopSender(txSession);
+    NormStopReceiver(rxSession);
     NormDestroyInstance(instance);
-    
+    delete[] txData;
-    printf("Success.\n");
+
-    return 0;
+    printf("normRatelessTest: %s\n", (0 == result) ? "SUCCESS" : "FAILURE");
+    return result;
 }

include/normEncoder.h

@@ -52,8 +52,13 @@ class NormEncoder
         virtual ~NormEncoder();
         virtual bool Init(unsigned int numData, unsigned int numParity, UINT16 vectorSize) = 0;
         virtual void Destroy() = 0;
-        virtual void Encode(unsigned int segmentId, const char *dataVector, char **parityVectorList) = 0;    
+        virtual void Encode(unsigned int segmentId, const char *dataVector, char **parityVectorList) = 0;
-        virtual void EncodeParity(unsigned int parityId, char* parityVector) { (void)parityId; (void)parityVector; }
+        // Rateless codes generate parity symbols on demand.  Unlike the block-oriented
+        // Encode() above, EncodeParity() is handed the block's source symbol vectors so a
+        // fountain/rateless encoder can synthesize the "parityId"-th (innovative) repair
+        // symbol without having to cache source state across calls.
+        virtual void EncodeParity(unsigned int parityId, const char** sourceVectorList, unsigned int numData, char* parityVector)
+            { (void)parityId; (void)sourceVectorList; (void)numData; (void)parityVector; }
         virtual bool IsRateless() const { return false; }
         virtual UINT16 CalculateProactiveParity(unsigned int blockId, UINT16 numData, const NormTelemetryContext& context) { return 0; }
         virtual UINT16 CalculateReactiveParity(unsigned int blockId, UINT16 requestedErasures, const NormTelemetryContext& context) { return requestedErasures; }
@@ -65,7 +70,13 @@ class NormDecoder
         virtual ~NormDecoder();
         virtual bool Init(unsigned int numData, unsigned int numParity, UINT16 vectorSize) = 0;
         virtual void Destroy() = 0;
-        virtual int Decode(char** vectorList, unsigned int numData,  unsigned int erasureCount, unsigned int* erasureLocs) = 0;    
+        // Decode in place, filling any erased source symbol vectors in "vectorList".
+        // For ideal (MDS) codes (e.g. Reed-Solomon) the return value is unused by the
+        // receiver, which assumes success whenever it holds >= erasureCount repair symbols.
+        // For rateless/non-ideal codes the return value MUST be the number of source
+        // symbol erasures that could NOT be recovered (0 == fully decoded); the receiver
+        // uses this to keep the block pending and NACK for additional repair symbols.
+        virtual int Decode(char** vectorList, unsigned int numData,  unsigned int erasureCount, unsigned int* erasureLocs) = 0;
 };  // end class NormDecoder
 
 #endif // _NORM_ENCODER

include/normSegment.h

@@ -163,6 +163,7 @@ class NormBlock
             erasure_count = ndata;
             parity_count = 0;
             parity_offset = 0;
+            decode_overhead = 0;
             flags = 0;
         }
         // Note: This invalidates the repair_mask state.
@@ -172,6 +173,10 @@ class NormBlock
         UINT16 ErasureCount() const {return erasure_count;}
         void IncrementParityCount() {parity_count++;}
         UINT16 ParityCount() const {return parity_count;}
+        // Extra repair symbols a (rateless) receiver needs beyond ErasureCount() because
+        // prior decode attempts were short of innovative symbols (0 for ideal MDS codes).
+        void IncrementDecodeOverhead() {decode_overhead++;}
+        UINT16 DecodeOverhead() const {return decode_overhead;}
         
         bool GetFirstPending(NormSymbolId& symbolId) const
         {
@@ -279,6 +284,7 @@ class NormBlock
         int          flags;
         UINT16       erasure_count;
         UINT16       parity_count;  // how many fresh parity we are currently planning to send
+        UINT16       decode_overhead;  // extra repair symbols a rateless rcvr needs after a short decode
         UINT16       parity_offset; // offset from where our fresh parity will be sent
         UINT16       seg_size_max;
         

include/normSession.h

@@ -560,8 +560,8 @@ class NormSession : public NormTelemetryContext
         
         void SenderEncode(unsigned int segmentId, const char* segment, char** parityVectorList)
             {encoder->Encode(segmentId, segment, parityVectorList);}
-        void SenderEncodeParity(unsigned int parityId, char* parityVector)
+        void SenderEncodeParity(unsigned int parityId, const char** sourceVectorList, unsigned int numData, char* parityVector)
-            {encoder->EncodeParity(parityId, parityVector);}
+            {encoder->EncodeParity(parityId, sourceVectorList, numData, parityVector);}
         
         
         NormBlock* SenderGetFreeBlock(NormObjectId objectId, NormBlockId blockId);

src/common/normObject.cpp

@@ -1609,42 +1609,66 @@ void NormObject::HandleObjectMessage(const NormObjectMsg& msg,
                         }  // end if (nextErasure < numData)
                     }  // end if (block->GetFirstPending(nextErasure))                 
                     
+                    bool ratelessDecodeIncomplete = false;
                     if (erasureCount)
                     {
-                        sender->Decode(block->SegmentList(), numData, erasureCount); 
+                        // The decoder returns the number of source erasures it could NOT
-                        for (UINT16 i = 0; i < erasureCount; i++) 
+                        // recover.  Ideal (MDS) codes always fully decode here, so their
+                        // return is effectively ignored; rateless/non-ideal codes may need
+                        // more repair symbols than the "ParityCount >= ErasureCount" guard
+                        // above assumes, so we must honor a partial-decode result.
+                        UINT16 unrecovered = sender->Decode(block->SegmentList(), numData, erasureCount);
+                        ratelessDecodeIncomplete = session.IsRatelessFec(fec_id) && (0 != unrecovered);
+                        if (!ratelessDecodeIncomplete)
                         {
-                            NormSegmentId sid = sender->GetErasureLoc(i);
+                            for (UINT16 i = 0; i < erasureCount; i++)
-                            if (sid < numData)
                             {
-                                if (WriteSegment(blockId, sid, block->GetSegment(sid)))
+                                NormSegmentId sid = sender->GetErasureLoc(i);
+                                if (sid < numData)
                                 {
-                                    objectUpdated = true;
+                                    if (WriteSegment(blockId, sid, block->GetSegment(sid)))
-                                    // For statistics only (TBD) #ifdef NORM_DEBUG
+                                    {
-                                    // "segmentLength" is not necessarily correct here (TBD - fix this)
+                                        objectUpdated = true;
-                                    sender->IncrementRecvGoodput(segmentLength);
+                                        // For statistics only (TBD) #ifdef NORM_DEBUG
-                                }  
+                                        // "segmentLength" is not necessarily correct here (TBD - fix this)
+                                        sender->IncrementRecvGoodput(segmentLength);
+                                    }
+                                    else
+                                    {
+                                        if (IsStream())
+                                            PLOG(PL_DEBUG, "NormObject::HandleObjectMessage() WriteSegment() error\n");
+                                        else
+                                            PLOG(PL_ERROR, "NormObject::HandleObjectMessage() WriteSegment() error\n");
+                                    }
+                                }
                                 else
                                 {
-                                    if (IsStream())
+                                    break;
-                                        PLOG(PL_DEBUG, "NormObject::HandleObjectMessage() WriteSegment() error\n");
+                                }
-                                    else
-                                        PLOG(PL_ERROR, "NormObject::HandleObjectMessage() WriteSegment() error\n");
-                                } 
-                            }
-                            else
-                            {
-                                break;
                             }
                         }
+                        else
+                        {
+                            // Not enough innovative repair symbols yet.  Note the shortfall so
+                            // the receiver's next NACK requests additional parity, and leave the
+                            // block pending (its missing source symbols are still marked pending).
+                            block->IncrementDecodeOverhead();
+                            PLOG(PL_DEBUG, "NormObject::HandleObjectMessage() node>%lu sender>%lu obj>%hu blk>%lu "
+                                           "rateless decode incomplete (%hu source symbol(s) short); awaiting more parity\n",
+                                 (unsigned long)LocalNodeId(), (unsigned long)sender->GetId(),
+                                 (UINT16)transport_id, (unsigned long)blockId.GetValue(), unrecovered);
+                        }
                     }
                     // Clear any temporarily retrieved segments for the block
-                    for (UINT16 i = 0; i < retrievalCount; i++) 
+                    for (UINT16 i = 0; i < retrievalCount; i++)
                         block->DetachSegment(sender->GetRetrievalLoc(i));
-                    // OK, we're done with this block
+                    if (!ratelessDecodeIncomplete)
-                    pending_mask.Unset(blockId.GetValue());
+                    {
-                    block_buffer.Remove(block);
+                        // OK, we're done with this block
-                    sender->PutFreeBlock(block); 
+                        pending_mask.Unset(blockId.GetValue());
+                        block_buffer.Remove(block);
+                        sender->PutFreeBlock(block);
+                    }
                 }  // if erasureCount <= parityCount (i.e., block complete)
                 // Notify application of new data available
                 // (TBD) this could be improved for stream objects
@@ -2068,8 +2092,27 @@ bool NormObject::NextSenderMsg(NormObjectMsg* msg)
                     memset(buffer+payloadLength, 0, payloadMax-payloadLength);
                 // (TBD) the encode routine could update the block's parity readiness
                 block->UpdateSegSizeMax(payloadLength);
-                session.SenderEncode(segmentId, data->AccessPayload(), block->SegmentList(numData)); 
+                session.SenderEncode(segmentId, data->AccessPayload(), block->SegmentList(numData));
-                block->IncreaseParityReadiness();     
+                block->IncreaseParityReadiness();
+                // Rateless codecs synthesize repair symbols on demand from the block's
+                // source symbols, so retain each source vector in the block as it is sent.
+                // (This incremental path drives ParityReadiness to completion, after which
+                // CalculateBlockParity() is skipped, so caching must happen here too.)
+                if (session.GetEncoder() && session.GetEncoder()->IsRateless() &&
+                    (NULL == block->GetSegment(segmentId)))
+                {
+                    char* srcSeg = session.SenderGetFreeSegment(transport_id, block->GetId());
+                    if (NULL != srcSeg)
+                    {
+                        memcpy(srcSeg, buffer, payloadMax);
+                        block->AttachSegment(segmentId, srcSeg);
+                    }
+                    else
+                    {
+                        PLOG(PL_INFO, "NormObject::NextSenderMsg() node>%lu warning: no free segment to cache "
+                                      "rateless source symbol.\n", (unsigned long)LocalNodeId());
+                    }
+                }
             }
         }
         else
@@ -2092,7 +2135,7 @@ bool NormObject::NextSenderMsg(NormObjectMsg* msg)
                         payloadMax = MIN(payloadMax, SIM_PAYLOAD_MAX);
 #endif // SIMULATE
                         memset(segment, 0, payloadMax);
-                        session.SenderEncodeParity(segmentId - numData, segment);
+                        session.SenderEncodeParity(segmentId - numData, (const char**)block->SegmentList(), numData, segment);
                         block->AttachSegment(segmentId, segment);
                     }
                     else
@@ -2260,10 +2303,29 @@ bool NormObject::CalculateBlockParity(NormBlock* block)
                 memset(buffer+payloadLength, 0, payloadMax-payloadLength+1);
             block->UpdateSegSizeMax(payloadLength);
             session.SenderEncode(i, buffer, block->SegmentList(numData));
+            // Rateless codecs synthesize repair symbols on demand (EncodeParity) from
+            // the block's source symbols, so cache the source vectors in the block.
+            // (Block codes like Reed-Solomon accumulate parity above and don't need this.)
+            if (session.GetEncoder() && session.GetEncoder()->IsRateless() &&
+                (NULL == block->GetSegment(i)))
+            {
+                char* srcSeg = session.SenderGetFreeSegment(transport_id, block->GetId());
+                if (NULL != srcSeg)
+                {
+                    memcpy(srcSeg, buffer, payloadMax);
+                    block->AttachSegment(i, srcSeg);
+                }
+                else
+                {
+                    PLOG(PL_INFO, "NormObject::CalculateBlockParity() node>%lu warning: no free segment to cache "
+                                  "rateless source symbol.\n", (unsigned long)LocalNodeId());
+                    return false;
+                }
+            }
         }
         else
         {
-            return false;   
+            return false;
         }
     }
     block->SetParityReadiness(numData);

src/common/normSegment.cpp

@@ -208,10 +208,16 @@ bool NormBlock::IsRepairPending(UINT16 numData, UINT16 numParity)
     }
     else
     {
+        // We need "erasure_count" repair symbols, plus "decode_overhead" more for
+        // rateless codes whose earlier repair symbols weren't innovative enough to
+        // decode (decode_overhead is always 0 for ideal MDS codes, preserving the
+        // original behavior).
+        UINT16 parityNeed = erasure_count + decode_overhead;
+        if (parityNeed > numParity) parityNeed = numParity;
         repair_mask.SetBits(0, numData);
-        repair_mask.SetBits(numData+erasure_count, numParity-erasure_count);
+        repair_mask.SetBits(numData+parityNeed, numParity-parityNeed);
     }
-    // Calculate repair_mask = pending_mask - repair_mask 
+    // Calculate repair_mask = pending_mask - repair_mask
     repair_mask.XCopy(pending_mask);
     return (repair_mask.IsSet());
 }  // end NormBlock::IsRepairPending()
@@ -535,7 +541,9 @@ bool NormBlock::AppendRepairRequest(NormNackMsg&    nack,
 {
     if (isRateless)
     {
-        int needed = (int)erasure_count - (int)parity_count + (int)fecOverhead;
+        // "decode_overhead" grows when a prior decode attempt fell short of innovative
+        // symbols, so the receiver keeps asking for more parity until the block decodes.
+        int needed = (int)erasure_count - (int)parity_count + (int)fecOverhead + (int)decode_overhead;
         if (needed <= 0) return false;
         
         UINT16 needed_clamped = (needed > 65535) ? 65535 : (UINT16)needed;