From 2d653628ce7849120c53d558c5db106f295d1988 Mon Sep 17 00:00:00 2001 From: Joseph Calderon Date: Sun, 12 Jul 2026 20:19:34 -0700 Subject: [PATCH] support partial rateless decode with re-NACK + enhance mock --- examples/normRatelessTest.cpp | 283 +++++++++++++++++++++++++++------- include/normEncoder.h | 17 +- include/normSegment.h | 6 + include/normSession.h | 4 +- src/common/normObject.cpp | 118 ++++++++++---- src/common/normSegment.cpp | 14 +- 6 files changed, 348 insertions(+), 94 deletions(-) diff --git a/examples/normRatelessTest.cpp b/examples/normRatelessTest.cpp index ba4b8a8..8921c22 100644 --- a/examples/normRatelessTest.cpp +++ b/examples/normRatelessTest.cpp @@ -1,40 +1,128 @@ #include #include + #include #include #include +#include +#include +#include -class MockRatelessEncoder : public NormEncoder { -public: - virtual bool Init(unsigned int numData, unsigned int numParity, UINT16 vectorSize) { - ndata = numData; - vec_size = vectorSize; - return true; +// FEC ID for the (partially-specified) rateless code family. +static const UINT8 RATELESS_FEC_ID = 131; + +// Deterministic permutation of [0, n) shared by encoder and decoder. +// The seed depends only on "n", so both sides produce the identical ordering. +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: - unsigned int ndata; - UINT16 vec_size; +class MockRatelessEncoder : public NormEncoder +{ + 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 { -public: - virtual bool Init(unsigned int numData, unsigned int numParity, UINT16 vectorSize) { - ndata = numData; - vec_size = vectorSize; - return true; - } - virtual void Destroy() {} - virtual int Decode(char** vectorList, unsigned int numData, unsigned int erasureCount, unsigned int* erasureLocs) { - return erasureCount; - } +class MockRatelessDecoder : public NormDecoder +{ + public: + MockRatelessDecoder() : ndata(0), nparity(0), vec_size(0), perm(NULL) {} + virtual ~MockRatelessDecoder() { Destroy(); } -private: - unsigned int ndata; - UINT16 vec_size; + virtual bool Init(unsigned int numData, unsigned int numParity, UINT16 vectorSize) + { + 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; - - NormSessionHandle session = NormCreateSession(instance, "127.0.0.1", 6003, 1); - - // Register Rateless Coder (FEC ID 131) - if (!NormRegisterFecCoder(session, 131, CreateMockEncoder, CreateMockDecoder, true)) { - fprintf(stderr, "Failed to register custom FEC codec.\n"); + if (NORM_INSTANCE_INVALID == instance) return -1; + + if (getenv("NORM_DEBUG") != NULL) NormSetDebugLevel((unsigned int)atoi(getenv("NORM_DEBUG"))); + srand((unsigned int)time(NULL)); + + // Use two sessions with DISTINCT node ids on the same multicast group so the + // 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); - NormSetMulticastLoopback(session, true); - - // Start Sender using FEC 131 - if (!NormStartSender(session, 1, 1024*1024, 1024, 64, 0, 131)) { - fprintf(stderr, "Failed to start sender.\n"); + + NormSetRxPortReuse(txSession, true); + NormSetRxPortReuse(rxSession, true); + NormSetMulticastLoopback(txSession, true); // let our data reach the rx session on this host + NormSetMulticastLoopback(rxSession, true); // let our NACKs reach the tx session on this host + NormSetTxRate(txSession, 10.0e+06); // 10 Mbps for a quick loopback run + + // 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"); - - NormObjectHandle obj = NormDataEnqueue(session, "testdata", 8); - - if (obj) { - bool running = true; - while (running) { - NormEvent event; - if (NormGetNextEvent(instance, &event)) { - if (event.type == NORM_TX_FLUSH_COMPLETED) { - printf("Transmission flushed completely.\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)) + { + fprintf(stderr, "normRatelessTest error: NormStartSender() failed\n"); + return -1; + } + + // Build a known payload spanning several blocks (incl. a short final block). + const UINT32 dataLen = 20000; + char* txData = new char[dataLen]; + 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; } } } - - NormStopSender(session); + if (running) + fprintf(stderr, "normRatelessTest error: timed out waiting for object reception\n"); + + NormStopSender(txSession); + NormStopReceiver(rxSession); NormDestroyInstance(instance); - - printf("Success.\n"); - return 0; + delete[] txData; + + printf("normRatelessTest: %s\n", (0 == result) ? "SUCCESS" : "FAILURE"); + return result; } diff --git a/include/normEncoder.h b/include/normEncoder.h index 4216198..bd0ed73 100755 --- a/include/normEncoder.h +++ b/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 EncodeParity(unsigned int parityId, char* parityVector) { (void)parityId; (void)parityVector; } + virtual void Encode(unsigned int segmentId, const char *dataVector, char **parityVectorList) = 0; + // 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 diff --git a/include/normSegment.h b/include/normSegment.h index cb68527..2a437c5 100755 --- a/include/normSegment.h +++ b/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; diff --git a/include/normSession.h b/include/normSession.h index a27350c..bcc0c87 100755 --- a/include/normSession.h +++ b/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) - {encoder->EncodeParity(parityId, parityVector);} + void SenderEncodeParity(unsigned int parityId, const char** sourceVectorList, unsigned int numData, char* parityVector) + {encoder->EncodeParity(parityId, sourceVectorList, numData, parityVector);} NormBlock* SenderGetFreeBlock(NormObjectId objectId, NormBlockId blockId); diff --git a/src/common/normObject.cpp b/src/common/normObject.cpp index eca1717..5fdd6e5 100755 --- a/src/common/normObject.cpp +++ b/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); - for (UINT16 i = 0; i < erasureCount; i++) + // The decoder returns the number of source erasures it could NOT + // 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); - if (sid < numData) + for (UINT16 i = 0; i < erasureCount; i++) { - if (WriteSegment(blockId, sid, block->GetSegment(sid))) + NormSegmentId sid = sender->GetErasureLoc(i); + if (sid < numData) { - objectUpdated = true; - // For statistics only (TBD) #ifdef NORM_DEBUG - // "segmentLength" is not necessarily correct here (TBD - fix this) - sender->IncrementRecvGoodput(segmentLength); - } + if (WriteSegment(blockId, sid, block->GetSegment(sid))) + { + objectUpdated = true; + // 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()) - PLOG(PL_DEBUG, "NormObject::HandleObjectMessage() WriteSegment() error\n"); - else - PLOG(PL_ERROR, "NormObject::HandleObjectMessage() WriteSegment() error\n"); - } - } - else - { - break; + 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 - pending_mask.Unset(blockId.GetValue()); - block_buffer.Remove(block); - sender->PutFreeBlock(block); + if (!ratelessDecodeIncomplete) + { + // OK, we're done with this 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)); - block->IncreaseParityReadiness(); + session.SenderEncode(segmentId, data->AccessPayload(), block->SegmentList(numData)); + 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); diff --git a/src/common/normSegment.cpp b/src/common/normSegment.cpp index d31a2da..10f381e 100755 --- a/src/common/normSegment.cpp +++ b/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;