From fe495011fa86188087df147ba26fcd418dad5731 Mon Sep 17 00:00:00 2001 From: joseph calderon Date: Sat, 13 Jun 2026 22:45:23 -0700 Subject: [PATCH 1/9] basic rateless codec support --- examples/normRatelessTest.cpp | 88 ++++++++++++++++++++++ include/normApi.h | 25 +++++++ include/normEncoder.h | 15 ++++ include/normMessage.h | 119 +++++++++++++++++++++++++++--- include/normSegment.h | 6 +- include/normSession.h | 37 +++++++++- makefiles/Makefile.common | 10 ++- src/common/normApi.cpp | 30 ++++++++ src/common/normNode.cpp | 133 +++++++++++++++++++++------------- src/common/normObject.cpp | 57 +++++++++++++-- src/common/normSegment.cpp | 30 +++++++- src/common/normSession.cpp | 61 ++++++++++++++-- 12 files changed, 532 insertions(+), 79 deletions(-) create mode 100644 examples/normRatelessTest.cpp diff --git a/examples/normRatelessTest.cpp b/examples/normRatelessTest.cpp new file mode 100644 index 00000000..ba4b8a80 --- /dev/null +++ b/examples/normRatelessTest.cpp @@ -0,0 +1,88 @@ +#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; + } + 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 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; + } + +private: + unsigned int ndata; + UINT16 vec_size; +}; + +extern "C" { + NormEncoder* CreateMockEncoder() { return new MockRatelessEncoder(); } + NormDecoder* CreateMockDecoder() { return new MockRatelessDecoder(); } +} + +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"); + 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"); + 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"); + running = false; + } + } + } + } + + NormStopSender(session); + NormDestroyInstance(instance); + + printf("Success.\n"); + return 0; +} diff --git a/include/normApi.h b/include/normApi.h index 0caa5c34..e29b8b2b 100755 --- a/include/normApi.h +++ b/include/normApi.h @@ -851,6 +851,31 @@ NORM_API_LINKAGE bool NormNodeDenySender(NormNodeId senderId); #ifdef __cplusplus +struct NormFecLayout +{ + UINT16 payloadIdLength; + UINT32 blockMask; + void (*packPayloadId)(UINT32* buffer, UINT32 blockId, UINT16 symbolId, UINT16 blockLen); + UINT32 (*unpackBlockId)(const UINT32* buffer); + UINT16 (*unpackSymbolId)(const UINT32* buffer); + UINT16 (*unpackBlockLength)(const UINT32* buffer); +}; + +class NormEncoder; +class NormDecoder; +typedef NormEncoder* (*NormEncoderFactory)(); +typedef class NormDecoder* (*NormDecoderFactory)(); + +NORM_API_LINKAGE +bool NormRegisterFecCoder(NormSessionHandle sessionHandle, + UINT8 fecId, + NormEncoderFactory encoderFactory, + NormDecoderFactory decoderFactory, + bool isRateless); + +NORM_API_LINKAGE +void NormRegisterFecLayout(NormInstanceHandle instance, UINT8 fecId, const NormFecLayout* layout); + } // end extern "C" #endif /* __cplusplus */ diff --git a/include/normEncoder.h b/include/normEncoder.h index 3c87875a..42161989 100755 --- a/include/normEncoder.h +++ b/include/normEncoder.h @@ -35,6 +35,17 @@ #include "protokit.h" // protolib stuff +class NormTelemetryContext +{ + public: + virtual ~NormTelemetryContext() {} + virtual double GetGrtt() const = 0; + virtual double GetTxRate() const = 0; + virtual unsigned int GetGroupSize() const = 0; + virtual double GetCurrentTime() const = 0; +}; + + class NormEncoder { public: @@ -42,6 +53,10 @@ class 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 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; } }; // end class NormEncoder class NormDecoder diff --git a/include/normMessage.h b/include/normMessage.h index 954c5e00..ddc8e95c 100755 --- a/include/normMessage.h +++ b/include/normMessage.h @@ -3,6 +3,7 @@ // PROTOLIB includes #include "protokit.h" +#include "normApi.h" // standard includes #include // for memcpy(), etc @@ -393,6 +394,8 @@ class NormHeaderExtension // FEC Payload Id content. The FEC Payload // Id format is dependent upon the "fec_id" (FEC Type) // and, in some cases, its field size ("m") parameter +struct NormFecLayout; + class NormPayloadId { public: @@ -400,31 +403,39 @@ class NormPayloadId { RS = 2, // fully-specified, general purpose Reed-Solomon RS8 = 5, // fully-specified 8-bit Reed-Solmon per RFC 5510 - SB = 129 // partially-specified "small block" codes + SB = 129, // partially-specified "small block" codes + RL = 131 // partially-specified "rateless" codes }; - static bool IsValid(UINT8 fecId) + static bool IsValid(UINT8 fecId, const NormFecLayout* layout = NULL) { + if (layout != NULL) + return true; switch (fecId) { case 2: case 5: case 129: + case RL: return true; default: return false; } } + NormPayloadId(UINT8 fecId, UINT8 m, UINT32* theBuffer) : fec_id(fecId), fec_m(m), buffer(theBuffer) {} NormPayloadId(UINT8 fecId, UINT8 m, const UINT32* theBuffer) : fec_id(fecId), fec_m(m), cbuffer(theBuffer) {} - static UINT16 GetLength(UINT8 fecId) + static UINT16 GetLength(UINT8 fecId, const NormFecLayout* layout = NULL) { + if (layout != NULL) + return layout->payloadIdLength; switch (fecId) { case 2: case 5: + case RL: return 4; case 129: return 8; @@ -433,8 +444,10 @@ class NormPayloadId } } - static UINT32 GetFecBlockMask(UINT8 fecId, UINT8 fecM) + static UINT32 GetFecBlockMask(UINT8 fecId, UINT8 fecM, const NormFecLayout* layout = NULL) { + if (layout != NULL) + return layout->blockMask; switch (fecId) { case 2: @@ -446,13 +459,20 @@ class NormPayloadId return 0x00ffffff; // 24-bit blockId case 129: return 0xffffffff; // 32-bit blockId + case RL: + return 0x0000ffff; // 16-bit blockId default: return 0x00000000; // invalid fecId } } - void SetFecPayloadId(UINT32 blockId, UINT16 symbolId, UINT16 blockLen) + void SetFecPayloadId(UINT32 blockId, UINT16 symbolId, UINT16 blockLen, const NormFecLayout* layout = NULL) { + if (layout != NULL && layout->packPayloadId != NULL) + { + layout->packPayloadId(buffer, blockId, symbolId, blockLen); + return; + } switch (fec_id) { case 2: @@ -473,17 +493,30 @@ class NormPayloadId *buffer = htonl(blockId); // 3 + 1 bytes break; case 129: + { *buffer = htonl(blockId); // 4 bytes UINT16* ptr = (UINT16*)(buffer + 1); ptr[0] = htons(blockLen); // 2 bytes ptr[1] = htons(symbolId); // 2 bytes break; + } + case RL: + { + UINT16* payloadId = (UINT16*)buffer; + payloadId[0] = htons(blockId); // 2 bytes + payloadId[1] = htons(symbolId); // 2 bytes + break; + } } } // Message processing methods - NormBlockId GetFecBlockId() const + NormBlockId GetFecBlockId(const NormFecLayout* layout = NULL) const { + if (layout != NULL && layout->unpackBlockId != NULL) + { + return layout->unpackBlockId(cbuffer); + } switch (fec_id) { case 2: @@ -504,14 +537,23 @@ class NormPayloadId } case 129: return ntohl(*cbuffer); + case RL: + { + UINT16* blockId = (UINT16*)cbuffer; + return ntohs(*blockId); + } default: ASSERT(0); return 0; } } - UINT16 GetFecSymbolId() const + UINT16 GetFecSymbolId(const NormFecLayout* layout = NULL) const { + if (layout != NULL && layout->unpackSymbolId != NULL) + { + return layout->unpackSymbolId(cbuffer); + } switch (fec_id) { case 2: @@ -535,14 +577,23 @@ class NormPayloadId UINT16* ptr = (UINT16*)(cbuffer + 1); return ntohs(ptr[1]); } + case RL: + { + UINT16* payloadId = (UINT16*)cbuffer; + return ntohs(payloadId[1]); + } default: ASSERT(0); return 0; } } - UINT16 GetFecBlockLength() const + UINT16 GetFecBlockLength(const NormFecLayout* layout = NULL) const { + if (layout != NULL && layout->unpackBlockLength != NULL) + { + return layout->unpackBlockLength(cbuffer); + } if (129 == fec_id) { UINT16* blockLen = (UINT16*)(cbuffer + 1); @@ -553,7 +604,6 @@ class NormPayloadId return 0; } } - private: @@ -1028,6 +1078,57 @@ class NormFtiExtension129 : public NormHeaderExtension }; }; // end class NormFtiExtension129 +class NormFtiExtension131 : public NormHeaderExtension +{ + public: + virtual void Init(UINT32* theBuffer, UINT16 numBytes) + { + AttachBuffer(theBuffer, numBytes); + SetType(FTI); + SetWords(4); + } + void SetFecInstanceId(UINT16 instanceId) + { ((UINT16*)buffer)[FEC_INSTANCE_OFFSET] = htons(instanceId);} + void SetFecMaxBlockLen(UINT16 ndata) + {((UINT16*)buffer)[FEC_NDATA_OFFSET] = htons(ndata);} + void SetFecNumParity(UINT16 nparity) + {((UINT16*)buffer)[FEC_NPARITY_OFFSET] = htons(nparity);} + void SetSegmentSize(UINT16 segmentSize) + {((UINT16*)buffer)[SEG_SIZE_OFFSET] = htons(segmentSize);} + void SetObjectSize(const NormObjectSize& objectSize) + { + ((UINT16*)buffer)[OBJ_SIZE_MSB_OFFSET] = htons(objectSize.MSB()); + buffer[OBJ_SIZE_LSB_OFFSET] = htonl(objectSize.LSB()); + } + + UINT16 GetFecInstanceId() const + { + return (ntohs(((UINT16*)buffer)[FEC_INSTANCE_OFFSET])); + } + UINT16 GetFecMaxBlockLen() const + {return (ntohs(((UINT16*)buffer)[FEC_NDATA_OFFSET]));} + UINT16 GetFecNumParity() const + {return (ntohs(((UINT16*)buffer)[FEC_NPARITY_OFFSET]));} + UINT16 GetSegmentSize() const + {return (ntohs(((UINT16*)buffer)[SEG_SIZE_OFFSET]));} + NormObjectSize GetObjectSize() const + { + return NormObjectSize(ntohs(((UINT16*)buffer)[OBJ_SIZE_MSB_OFFSET]), + ntohl(buffer[OBJ_SIZE_LSB_OFFSET])); + } + + private: + enum + { + OBJ_SIZE_MSB_OFFSET = (LENGTH_OFFSET + 1)/2, + OBJ_SIZE_LSB_OFFSET = ((OBJ_SIZE_MSB_OFFSET*2)+2)/4, + FEC_INSTANCE_OFFSET = ((OBJ_SIZE_LSB_OFFSET*4)+4)/2, + SEG_SIZE_OFFSET = ((FEC_INSTANCE_OFFSET*2)+2)/2, + FEC_NDATA_OFFSET = ((SEG_SIZE_OFFSET*2)+2)/2, + FEC_NPARITY_OFFSET = ((FEC_NDATA_OFFSET*2)+2)/2 + }; +}; // end class NormFtiExtension131 + class NormInfoMsg : public NormObjectMsg { diff --git a/include/normSegment.h b/include/normSegment.h index cf88173d..cb685273 100755 --- a/include/normSegment.h +++ b/include/normSegment.h @@ -238,14 +238,16 @@ class NormBlock NormBlockId finalBlockId, UINT16 finalSegmentSize) const; - bool AppendRepairRequest(NormNackMsg& nack, + bool AppendRepairRequest(class NormNackMsg& nack, UINT8 fecId, UINT8 fecM, UINT16 numData, UINT16 numParity, NormObjectId objectId, bool pendingInfo, - UINT16 payloadMax); + UINT16 payloadMax, + UINT16 fecOverhead = 2, + bool isRateless = false); void SetLastNackTime(const ProtoTime& theTime) diff --git a/include/normSession.h b/include/normSession.h index 38971c6e..5ee2d14b 100755 --- a/include/normSession.h +++ b/include/normSession.h @@ -4,9 +4,9 @@ #include "normMessage.h" #include "normObject.h" #include "normNode.h" -#include "normEncoder.h" -#include "protokit.h" +#include "normEncoder.h" +#include #include "protoCap.h" // for ProtoCap for ECN_SUPPORT @@ -27,6 +27,10 @@ class NormController { public: virtual ~NormController() {} + NormController() { memset(fec_layouts, 0, sizeof(fec_layouts)); } + + void SetFecLayout(UINT8 id, const NormFecLayout* layout) { fec_layouts[id] = layout; } + const NormFecLayout* GetFecLayout(UINT8 id) const { return fec_layouts[id]; } enum Event { EVENT_INVALID = 0, @@ -68,6 +72,7 @@ class NormController class NormNode* node, class NormObject* object) = 0; + const NormFecLayout* fec_layouts[256]; }; // end class NormController class NormSessionMgr @@ -123,10 +128,13 @@ class NormSessionMgr }; // end class NormSessionMgr +typedef class NormEncoder* (*NormEncoderFactory)(); +typedef class NormDecoder* (*NormDecoderFactory)(); -class NormSession +class NormSession : public NormTelemetryContext { friend class NormSessionMgr; + public: enum {DEFAULT_MESSAGE_POOL_DEPTH = 16}; @@ -465,6 +473,13 @@ class NormSession void SenderSetExtraParity(UINT16 extraParity) {extra_parity = extraParity;} + void RegisterFecCoder(UINT8 fecId, NormEncoderFactory encoderFactory, NormDecoderFactory decoderFactory, bool isRateless); + NormEncoderFactory GetEncoderFactory(UINT8 fecId) const; + NormDecoderFactory GetDecoderFactory(UINT8 fecId) const; + bool IsRatelessFec(UINT8 fecId) const { return is_rateless_codec[fecId]; } + + NormEncoder* GetEncoder() const { return encoder; } + INT32 Difference(NormBlockId a, NormBlockId b) const {return NormBlockId::Difference(a, b, fec_block_mask);} int Compare(NormBlockId a, NormBlockId b) const @@ -521,8 +536,21 @@ class NormSession void SenderSetFtiMode(FtiMode ftiMode) {fti_mode = ftiMode;} + void SetReceiverMaxDelay(unsigned int msec) + {rcvr_max_delay = msec;} + unsigned int GetReceiverMaxDelay() const + {return rcvr_max_delay;} + + // NormTelemetryContext Implementation + double GetGrtt() const override { return SenderGrtt(); } + double GetTxRate() const override { return tx_rate; } + virtual unsigned int GetGroupSize() const override { return (unsigned int)((NormNodeList*)&cc_node_list)->GetCount(); } + double GetCurrentTime() const override { ProtoTime t; t.GetCurrentTime(); return t.GetValue(); } + 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);} NormBlock* SenderGetFreeBlock(NormObjectId objectId, NormBlockId blockId); @@ -789,6 +817,9 @@ class NormSession NormBlockPool block_pool; NormSegmentPool segment_pool; NormEncoder* encoder; + NormEncoderFactory encoder_factories[256]; + NormDecoderFactory decoder_factories[256]; + bool is_rateless_codec[256]; UINT8 fec_id; UINT8 fec_m; INT32 fec_block_mask; diff --git a/makefiles/Makefile.common b/makefiles/Makefile.common index fb18d784..c747a78d 100755 --- a/makefiles/Makefile.common +++ b/makefiles/Makefile.common @@ -234,6 +234,15 @@ normClient: $(CLIENT_OBJ) libnorm.a $(LIBPROTO) $(CC) $(CFLAGS) -o $@ $(CLIENT_OBJ) $(LDFLAGS) libnorm.a $(LIBPROTO) $(LIBS) mkdir -p ../bin cp $@ ../bin/$@ + +# (normRatelessTest) rateless fec tester +RATELESS_SRC = $(EXAMPLE)/normRatelessTest.cpp +RATELESS_OBJ = $(RATELESS_SRC:.cpp=.o) + +normRatelessTest: $(RATELESS_OBJ) libnorm.a $(LIBPROTO) + $(CC) $(CFLAGS) -o $@ $(RATELESS_OBJ) $(LDFLAGS) libnorm.a $(LIBPROTO) $(LIBS) + mkdir -p ../bin + cp $@ ../bin/$@ @@ -308,4 +317,3 @@ distclean: clean # DO NOT DELETE THIS LINE -- mkdep uses it. # DO NOT PUT ANYTHING AFTER THIS LINE, IT WILL GO AWAY. - diff --git a/src/common/normApi.cpp b/src/common/normApi.cpp index a151e940..2fc81469 100755 --- a/src/common/normApi.cpp +++ b/src/common/normApi.cpp @@ -1676,6 +1676,36 @@ void NormSetAutoParity(NormSessionHandle sessionHandle, unsigned char autoParity } } // end NormSetAutoParity() +NORM_API_LINKAGE +bool NormRegisterFecCoder(NormSessionHandle sessionHandle, + UINT8 fecId, + NormEncoderFactory encoderFactory, + NormDecoderFactory decoderFactory, + bool isRateless) +{ + NormInstance* instance = NormInstance::GetInstanceFromSession(sessionHandle); + if (instance && instance->dispatcher.SuspendThread()) + { + NormSession* session = (NormSession*)sessionHandle; + if (session) + { + session->RegisterFecCoder(fecId, encoderFactory, decoderFactory, isRateless); + instance->dispatcher.ResumeThread(); + return true; + } + instance->dispatcher.ResumeThread(); + } + return false; +} // end NormRegisterFecCoder() + +NORM_API_LINKAGE +void NormRegisterFecLayout(NormInstanceHandle instanceHandle, UINT8 fecId, const NormFecLayout* layout) +{ + NormInstance* instance = (NormInstance*)instanceHandle; + if (instance) + instance->SetFecLayout(fecId, layout); +} // end NormRegisterFecLayout() + NORM_API_LINKAGE void NormSetGrttEstimate(NormSessionHandle sessionHandle, double grttEstimate) diff --git a/src/common/normNode.cpp b/src/common/normNode.cpp index 63619f0c..34e8c561 100755 --- a/src/common/normNode.cpp +++ b/src/common/normNode.cpp @@ -287,72 +287,85 @@ bool NormSenderNode::AllocateBuffers(unsigned int bufferSpace, if (0 != numParity) { - switch (fecId) + NormDecoderFactory decoderFactory = session.GetDecoderFactory(fecId); + if (NULL != decoderFactory) { - case 2: - if (8 == fecM) - { + if (NULL == (decoder = decoderFactory())) + { + PLOG(PL_FATAL, "NormSenderNode::AllocateBuffers() new custom decoder error\n"); + Close(); + return false; + } + } + else + { + switch (fecId) + { + case 2: + if (8 == fecM) + { + if (NULL == (decoder = new NormDecoderRS8)) + { + PLOG(PL_FATAL, "NormSenderNode::AllocateBuffers() new NormDecoderRS8 error: %s\n", GetErrorString()); + Close(); + return false; + } + } + else if (16 == fecM) + { + if (NULL == (decoder = new NormDecoderRS16)) + { + PLOG(PL_FATAL, "NormSenderNode::AllocateBuffers() new NormDecoderRS16 error: %s\n", GetErrorString()); + Close(); + return false; + } + } + else + { + PLOG(PL_FATAL, "NormSenderNode::AllocateBuffers() error: unsupported fecId=2 'm' value %d!\n", fecM); + Close(); + return false; + } + break; + case 5: if (NULL == (decoder = new NormDecoderRS8)) { PLOG(PL_FATAL, "NormSenderNode::AllocateBuffers() new NormDecoderRS8 error: %s\n", GetErrorString()); Close(); return false; } - } - else if (16 == fecM) - { - if (NULL == (decoder = new NormDecoderRS16)) + break; + case 129: + #ifdef ASSUME_MDP_FEC + if (NULL == (decoder = new NormDecoderMDP)) { - PLOG(PL_FATAL, "NormSenderNode::AllocateBuffers() new NormDecoderRS16 error: %s\n", GetErrorString()); + PLOG(PL_FATAL, "NormSenderNode::AllocateBuffers() new NormDecoderMDP error: %s\n", GetErrorString()); Close(); return false; } - } - else - { - PLOG(PL_FATAL, "NormSenderNode::AllocateBuffers() error: unsupported fecId=2 'm' value %d!\n", fecM); - Close(); - return false; - } - break; - case 5: - if (NULL == (decoder = new NormDecoderRS8)) - { - PLOG(PL_FATAL, "NormSenderNode::AllocateBuffers() new NormDecoderRS8 error: %s\n", GetErrorString()); - Close(); - return false; - } - break; - case 129: -#ifdef ASSUME_MDP_FEC - if (NULL == (decoder = new NormDecoderMDP)) - { - PLOG(PL_FATAL, "NormSenderNode::AllocateBuffers() new NormDecoderMDP error: %s\n", GetErrorString()); - Close(); - return false; - } -#else - if (0 == fecInstanceId) - { - if (NULL == (decoder = new NormDecoderRS8)) + #else + if (0 == fecInstanceId) { - PLOG(PL_FATAL, "NormSenderNode::AllocateBuffers() new NormDecoderRS8 error: %s\n", GetErrorString()); + if (NULL == (decoder = new NormDecoderRS8)) + { + PLOG(PL_FATAL, "NormSenderNode::AllocateBuffers() new NormDecoderRS8 error: %s\n", GetErrorString()); + Close(); + return false; + } + } + else + { + PLOG(PL_FATAL, "NormSenderNode::AllocateBuffers() error: unknown fecId=129 instanceId!\n"); Close(); - return false; + return false; } - } - else - { - PLOG(PL_FATAL, "NormSenderNode::AllocateBuffers() error: unknown fecId=129 instanceId!\n"); + #endif // if/else ASSUME_MDP_FEC + break; + default: + PLOG(PL_FATAL, "NormSenderNode::AllocateBuffers() error: unknown fecId>%d!\n", fecId); Close(); - return false; - } -#endif // if/else ASSUME_MDP_FEC - break; - default: - PLOG(PL_FATAL, "NormSenderNode::AllocateBuffers() error: unknown fecId>%d!\n", fecId); - Close(); - return false; + return false; + } } if (!decoder->Init(numData, numParity, segmentSize+NormDataMsg::GetStreamPayloadHeaderLength())) { @@ -1431,6 +1444,24 @@ bool NormSenderNode::GetFtiData(const NormObjectMsg& msg, NormFtiData& ftiData) } break; } + case 131: + { + NormFtiExtension131 fti; + while (msg.GetNextExtension(fti)) + { + if (NormHeaderExtension::FTI == fti.GetType()) + { + ftiData.SetFecInstanceId(fti.GetFecInstanceId()); + ftiData.SetFecFieldSize(16); + ftiData.SetSegmentSize(fti.GetSegmentSize()); + ftiData.SetFecMaxBlockLen(fti.GetFecMaxBlockLen()); + ftiData.SetFecNumParity(fti.GetFecNumParity()); + ftiData.SetObjectSize(fti.GetObjectSize()); + return true; + } + } + break; + } default: PLOG(PL_ERROR, "NormSenderNode::GetFtiData() node>%lu sender>%lu unknown fec_id type:%d\n", (unsigned long)LocalNodeId(), (unsigned long)GetId(), (int)fecId); diff --git a/src/common/normObject.cpp b/src/common/normObject.cpp index fdda3da5..eca1717f 100755 --- a/src/common/normObject.cpp +++ b/src/common/normObject.cpp @@ -1305,11 +1305,13 @@ bool NormObject::AppendRepairRequest(NormNackMsg& nack, requestAppended = true; } bool blockRequestAppended; + bool isRateless = session.IsRatelessFec(fec_id); if (flush || (nextId != max_pending_block)) { blockRequestAppended = block->AppendRepairRequest(nack, fec_id, fec_m, numData, nparity, - transport_id, pending_info, payloadMax); + transport_id, pending_info, payloadMax, + 2, isRateless); } else { @@ -1317,13 +1319,15 @@ bool NormObject::AppendRepairRequest(NormNackMsg& nack, { blockRequestAppended = block->AppendRepairRequest(nack, fec_id, fec_m, max_pending_segment, 0, - transport_id, pending_info, payloadMax); + transport_id, pending_info, payloadMax, + 2, isRateless); } else { blockRequestAppended = block->AppendRepairRequest(nack, fec_id, fec_m, numData, nparity, - transport_id, pending_info, payloadMax); + transport_id, pending_info, payloadMax, + 2, isRateless); } } if (blockRequestAppended) @@ -1850,6 +1854,17 @@ bool NormObject::NextSenderMsg(NormObjectMsg* msg) fti.SetFecNumParity(nparity); break; } + case 131: // RL + { + NormFtiExtension131 fti; + msg->AttachExtension(fti); + fti.SetObjectSize(object_size); + fti.SetFecInstanceId(0); + fti.SetSegmentSize(segment_size); + fti.SetFecMaxBlockLen(ndata); + fti.SetFecNumParity(nparity); + break; + } default: ASSERT(0); return false; @@ -1907,7 +1922,8 @@ bool NormObject::NextSenderMsg(NormObjectMsg* msg) } // Load block with zero initialized parity segments UINT16 totalBlockLen = numData + nparity; - for (UINT16 i = numData; i < totalBlockLen; i++) + bool doPreallocate = (!session.GetEncoder() || !session.GetEncoder()->IsRateless()); + for (UINT16 i = numData; doPreallocate && (i < totalBlockLen); i++) { char* s = session.SenderGetFreeSegment(transport_id, blockId); if (s) @@ -1927,7 +1943,11 @@ bool NormObject::NextSenderMsg(NormObjectMsg* msg) return false; } } - block->TxInit(blockId, numData, session.SenderAutoParity()); + UINT16 autoParity = session.SenderAutoParity(); + if (session.GetEncoder() && session.GetEncoder()->IsRateless()) { + autoParity += session.GetEncoder()->CalculateProactiveParity(blockId.GetValue(), numData, session); + } + block->TxInit(blockId, numData, autoParity); //if (blockId < max_pending_block) if (Compare(blockId, max_pending_block) < 0) block->SetFlag(NormBlock::IN_REPAIR); @@ -2060,6 +2080,28 @@ bool NormObject::NextSenderMsg(NormObjectMsg* msg) CalculateBlockParity(block); } char* segment = block->GetSegment(segmentId); + if (session.GetEncoder() && session.GetEncoder()->IsRateless()) + { + if (NULL == segment) + { + segment = session.SenderGetFreeSegment(transport_id, blockId); + if (segment) + { + UINT16 payloadMax = segment_size + NormDataMsg::GetStreamPayloadHeaderLength(); +#ifdef SIMULATE + payloadMax = MIN(payloadMax, SIM_PAYLOAD_MAX); +#endif // SIMULATE + memset(segment, 0, payloadMax); + session.SenderEncodeParity(segmentId - numData, segment); + block->AttachSegment(segmentId, segment); + } + else + { + PLOG(PL_INFO, "NormObject::NextSenderMsg() node>%lu warning: sender resource constrained (no free segments for on-demand parity).\n", (unsigned long)LocalNodeId()); + return false; + } + } + } ASSERT(NULL != segment); // We only need to send FEC content to cover the biggest segment // sent for the block. @@ -2202,7 +2244,7 @@ NormBlockId NormStreamObject::RepairWindowLo() const bool NormObject::CalculateBlockParity(NormBlock* block) { - if (0 == nparity) return true; + if (0 == nparity && (!session.GetEncoder() || !session.GetEncoder()->IsRateless())) return true; char buffer[NormMsg::MAX_SIZE]; UINT16 numData = GetBlockSize(block->GetId()); for (UINT16 i = 0; i < numData; i++) @@ -2238,7 +2280,8 @@ NormBlock* NormObject::SenderRecoverBlock(NormBlockId blockId) block->TxRecover(blockId, numData, nparity); // Fill block with zero initialized parity segments UINT16 totalBlockLen = numData + nparity; - for (UINT16 i = numData; i < totalBlockLen; i++) + bool doPreallocate = (!session.GetEncoder() || !session.GetEncoder()->IsRateless()); + for (UINT16 i = numData; doPreallocate && (i < totalBlockLen); i++) { char* s = session.SenderGetFreeSegment(transport_id, blockId); if (s) diff --git a/src/common/normSegment.cpp b/src/common/normSegment.cpp index e2a2fb7d..d31a2da0 100755 --- a/src/common/normSegment.cpp +++ b/src/common/normSegment.cpp @@ -1,4 +1,5 @@ #include "normSegment.h" +#include "normEncoder.h" NormSegmentPool::NormSegmentPool() : seg_size(0), seg_count(0), seg_total(0), seg_list(NULL), seg_pool(NULL), @@ -528,8 +529,35 @@ bool NormBlock::AppendRepairRequest(NormNackMsg& nack, UINT16 numParity, NormObjectId objectId, bool pendingInfo, - UINT16 payloadMax) + UINT16 payloadMax, + UINT16 fecOverhead, + bool isRateless) { + if (isRateless) + { + int needed = (int)erasure_count - (int)parity_count + (int)fecOverhead; + if (needed <= 0) return false; + + UINT16 needed_clamped = (needed > 65535) ? 65535 : (UINT16)needed; + + NormRepairRequest req; + nack.AttachRepairRequest(req, payloadMax); + req.SetForm(NormRepairRequest::ERASURES); + req.SetFlag(NormRepairRequest::SEGMENT); + if (pendingInfo) req.SetFlag(NormRepairRequest::INFO); + + if (req.AppendErasureCount(fecId, fecM, objectId, blk_id, numData, needed_clamped)) + { + if (0 == nack.PackRepairRequest(req)) + { + PLOG(PL_WARN, "NormBlock::AppendRepairRequest() warning: full NACK msg\n"); + return false; + } + return true; + } + return false; + } + bool requestAppended = false; NormSegmentId nextId = 0; NormSegmentId endId; diff --git a/src/common/normSession.cpp b/src/common/normSession.cpp index 6c738559..c06edd55 100755 --- a/src/common/normSession.cpp +++ b/src/common/normSession.cpp @@ -133,6 +133,11 @@ NormSession::NormSession(NormSessionMgr &sessionMgr, NormNodeId localNodeId) user_timer.SetListener(this, &NormSession::OnUserTimeout); user_timer.SetInterval(0.0); user_timer.SetRepeat(0); + + memset(encoder_factories, 0, sizeof(encoder_factories)); + memset(decoder_factories, 0, sizeof(decoder_factories)); + memset(is_rateless_codec, 0, sizeof(is_rateless_codec)); + is_rateless_codec[NormPayloadId::RL] = true; // Built-in RL codec } NormSession::~NormSession() @@ -147,6 +152,23 @@ NormSession::~NormSession() Close(); } +void NormSession::RegisterFecCoder(UINT8 fecId, NormEncoderFactory encoderFactory, NormDecoderFactory decoderFactory, bool isRateless) +{ + encoder_factories[fecId] = encoderFactory; + decoder_factories[fecId] = decoderFactory; + is_rateless_codec[fecId] = isRateless; +} + +NormEncoderFactory NormSession::GetEncoderFactory(UINT8 fecId) const +{ + return encoder_factories[fecId]; +} + +NormDecoderFactory NormSession::GetDecoderFactory(UINT8 fecId) const +{ + return decoder_factories[fecId]; +} + bool NormSession::Open() { ASSERT(address.IsValid()); @@ -836,7 +858,18 @@ bool NormSession::StartSender(UINT16 instanceId, if (NULL != encoder) delete encoder; - if (blockSize <= 255) + NormEncoderFactory encoderFactory = GetEncoderFactory(fecId); + if (NULL != encoderFactory) + { + if (NULL == (encoder = encoderFactory())) + { + PLOG(PL_FATAL, "NormSession::StartSender() new custom encoder error\n"); + StopSender(); + return false; + } + fec_id = fecId; + } + else if (blockSize <= 255) { #ifdef ASSUME_MDP_FEC if (NULL == (encoder = new NormEncoderMDP)) @@ -4036,6 +4069,15 @@ void NormSession::SenderHandleNackMessage(const struct timeval ¤tTime, Nor } break; case SEGMENT: + { + UINT16 numErasuresVal = 0; + if (NormRepairRequest::ERASURES == requestForm) + { + numErasuresVal = nextSegmentId; + if (numErasuresVal > nparity) numErasuresVal = nparity; + nextSegmentId = ndata; + lastSegmentId = ndata; + } PLOG(PL_DETAIL, "NormSession::SenderHandleNackMessage(SEGMENT) obj>%hu blk>%lu segs>%hu:%hu\n", (UINT16)nextObjectId, (unsigned long)nextBlockId.GetValue(), (UINT16)nextSegmentId, (UINT16)lastSegmentId); @@ -4098,12 +4140,13 @@ void NormSession::SenderHandleNackMessage(const struct timeval ¤tTime, Nor { // mark nack time for potential flow control static_cast(object)->SetLastNackTime(nextBlockId, ProtoTime(currentTime)); - if (nextSegmentId < ndata) + if ((nextSegmentId < ndata) || (NormRepairRequest::ERASURES == requestForm)) { bool attemptLock = true; - NormSegmentId firstLockId = nextSegmentId; + NormSegmentId firstLockId = (NormRepairRequest::ERASURES == requestForm) ? 0 : nextSegmentId; NormSegmentId lastLockId = ndata - 1; - lastLockId = MIN(lastLockId, lastSegmentId); + if (NormRepairRequest::ERASURES != requestForm) + lastLockId = MIN(lastLockId, lastSegmentId); if (holdoff) { if (nextObjectId == txObjectIndex) @@ -4164,7 +4207,10 @@ void NormSession::SenderHandleNackMessage(const struct timeval ¤tTime, Nor // With a series of SEGMENT repair requests for a block, "numErasures" will // eventually total the number of missing segments in the block. - numErasures += (lastSegmentId - nextSegmentId + 1); + if (NormRepairRequest::ERASURES == requestForm) + numErasures = numErasuresVal; + else + numErasures += (lastSegmentId - nextSegmentId + 1); if (holdoff) { if (nextObjectId > txObjectIndex) @@ -4238,11 +4284,16 @@ void NormSession::SenderHandleNackMessage(const struct timeval ¤tTime, Nor tx_repair_block_min = nextBlockId; tx_repair_segment_min = (nextSegmentId < nextBlockSize) ? nextSegmentId : (nextBlockSize - 1); } + if (GetEncoder() && GetEncoder()->IsRateless()) { + numErasures = GetEncoder()->CalculateReactiveParity(nextBlockId.GetValue(), numErasures, *this); + if (numErasures > nparity) numErasures = nparity; + } block->HandleSegmentRequest(nextSegmentId, lastSegmentId, nextBlockSize, nparity, numErasures); startTimer = true; } // end if/else (holdoff) + } break; case INFO: // We already dealt with INFO request above with respect to initiating repair From c7d943aea82e9f412f1418601bc11fb9d649a77b Mon Sep 17 00:00:00 2001 From: Joseph Calderon Date: Sun, 12 Jul 2026 20:19:06 -0700 Subject: [PATCH 2/9] move fec codec factory registry to NormSessionMgr --- include/normSession.h | 44 ++++++++++++++++++++++---------------- src/common/normApi.cpp | 2 +- src/common/normSession.cpp | 33 ++++++++++------------------ 3 files changed, 38 insertions(+), 41 deletions(-) diff --git a/include/normSession.h b/include/normSession.h index 5ee2d14b..a27350ce 100755 --- a/include/normSession.h +++ b/include/normSession.h @@ -75,6 +75,9 @@ class NormController const NormFecLayout* fec_layouts[256]; }; // end class NormController +typedef class NormEncoder* (*NormEncoderFactory)(); +typedef class NormDecoder* (*NormDecoderFactory)(); + class NormSessionMgr { friend class NormSession; @@ -116,20 +119,27 @@ class NormSessionMgr {data_free_func = freeFunc;} NormDataObject::DataFreeFunctionHandle GetDataFreeFunction() const {return data_free_func;} - - private: - ProtoTimerMgr& timer_mgr; - ProtoSocket::Notifier& socket_notifier; - ProtoChannel::Notifier* channel_notifier; - NormController* controller; + + // FEC codec factory registry (shared by all sessions this manager owns) + void RegisterFecCoder(UINT8 fecId, NormEncoderFactory encoderFactory, NormDecoderFactory decoderFactory, bool isRateless); + NormEncoderFactory GetEncoderFactory(UINT8 fecId) const {return encoder_factories[fecId];} + NormDecoderFactory GetDecoderFactory(UINT8 fecId) const {return decoder_factories[fecId];} + bool IsRatelessFec(UINT8 fecId) const {return is_rateless_codec[fecId];} + + private: + ProtoTimerMgr& timer_mgr; + ProtoSocket::Notifier& socket_notifier; + ProtoChannel::Notifier* channel_notifier; + NormController* controller; NormDataObject::DataFreeFunctionHandle data_free_func; - + class NormSession* top_session; // top of NormSession list - -}; // end class NormSessionMgr -typedef class NormEncoder* (*NormEncoderFactory)(); -typedef class NormDecoder* (*NormDecoderFactory)(); + NormEncoderFactory encoder_factories[256]; + NormDecoderFactory decoder_factories[256]; + bool is_rateless_codec[256]; + +}; // end class NormSessionMgr class NormSession : public NormTelemetryContext { @@ -473,10 +483,11 @@ class NormSession : public NormTelemetryContext void SenderSetExtraParity(UINT16 extraParity) {extra_parity = extraParity;} - void RegisterFecCoder(UINT8 fecId, NormEncoderFactory encoderFactory, NormDecoderFactory decoderFactory, bool isRateless); - NormEncoderFactory GetEncoderFactory(UINT8 fecId) const; - NormDecoderFactory GetDecoderFactory(UINT8 fecId) const; - bool IsRatelessFec(UINT8 fecId) const { return is_rateless_codec[fecId]; } + // FEC codec factories are registered/owned by the NormSessionMgr so a + // single registry is shared across all sessions of a NORM API instance. + NormEncoderFactory GetEncoderFactory(UINT8 fecId) const {return session_mgr.GetEncoderFactory(fecId);} + NormDecoderFactory GetDecoderFactory(UINT8 fecId) const {return session_mgr.GetDecoderFactory(fecId);} + bool IsRatelessFec(UINT8 fecId) const {return session_mgr.IsRatelessFec(fecId);} NormEncoder* GetEncoder() const { return encoder; } @@ -817,9 +828,6 @@ class NormSession : public NormTelemetryContext NormBlockPool block_pool; NormSegmentPool segment_pool; NormEncoder* encoder; - NormEncoderFactory encoder_factories[256]; - NormDecoderFactory decoder_factories[256]; - bool is_rateless_codec[256]; UINT8 fec_id; UINT8 fec_m; INT32 fec_block_mask; diff --git a/src/common/normApi.cpp b/src/common/normApi.cpp index 2fc81469..29b071b9 100755 --- a/src/common/normApi.cpp +++ b/src/common/normApi.cpp @@ -1689,7 +1689,7 @@ bool NormRegisterFecCoder(NormSessionHandle sessionHandle, NormSession* session = (NormSession*)sessionHandle; if (session) { - session->RegisterFecCoder(fecId, encoderFactory, decoderFactory, isRateless); + session->GetSessionMgr().RegisterFecCoder(fecId, encoderFactory, decoderFactory, isRateless); instance->dispatcher.ResumeThread(); return true; } diff --git a/src/common/normSession.cpp b/src/common/normSession.cpp index c06edd55..d4ce9cae 100755 --- a/src/common/normSession.cpp +++ b/src/common/normSession.cpp @@ -133,11 +133,6 @@ NormSession::NormSession(NormSessionMgr &sessionMgr, NormNodeId localNodeId) user_timer.SetListener(this, &NormSession::OnUserTimeout); user_timer.SetInterval(0.0); user_timer.SetRepeat(0); - - memset(encoder_factories, 0, sizeof(encoder_factories)); - memset(decoder_factories, 0, sizeof(decoder_factories)); - memset(is_rateless_codec, 0, sizeof(is_rateless_codec)); - is_rateless_codec[NormPayloadId::RL] = true; // Built-in RL codec } NormSession::~NormSession() @@ -152,23 +147,6 @@ NormSession::~NormSession() Close(); } -void NormSession::RegisterFecCoder(UINT8 fecId, NormEncoderFactory encoderFactory, NormDecoderFactory decoderFactory, bool isRateless) -{ - encoder_factories[fecId] = encoderFactory; - decoder_factories[fecId] = decoderFactory; - is_rateless_codec[fecId] = isRateless; -} - -NormEncoderFactory NormSession::GetEncoderFactory(UINT8 fecId) const -{ - return encoder_factories[fecId]; -} - -NormDecoderFactory NormSession::GetDecoderFactory(UINT8 fecId) const -{ - return decoder_factories[fecId]; -} - bool NormSession::Open() { ASSERT(address.IsValid()); @@ -5841,8 +5819,19 @@ NormSessionMgr::NormSessionMgr(ProtoTimerMgr &timerMgr, : timer_mgr(timerMgr), socket_notifier(socketNotifier), channel_notifier(channelNotifier), controller(NULL), data_free_func(NULL), top_session(NULL) { + memset(encoder_factories, 0, sizeof(encoder_factories)); + memset(decoder_factories, 0, sizeof(decoder_factories)); + memset(is_rateless_codec, 0, sizeof(is_rateless_codec)); + is_rateless_codec[NormPayloadId::RL] = true; // Built-in RL codec } +void NormSessionMgr::RegisterFecCoder(UINT8 fecId, NormEncoderFactory encoderFactory, NormDecoderFactory decoderFactory, bool isRateless) +{ + encoder_factories[fecId] = encoderFactory; + decoder_factories[fecId] = decoderFactory; + is_rateless_codec[fecId] = isRateless; +} // end NormSessionMgr::RegisterFecCoder() + NormSessionMgr::~NormSessionMgr() { Destroy(); From 2d653628ce7849120c53d558c5db106f295d1988 Mon Sep 17 00:00:00 2001 From: Joseph Calderon Date: Sun, 12 Jul 2026 20:19:34 -0700 Subject: [PATCH 3/9] 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 ba4b8a80..8921c225 100644 --- a/examples/normRatelessTest.cpp +++ b/examples/normRatelessTest.cpp @@ -1,40 +1,128 @@ #include #include + #include #include #include +#include +#include +#include + +// FEC ID for the (partially-specified) rateless code family. +static const UINT8 RATELESS_FEC_ID = 131; -class MockRatelessEncoder : public NormEncoder { -public: - virtual bool Init(unsigned int numData, unsigned int numParity, UINT16 vectorSize) { - ndata = numData; - vec_size = vectorSize; - return true; +// 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; } +} + +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; + 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(); } + + 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; - UINT16 vec_size; + 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(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; } - - 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"); + + // 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("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"); + 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 42161989..bd0ed732 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 cb685273..2a437c55 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 a27350ce..bcc0c874 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 eca1717f..5fdd6e54 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 d31a2da0..10f381e9 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; From 6967fd69064bde1e416b911fbe6220b187c5e6fe Mon Sep 17 00:00:00 2001 From: Joseph Calderon Date: Tue, 11 Aug 2026 19:06:02 -0700 Subject: [PATCH 4/9] refine rateless FEC api and repair handling --- include/normApi.h | 14 +++- include/normEncoder.h | 15 +++-- include/normMessage.h | 48 ++++++++++--- include/normSegment.h | 11 ++- include/normSession.h | 42 ++++++------ src/common/normApi.cpp | 32 ++++++--- src/common/normMessage.cpp | 49 ++++++++++++++ src/common/normNode.cpp | 134 ++++++++++++++++++++----------------- src/common/normObject.cpp | 107 ++++++++++++++++++----------- src/common/normSegment.cpp | 80 ++++++++++++++-------- src/common/normSession.cpp | 47 +++++++------ 11 files changed, 378 insertions(+), 201 deletions(-) diff --git a/include/normApi.h b/include/normApi.h index e29b8b2b..ab38a8b1 100755 --- a/include/normApi.h +++ b/include/normApi.h @@ -866,15 +866,25 @@ class NormDecoder; typedef NormEncoder* (*NormEncoderFactory)(); typedef class NormDecoder* (*NormDecoderFactory)(); +// Register a custom FEC codec for "fecId" with the given NORM instance. The +// registry is shared by every session the instance owns, so this must be called +// before any session is created (it fails otherwise). NORM_API_LINKAGE -bool NormRegisterFecCoder(NormSessionHandle sessionHandle, +bool NormRegisterFecCoder(NormInstanceHandle instanceHandle, UINT8 fecId, NormEncoderFactory encoderFactory, NormDecoderFactory decoderFactory, bool isRateless); +// Describe the FEC payload id wire format for a custom "fecId" (the built-in codecs +// already know their own). Since "fecId" is the on-wire codec discriminator, layouts +// are keyed by it process-wide: any number of codecs may be registered under distinct +// ids, but claiming an id already held by a layout describing a different wire format +// fails. Re-registering an equivalent layout succeeds, so separate NORM instances in +// one process may each register the codecs they use. The layout is copied into +// library-owned storage and remains registered for process lifetime. NORM_API_LINKAGE -void NormRegisterFecLayout(NormInstanceHandle instance, UINT8 fecId, const NormFecLayout* layout); +bool NormRegisterFecLayout(NormInstanceHandle instance, UINT8 fecId, const NormFecLayout* layout); } // end extern "C" #endif /* __cplusplus */ diff --git a/include/normEncoder.h b/include/normEncoder.h index bd0ed732..fd374603 100755 --- a/include/normEncoder.h +++ b/include/normEncoder.h @@ -33,6 +33,7 @@ #ifndef _NORM_ENCODER #define _NORM_ENCODER +#include "normApi.h" #include "protokit.h" // protolib stuff class NormTelemetryContext @@ -40,19 +41,23 @@ class NormTelemetryContext public: virtual ~NormTelemetryContext() {} virtual double GetGrtt() const = 0; - virtual double GetTxRate() const = 0; + virtual double GetTxRateBits() const = 0; // bits/sec + virtual double GetTxRateBytes() const = 0; // bytes/sec + virtual double GetLossRate() const = 0; // CLR-reported loss fraction, 0.0 if none + // Sender's advertised group size estimate (see NormSetGroupSize()), which is a + // configured value rather than measured receiver membership. virtual unsigned int GetGroupSize() const = 0; virtual double GetCurrentTime() const = 0; }; -class NormEncoder +class NORM_API_LINKAGE NormEncoder { public: 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; // 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 @@ -64,7 +69,7 @@ class NormEncoder virtual UINT16 CalculateReactiveParity(unsigned int blockId, UINT16 requestedErasures, const NormTelemetryContext& context) { return requestedErasures; } }; // end class NormEncoder -class NormDecoder +class NORM_API_LINKAGE NormDecoder { public: virtual ~NormDecoder(); @@ -76,7 +81,7 @@ class NormDecoder // 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; + 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/normMessage.h b/include/normMessage.h index ddc8e95c..6c4237e7 100755 --- a/include/normMessage.h +++ b/include/normMessage.h @@ -9,6 +9,7 @@ #include // for memcpy(), etc #include #include // for rand(), etc +#include #ifdef _WIN32_WCE #include @@ -406,8 +407,25 @@ class NormPayloadId SB = 129, // partially-specified "small block" codes RL = 131 // partially-specified "rateless" codes }; - static bool IsValid(UINT8 fecId, const NormFecLayout* layout = NULL) - { + // Several NormPayloadId call sites (FTI/NACK parsing, block mask setup) have no + // session or controller handy to supply the layout for a custom "fecId", so the + // layouts registered via NormRegisterFecLayout() are also kept here and used + // whenever the caller doesn't pass one explicitly. + // + // The table is keyed by "fecId", which is the on-wire codec discriminator, so + // any number of codecs coexist here as long as they use distinct ids. Claiming + // an id already held by a layout describing a *different* wire format is refused: + // that would mean two payload id formats sharing one discriminator, which cannot + // interoperate on a group no matter how the layouts are scoped. Re-registering + // an equivalent layout (as separate NORM instances in one process do) is allowed. + // Published layouts are copied into library-owned storage and remain valid for + // process lifetime. A NULL layout succeeds only when the id is unclaimed. + static bool SetFecLayout(UINT8 fecId, const NormFecLayout* layout); + static const NormFecLayout* GetFecLayout(UINT8 fecId) {return fec_layouts[fecId].load();} + + static bool IsValid(UINT8 fecId) + { + const NormFecLayout* layout = fec_layouts[fecId].load(); if (layout != NULL) return true; switch (fecId) @@ -421,14 +439,14 @@ class NormPayloadId return false; } } - NormPayloadId(UINT8 fecId, UINT8 m, UINT32* theBuffer) : fec_id(fecId), fec_m(m), buffer(theBuffer) {} NormPayloadId(UINT8 fecId, UINT8 m, const UINT32* theBuffer) : fec_id(fecId), fec_m(m), cbuffer(theBuffer) {} - static UINT16 GetLength(UINT8 fecId, const NormFecLayout* layout = NULL) + static UINT16 GetLength(UINT8 fecId) { + const NormFecLayout* layout = fec_layouts[fecId].load(); if (layout != NULL) return layout->payloadIdLength; switch (fecId) @@ -444,8 +462,9 @@ class NormPayloadId } } - static UINT32 GetFecBlockMask(UINT8 fecId, UINT8 fecM, const NormFecLayout* layout = NULL) + static UINT32 GetFecBlockMask(UINT8 fecId, UINT8 fecM) { + const NormFecLayout* layout = fec_layouts[fecId].load(); if (layout != NULL) return layout->blockMask; switch (fecId) @@ -466,8 +485,9 @@ class NormPayloadId } } - void SetFecPayloadId(UINT32 blockId, UINT16 symbolId, UINT16 blockLen, const NormFecLayout* layout = NULL) + void SetFecPayloadId(UINT32 blockId, UINT16 symbolId, UINT16 blockLen) { + const NormFecLayout* layout = fec_layouts[fec_id].load(); if (layout != NULL && layout->packPayloadId != NULL) { layout->packPayloadId(buffer, blockId, symbolId, blockLen); @@ -499,7 +519,7 @@ class NormPayloadId ptr[0] = htons(blockLen); // 2 bytes ptr[1] = htons(symbolId); // 2 bytes break; - } + } case RL: { UINT16* payloadId = (UINT16*)buffer; @@ -511,8 +531,9 @@ class NormPayloadId } // Message processing methods - NormBlockId GetFecBlockId(const NormFecLayout* layout = NULL) const + NormBlockId GetFecBlockId() const { + const NormFecLayout* layout = fec_layouts[fec_id].load(); if (layout != NULL && layout->unpackBlockId != NULL) { return layout->unpackBlockId(cbuffer); @@ -548,8 +569,9 @@ class NormPayloadId } } - UINT16 GetFecSymbolId(const NormFecLayout* layout = NULL) const + UINT16 GetFecSymbolId() const { + const NormFecLayout* layout = fec_layouts[fec_id].load(); if (layout != NULL && layout->unpackSymbolId != NULL) { return layout->unpackSymbolId(cbuffer); @@ -588,8 +610,9 @@ class NormPayloadId } } - UINT16 GetFecBlockLength(const NormFecLayout* layout = NULL) const + UINT16 GetFecBlockLength() const { + const NormFecLayout* layout = fec_layouts[fec_id].load(); if (layout != NULL && layout->unpackBlockLength != NULL) { return layout->unpackBlockLength(cbuffer); @@ -604,9 +627,12 @@ class NormPayloadId return 0; } } - + private: + // Registered via NormRegisterFecLayout(); read on the rx thread, so atomic. + static std::atomic fec_layouts[256]; + UINT8 fec_id; UINT8 fec_m; union diff --git a/include/normSegment.h b/include/normSegment.h index 2a437c55..997b0dc5 100755 --- a/include/normSegment.h +++ b/include/normSegment.h @@ -95,6 +95,15 @@ class NormBlock segment_table[sid] = (char*)NULL; return segment; } + // Are the first "count" segment vectors all present? A rateless sender needs + // every source vector cached before it can synthesize a repair symbol from them. + bool HasSegments(UINT16 count) const + { + ASSERT(count <= size); + for (UINT16 i = 0; i < count; i++) + if (NULL == segment_table[i]) return false; + return true; + } void SetSegment(NormSegmentId sid, char* segment) { ASSERT(sid < size); @@ -243,7 +252,7 @@ class NormBlock NormBlockId finalBlockId, UINT16 finalSegmentSize) const; - bool AppendRepairRequest(class NormNackMsg& nack, + bool AppendRepairRequest(class NormNackMsg& nack, UINT8 fecId, UINT8 fecM, UINT16 numData, diff --git a/include/normSession.h b/include/normSession.h index bcc0c874..be822978 100755 --- a/include/normSession.h +++ b/include/normSession.h @@ -4,9 +4,9 @@ #include "normMessage.h" #include "normObject.h" #include "normNode.h" - #include "normEncoder.h" -#include + +#include "protokit.h" #include "protoCap.h" // for ProtoCap for ECN_SUPPORT @@ -27,10 +27,6 @@ class NormController { public: virtual ~NormController() {} - NormController() { memset(fec_layouts, 0, sizeof(fec_layouts)); } - - void SetFecLayout(UINT8 id, const NormFecLayout* layout) { fec_layouts[id] = layout; } - const NormFecLayout* GetFecLayout(UINT8 id) const { return fec_layouts[id]; } enum Event { EVENT_INVALID = 0, @@ -72,7 +68,6 @@ class NormController class NormNode* node, class NormObject* object) = 0; - const NormFecLayout* fec_layouts[256]; }; // end class NormController typedef class NormEncoder* (*NormEncoderFactory)(); @@ -119,6 +114,8 @@ class NormSessionMgr {data_free_func = freeFunc;} NormDataObject::DataFreeFunctionHandle GetDataFreeFunction() const {return data_free_func;} + + bool HasSessions() const {return (NULL != top_session);} // FEC codec factory registry (shared by all sessions this manager owns) void RegisterFecCoder(UINT8 fecId, NormEncoderFactory encoderFactory, NormDecoderFactory decoderFactory, bool isRateless); @@ -126,15 +123,15 @@ class NormSessionMgr NormDecoderFactory GetDecoderFactory(UINT8 fecId) const {return decoder_factories[fecId];} bool IsRatelessFec(UINT8 fecId) const {return is_rateless_codec[fecId];} - private: - ProtoTimerMgr& timer_mgr; - ProtoSocket::Notifier& socket_notifier; - ProtoChannel::Notifier* channel_notifier; - NormController* controller; + private: + ProtoTimerMgr& timer_mgr; + ProtoSocket::Notifier& socket_notifier; + ProtoChannel::Notifier* channel_notifier; + NormController* controller; NormDataObject::DataFreeFunctionHandle data_free_func; - + class NormSession* top_session; // top of NormSession list - + NormEncoderFactory encoder_factories[256]; NormDecoderFactory decoder_factories[256]; bool is_rateless_codec[256]; @@ -144,7 +141,6 @@ class NormSessionMgr class NormSession : public NormTelemetryContext { friend class NormSessionMgr; - public: enum {DEFAULT_MESSAGE_POOL_DEPTH = 16}; @@ -547,15 +543,17 @@ class NormSession : public NormTelemetryContext void SenderSetFtiMode(FtiMode ftiMode) {fti_mode = ftiMode;} - void SetReceiverMaxDelay(unsigned int msec) - {rcvr_max_delay = msec;} - unsigned int GetReceiverMaxDelay() const - {return rcvr_max_delay;} - // NormTelemetryContext Implementation double GetGrtt() const override { return SenderGrtt(); } - double GetTxRate() const override { return tx_rate; } - virtual unsigned int GetGroupSize() const override { return (unsigned int)((NormNodeList*)&cc_node_list)->GetCount(); } + double GetTxRateBits() const override { return tx_rate * 8.0; } + double GetTxRateBytes() const override { return tx_rate; } + double GetLossRate() const override + { + const NormCCNode* clr = (const NormCCNode*)((NormNodeList*)&cc_node_list)->Head(); + return (NULL != clr) ? clr->GetLoss() : 0.0; + } + virtual unsigned int GetGroupSize() const override + { return (unsigned int)(gsize_advertised + 0.5); } double GetCurrentTime() const override { ProtoTime t; t.GetCurrentTime(); return t.GetValue(); } void SenderEncode(unsigned int segmentId, const char* segment, char** parityVectorList) diff --git a/src/common/normApi.cpp b/src/common/normApi.cpp index 29b071b9..67127fa5 100755 --- a/src/common/normApi.cpp +++ b/src/common/normApi.cpp @@ -1,4 +1,6 @@ +#ifndef _NORM_API_BUILD #define _NORM_API_BUILD // force 'dllexport' in "normApi.h" +#endif // !_NORM_API_BUILD #include "normApi.h" #include "normSession.h" @@ -1677,33 +1679,43 @@ void NormSetAutoParity(NormSessionHandle sessionHandle, unsigned char autoParity } // end NormSetAutoParity() NORM_API_LINKAGE -bool NormRegisterFecCoder(NormSessionHandle sessionHandle, +bool NormRegisterFecCoder(NormInstanceHandle instanceHandle, UINT8 fecId, NormEncoderFactory encoderFactory, NormDecoderFactory decoderFactory, bool isRateless) { - NormInstance* instance = NormInstance::GetInstanceFromSession(sessionHandle); + NormInstance* instance = (NormInstance*)instanceHandle; if (instance && instance->dispatcher.SuspendThread()) { - NormSession* session = (NormSession*)sessionHandle; - if (session) + if ((NULL == encoderFactory) || (NULL == decoderFactory)) + { + instance->dispatcher.ResumeThread(); + return false; + } + // The registry is instance-wide, so refuse to swap a codec out from + // underneath sessions that may already be encoding/decoding with it. + if (instance->session_mgr.HasSessions()) { - session->GetSessionMgr().RegisterFecCoder(fecId, encoderFactory, decoderFactory, isRateless); instance->dispatcher.ResumeThread(); - return true; + return false; } + instance->session_mgr.RegisterFecCoder(fecId, encoderFactory, decoderFactory, isRateless); instance->dispatcher.ResumeThread(); + return true; } return false; } // end NormRegisterFecCoder() NORM_API_LINKAGE -void NormRegisterFecLayout(NormInstanceHandle instanceHandle, UINT8 fecId, const NormFecLayout* layout) +bool NormRegisterFecLayout(NormInstanceHandle instanceHandle, UINT8 fecId, const NormFecLayout* layout) { - NormInstance* instance = (NormInstance*)instanceHandle; - if (instance) - instance->SetFecLayout(fecId, layout); + // Validate before touching the process-wide table, so a bad handle can't leave a + // registration behind. + if (NULL == (NormInstance*)instanceHandle) return false; + // Layouts are keyed by fecId process-wide: the NormPayloadId call sites that need + // them (FTI/NACK parsing, block mask setup) have no instance context to consult. + return NormPayloadId::SetFecLayout(fecId, layout); } // end NormRegisterFecLayout() NORM_API_LINKAGE diff --git a/src/common/normMessage.cpp b/src/common/normMessage.cpp index 4a01e099..abede052 100755 --- a/src/common/normMessage.cpp +++ b/src/common/normMessage.cpp @@ -1,4 +1,53 @@ #include "normMessage.h" +#include + +std::atomic NormPayloadId::fec_layouts[256] = {}; +static NormFecLayout norm_fec_layout_storage[256] = {}; +static std::mutex norm_fec_layout_mutex; + +// Do two layouts describe the same payload id wire format? Compared field-wise +// rather than by memcmp() so struct padding can't make equivalent layouts differ. +static bool NormFecLayoutsEquivalent(const NormFecLayout* a, const NormFecLayout* b) +{ + return ((a->payloadIdLength == b->payloadIdLength) && + (a->blockMask == b->blockMask) && + (a->packPayloadId == b->packPayloadId) && + (a->unpackBlockId == b->unpackBlockId) && + (a->unpackSymbolId == b->unpackSymbolId) && + (a->unpackBlockLength == b->unpackBlockLength)); +} // end NormFecLayoutsEquivalent() + +bool NormPayloadId::SetFecLayout(UINT8 fecId, const NormFecLayout* layout) +{ + if (NULL == layout) + { + // Published layouts may be in use by any NORM instance in this process. + // Keep them immutable for process lifetime so readers and callback targets + // cannot be invalidated underneath packet processing. + return (NULL == fec_layouts[fecId].load()); + } + if ((0 == layout->payloadIdLength) || (0 != (layout->payloadIdLength & 0x03)) || + (0 == layout->blockMask) || (NULL == layout->packPayloadId) || + (NULL == layout->unpackBlockId) || (NULL == layout->unpackSymbolId)) + { + PLOG(PL_ERROR, "NormPayloadId::SetFecLayout() error: invalid layout for fec_id %u\n", fecId); + return false; + } + + std::lock_guard lock(norm_fec_layout_mutex); + const NormFecLayout* current = fec_layouts[fecId].load(); + if (NULL == current) + { + norm_fec_layout_storage[fecId] = *layout; + fec_layouts[fecId].store(&norm_fec_layout_storage[fecId]); + return true; + } + if (NormFecLayoutsEquivalent(current, layout)) + return true; // same wire format, so nothing to disagree about + PLOG(PL_ERROR, "NormPayloadId::SetFecLayout() error: fec_id %u is already registered " + "with a different payload id layout\n", fecId); + return false; +} // end NormPayloadId::SetFecLayout() NormHeaderExtension::NormHeaderExtension() diff --git a/src/common/normNode.cpp b/src/common/normNode.cpp index 34e8c561..b4dc5231 100755 --- a/src/common/normNode.cpp +++ b/src/common/normNode.cpp @@ -221,10 +221,20 @@ bool NormSenderNode::AllocateBuffers(unsigned int bufferSpace, unsigned long segPerBlock = (unsigned long) ((bufferFactor * (double)numData) + ((1.0 - bufferFactor) * (double)numParity) + 0.5); - if (segPerBlock > numData) segPerBlock = numData; - // If there's no parity, no segment buffering for decoding is required at all! - // (Thus, the full rxbuffer space can be used for block state) - if (0 == numParity) segPerBlock = 0; + if (session.IsRatelessFec(fecId)) + { + // Rateless decoding is not "one erasure per repair symbol": a decode attempt can + // fall short and need further repair symbols, so the receiver must retain every + // symbol it has received for the block (source and repair alike) across attempts. + segPerBlock = numData + numParity; + } + else + { + if (segPerBlock > numData) segPerBlock = numData; + // If there's no parity, no segment buffering for decoding is required at all! + // (Thus, the full rxbuffer space can be used for block state) + if (0 == numParity) segPerBlock = 0; + } unsigned long blockSegmentSpace = segPerBlock * (segmentSize + NormDataMsg::GetStreamPayloadHeaderLength()); unsigned long blockSpace = blockStateSpace + blockSegmentSpace; unsigned long numBlocks = bufferSpace / blockSpace; @@ -297,75 +307,72 @@ bool NormSenderNode::AllocateBuffers(unsigned int bufferSpace, return false; } } - else + else switch (fecId) // no registered codec, so use the built-in one for "fecId" { - switch (fecId) - { - case 2: - if (8 == fecM) - { - if (NULL == (decoder = new NormDecoderRS8)) - { - PLOG(PL_FATAL, "NormSenderNode::AllocateBuffers() new NormDecoderRS8 error: %s\n", GetErrorString()); - Close(); - return false; - } - } - else if (16 == fecM) - { - if (NULL == (decoder = new NormDecoderRS16)) - { - PLOG(PL_FATAL, "NormSenderNode::AllocateBuffers() new NormDecoderRS16 error: %s\n", GetErrorString()); - Close(); - return false; - } - } - else - { - PLOG(PL_FATAL, "NormSenderNode::AllocateBuffers() error: unsupported fecId=2 'm' value %d!\n", fecM); - Close(); - return false; - } - break; - case 5: + case 2: + if (8 == fecM) + { if (NULL == (decoder = new NormDecoderRS8)) { PLOG(PL_FATAL, "NormSenderNode::AllocateBuffers() new NormDecoderRS8 error: %s\n", GetErrorString()); Close(); return false; } - break; - case 129: - #ifdef ASSUME_MDP_FEC - if (NULL == (decoder = new NormDecoderMDP)) + } + else if (16 == fecM) + { + if (NULL == (decoder = new NormDecoderRS16)) { - PLOG(PL_FATAL, "NormSenderNode::AllocateBuffers() new NormDecoderMDP error: %s\n", GetErrorString()); + PLOG(PL_FATAL, "NormSenderNode::AllocateBuffers() new NormDecoderRS16 error: %s\n", GetErrorString()); Close(); return false; } - #else - if (0 == fecInstanceId) - { - if (NULL == (decoder = new NormDecoderRS8)) - { - PLOG(PL_FATAL, "NormSenderNode::AllocateBuffers() new NormDecoderRS8 error: %s\n", GetErrorString()); - Close(); - return false; - } - } - else + } + else + { + PLOG(PL_FATAL, "NormSenderNode::AllocateBuffers() error: unsupported fecId=2 'm' value %d!\n", fecM); + Close(); + return false; + } + break; + case 5: + if (NULL == (decoder = new NormDecoderRS8)) + { + PLOG(PL_FATAL, "NormSenderNode::AllocateBuffers() new NormDecoderRS8 error: %s\n", GetErrorString()); + Close(); + return false; + } + break; + case 129: +#ifdef ASSUME_MDP_FEC + if (NULL == (decoder = new NormDecoderMDP)) + { + PLOG(PL_FATAL, "NormSenderNode::AllocateBuffers() new NormDecoderMDP error: %s\n", GetErrorString()); + Close(); + return false; + } +#else + if (0 == fecInstanceId) + { + if (NULL == (decoder = new NormDecoderRS8)) { - PLOG(PL_FATAL, "NormSenderNode::AllocateBuffers() error: unknown fecId=129 instanceId!\n"); + PLOG(PL_FATAL, "NormSenderNode::AllocateBuffers() new NormDecoderRS8 error: %s\n", GetErrorString()); Close(); - return false; + return false; } - #endif // if/else ASSUME_MDP_FEC - break; - default: - PLOG(PL_FATAL, "NormSenderNode::AllocateBuffers() error: unknown fecId>%d!\n", fecId); + } + else + { + PLOG(PL_FATAL, "NormSenderNode::AllocateBuffers() error: unknown fecId=129 instanceId!\n"); Close(); - return false; - } + return false; + } +#endif // if/else ASSUME_MDP_FEC + break; + default: + PLOG(PL_FATAL, "NormSenderNode::AllocateBuffers() error: unknown fecId>%d!\n", fecId); + Close(); + return false; } if (!decoder->Init(numData, numParity, segmentSize+NormDataMsg::GetStreamPayloadHeaderLength())) { @@ -1444,8 +1451,13 @@ bool NormSenderNode::GetFtiData(const NormObjectMsg& msg, NormFtiData& ftiData) } break; } - case 131: + default: { + // The rateless scheme (131) and any codec registered for a custom fecId + // share these generic FTI fields; only the FEC payload id format differs, + // and that comes from NormRegisterFecLayout() rather than the FTI. + if ((131 != fecId) && (NULL == session.GetDecoderFactory(fecId))) + break; NormFtiExtension131 fti; while (msg.GetNextExtension(fti)) { @@ -1462,10 +1474,6 @@ bool NormSenderNode::GetFtiData(const NormObjectMsg& msg, NormFtiData& ftiData) } break; } - default: - PLOG(PL_ERROR, "NormSenderNode::GetFtiData() node>%lu sender>%lu unknown fec_id type:%d\n", - (unsigned long)LocalNodeId(), (unsigned long)GetId(), (int)fecId); - break; } // end switch (fecId) PLOG(PL_ERROR, "NormSenderNode::GetFtiData() node>%lu sender>%lu unknown fec_id type:%d\n", (unsigned long)LocalNodeId(), (unsigned long)GetId(), (int)fecId); diff --git a/src/common/normObject.cpp b/src/common/normObject.cpp index 5fdd6e54..895c0949 100755 --- a/src/common/normObject.cpp +++ b/src/common/normObject.cpp @@ -1320,14 +1320,14 @@ bool NormObject::AppendRepairRequest(NormNackMsg& nack, blockRequestAppended = block->AppendRepairRequest(nack, fec_id, fec_m, max_pending_segment, 0, transport_id, pending_info, payloadMax, - 2, isRateless); + 2, isRateless); } else { blockRequestAppended = block->AppendRepairRequest(nack, fec_id, fec_m, numData, nparity, transport_id, pending_info, payloadMax, - 2, isRateless); + 2, isRateless); } } if (blockRequestAppended) @@ -1621,32 +1621,32 @@ void NormObject::HandleObjectMessage(const NormObjectMsg& msg, ratelessDecodeIncomplete = session.IsRatelessFec(fec_id) && (0 != unrecovered); if (!ratelessDecodeIncomplete) { - for (UINT16 i = 0; i < erasureCount; i++) + for (UINT16 i = 0; i < erasureCount; i++) + { + NormSegmentId sid = sender->GetErasureLoc(i); + if (sid < numData) { - NormSegmentId sid = sender->GetErasureLoc(i); - if (sid < numData) + if (WriteSegment(blockId, sid, block->GetSegment(sid))) { - 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"); - } - } + objectUpdated = true; + // For statistics only (TBD) #ifdef NORM_DEBUG + // "segmentLength" is not necessarily correct here (TBD - fix this) + sender->IncrementRecvGoodput(segmentLength); + } else { - break; - } + if (IsStream()) + 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 @@ -1660,14 +1660,14 @@ void NormObject::HandleObjectMessage(const NormObjectMsg& msg, } } // 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)); if (!ratelessDecodeIncomplete) { - // OK, we're done with this block - pending_mask.Unset(blockId.GetValue()); - block_buffer.Remove(block); - sender->PutFreeBlock(block); + // 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 @@ -1878,8 +1878,16 @@ bool NormObject::NextSenderMsg(NormObjectMsg* msg) fti.SetFecNumParity(nparity); break; } - case 131: // RL + default: { + // The rateless scheme (131) and registered codecs use the same generic + // FTI fields. Their FEC payload-id format differs, but that is supplied + // separately by NormRegisterFecLayout(). + if ((131 != fec_id) && (NULL == session.GetEncoderFactory(fec_id))) + { + ASSERT(0); + return false; + } NormFtiExtension131 fti; msg->AttachExtension(fti); fti.SetObjectSize(object_size); @@ -1889,9 +1897,6 @@ bool NormObject::NextSenderMsg(NormObjectMsg* msg) fti.SetFecNumParity(nparity); break; } - default: - ASSERT(0); - return false; } } if (pending_info) @@ -1970,6 +1975,10 @@ bool NormObject::NextSenderMsg(NormObjectMsg* msg) UINT16 autoParity = session.SenderAutoParity(); if (session.GetEncoder() && session.GetEncoder()->IsRateless()) { autoParity += session.GetEncoder()->CalculateProactiveParity(blockId.GetValue(), numData, session); + // Clamp as SenderHandleNackMessage() does for CalculateReactiveParity(): + // autoParity indexes the block's parity space, and letting an app codec + // push it past nparity makes parity_offset overflow the segment table. + if (autoParity > nparity) autoParity = nparity; } block->TxInit(blockId, numData, autoParity); //if (blockId < max_pending_block) @@ -2092,8 +2101,8 @@ 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 @@ -2109,24 +2118,46 @@ bool NormObject::NextSenderMsg(NormObjectMsg* msg) } else { - PLOG(PL_INFO, "NormObject::NextSenderMsg() node>%lu warning: no free segment to cache " - "rateless source symbol.\n", (unsigned long)LocalNodeId()); + // Not fatal here -- the data segment itself is still sent. The block + // just can't synthesize repair symbols until every source vector is + // cached, which the HasSegments() guard below enforces. + PLOG(PL_WARN, "NormObject::NextSenderMsg() node>%lu warning: no free segment to cache " + "rateless source symbol (block %lu cannot produce parity).\n", + (unsigned long)LocalNodeId(), (unsigned long)block->GetId().GetValue()); } } } } else { + bool isRateless = (NULL != session.GetEncoder()) && session.GetEncoder()->IsRateless(); if (!block->ParityReady(numData)) { ASSERT(0 == block->ParityReadiness()); - CalculateBlockParity(block); + // For a rateless block this also caches the source vectors that + // EncodeParity() needs, so a failure here must not be ignored. + if (!CalculateBlockParity(block) && isRateless) + { + PLOG(PL_WARN, "NormObject::NextSenderMsg() node>%lu warning: could not prepare " + "rateless block %lu source symbols\n", + (unsigned long)LocalNodeId(), (unsigned long)blockId.GetValue()); + return false; + } } char* segment = block->GetSegment(segmentId); - if (session.GetEncoder() && session.GetEncoder()->IsRateless()) + if (isRateless) { if (NULL == segment) { + // Never hand the codec a partially-NULL source vector list -- a source + // symbol may have gone uncached above when the segment pool was dry. + if (!block->HasSegments(numData)) + { + PLOG(PL_WARN, "NormObject::NextSenderMsg() node>%lu warning: rateless block %lu " + "is missing cached source symbols, cannot synthesize parity\n", + (unsigned long)LocalNodeId(), (unsigned long)blockId.GetValue()); + return false; + } segment = session.SenderGetFreeSegment(transport_id, blockId); if (segment) { @@ -2325,7 +2356,7 @@ bool NormObject::CalculateBlockParity(NormBlock* block) } else { - return false; + return false; } } block->SetParityReadiness(numData); diff --git a/src/common/normSegment.cpp b/src/common/normSegment.cpp index 10f381e9..7190a3c8 100755 --- a/src/common/normSegment.cpp +++ b/src/common/normSegment.cpp @@ -214,10 +214,14 @@ bool NormBlock::IsRepairPending(UINT16 numData, UINT16 numParity) // original behavior). UINT16 parityNeed = erasure_count + decode_overhead; if (parityNeed > numParity) parityNeed = numParity; - repair_mask.SetBits(0, numData); + // Once every advertised repair symbol has arrived, a rateless decoder can + // still report a residual shortfall. Leave the missing source symbols + // visible so the next NACK can fall back to explicit source repair. + if (!((0 != decode_overhead) && (parity_count >= numParity))) + repair_mask.SetBits(0, numData); 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() @@ -539,33 +543,6 @@ bool NormBlock::AppendRepairRequest(NormNackMsg& nack, UINT16 fecOverhead, bool isRateless) { - if (isRateless) - { - // "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; - - NormRepairRequest req; - nack.AttachRepairRequest(req, payloadMax); - req.SetForm(NormRepairRequest::ERASURES); - req.SetFlag(NormRepairRequest::SEGMENT); - if (pendingInfo) req.SetFlag(NormRepairRequest::INFO); - - if (req.AppendErasureCount(fecId, fecM, objectId, blk_id, numData, needed_clamped)) - { - if (0 == nack.PackRepairRequest(req)) - { - PLOG(PL_WARN, "NormBlock::AppendRepairRequest() warning: full NACK msg\n"); - return false; - } - return true; - } - return false; - } - bool requestAppended = false; NormSegmentId nextId = 0; NormSegmentId endId; @@ -582,6 +559,51 @@ bool NormBlock::AppendRepairRequest(NormNackMsg& nack, } endId = numData + numParity; } + else if (isRateless) + { + // Nothing erased means nothing to repair -- don't let a non-zero "fecOverhead" + // conjure up a NACK for a block that is already complete. + if (0 == erasure_count) return false; + nextId = numData; + if (!GetNextPending(nextId) || (nextId >= (NormSegmentId)(numData + numParity))) + { + // The finite repair-symbol namespace is exhausted. Request the + // remaining source symbols explicitly instead of leaving the block + // pending forever. + nextId = 0; + if (!GetFirstPending(nextId) || (nextId >= numData)) return false; + endId = numData; + } + else + { + // A rateless code is not MDS: recovery takes "erasure_count" innovative + // repair symbols plus slack, and "decode_overhead" grows whenever a + // decode attempt fell short, so keep asking until the block decodes. + int needed = (int)erasure_count - (int)parity_count + + (int)fecOverhead + (int)decode_overhead; + if (needed <= 0) return false; + // Name the repair symbols we lack by index, rather than sending an ERASURES + // count. A count leaves the sender to pick indices from its own parity + // accounting, and a block rebuilt by SenderRecoverBlock() has none left + // (TxRecover() sets parity_offset = parity_count = nparity), so the count + // collapses to a single index and the sender re-sends repair symbol 0 forever. + // Select by count across the whole parity range -- unlike the MDS case below, + // the symbols we already hold do not reduce how many more we need, so the + // range we ask over cannot be capped at "numData + needed". + NormSegmentId lastId = nextId; + NormSegmentId scanId = nextId; + int selected = 0; + while (1) + { + lastId = scanId; + if (++selected >= needed) break; + scanId++; + if (!GetNextPending(scanId)) break; + if (scanId >= (NormSegmentId)(numData + numParity)) break; + } + endId = lastId + 1; + } + } else { nextId = numData; diff --git a/src/common/normSession.cpp b/src/common/normSession.cpp index d4ce9cae..71c62d43 100755 --- a/src/common/normSession.cpp +++ b/src/common/normSession.cpp @@ -761,6 +761,21 @@ bool NormSession::StartSender(UINT16 instanceId, UINT16 numParity, UINT8 fecId) { + if ((0 != fecId) && !NormPayloadId::IsValid(fecId)) + { + PLOG(PL_FATAL, "NormSession::StartSender() error: no payload-id layout registered " + "for fec_id %u (see NormRegisterFecLayout())\n", fecId); + return false; + } + // A rateless fecId has no built-in codec -- without a registered one we would + // silently build a Reed-Solomon encoder while advertising "fecId" on the wire, and + // every receiver would then reject the stream as an unknown fecId. Fail up front. + if (IsRatelessFec(fecId) && (NULL == GetEncoderFactory(fecId))) + { + PLOG(PL_FATAL, "NormSession::StartSender() error: no FEC codec registered for " + "rateless fec_id %u (see NormRegisterFecCoder())\n", fecId); + return false; + } UINT16 blockSize = numData + numParity; if (blockSize <= 255) fec_m = 8; @@ -805,17 +820,23 @@ bool NormSession::StartSender(UINT16 instanceId, unsigned long maskSize = blockSize >> 3; if (0 != (blockSize & 0x07)) maskSize++; + // A rateless sender synthesizes repair symbols on demand from the block's source + // symbols (see NormObject::CalculateBlockParity()), so it must also cache the + // source vectors -- budget numData + numParity segments per block instead of just + // the numParity a block code needs for its precomputed parity. + unsigned long segmentsPerBlock = IsRatelessFec(fecId) ? (numData + numParity) : numParity; + unsigned long blockSpace = sizeof(NormBlock) + blockSize * sizeof(char *) + 2 * maskSize + - numParity * (segmentSize + NormDataMsg::GetStreamPayloadHeaderLength()); + segmentsPerBlock * (segmentSize + NormDataMsg::GetStreamPayloadHeaderLength()); unsigned long numBlocks = bufferSpace / blockSpace; if (bufferSpace > (numBlocks * blockSpace)) numBlocks++; if (numBlocks < 2) numBlocks = 2; - unsigned long numSegments = numBlocks * numParity; + unsigned long numSegments = numBlocks * segmentsPerBlock; if (!block_pool.Init((UINT32)numBlocks, blockSize)) { @@ -4047,15 +4068,6 @@ void NormSession::SenderHandleNackMessage(const struct timeval ¤tTime, Nor } break; case SEGMENT: - { - UINT16 numErasuresVal = 0; - if (NormRepairRequest::ERASURES == requestForm) - { - numErasuresVal = nextSegmentId; - if (numErasuresVal > nparity) numErasuresVal = nparity; - nextSegmentId = ndata; - lastSegmentId = ndata; - } PLOG(PL_DETAIL, "NormSession::SenderHandleNackMessage(SEGMENT) obj>%hu blk>%lu segs>%hu:%hu\n", (UINT16)nextObjectId, (unsigned long)nextBlockId.GetValue(), (UINT16)nextSegmentId, (UINT16)lastSegmentId); @@ -4118,13 +4130,12 @@ void NormSession::SenderHandleNackMessage(const struct timeval ¤tTime, Nor { // mark nack time for potential flow control static_cast(object)->SetLastNackTime(nextBlockId, ProtoTime(currentTime)); - if ((nextSegmentId < ndata) || (NormRepairRequest::ERASURES == requestForm)) + if (nextSegmentId < ndata) { bool attemptLock = true; - NormSegmentId firstLockId = (NormRepairRequest::ERASURES == requestForm) ? 0 : nextSegmentId; + NormSegmentId firstLockId = nextSegmentId; NormSegmentId lastLockId = ndata - 1; - if (NormRepairRequest::ERASURES != requestForm) - lastLockId = MIN(lastLockId, lastSegmentId); + lastLockId = MIN(lastLockId, lastSegmentId); if (holdoff) { if (nextObjectId == txObjectIndex) @@ -4185,10 +4196,7 @@ void NormSession::SenderHandleNackMessage(const struct timeval ¤tTime, Nor // With a series of SEGMENT repair requests for a block, "numErasures" will // eventually total the number of missing segments in the block. - if (NormRepairRequest::ERASURES == requestForm) - numErasures = numErasuresVal; - else - numErasures += (lastSegmentId - nextSegmentId + 1); + numErasures += (lastSegmentId - nextSegmentId + 1); if (holdoff) { if (nextObjectId > txObjectIndex) @@ -4271,7 +4279,6 @@ void NormSession::SenderHandleNackMessage(const struct timeval ¤tTime, Nor numErasures); startTimer = true; } // end if/else (holdoff) - } break; case INFO: // We already dealt with INFO request above with respect to initiating repair From ec518255680cdb1db91234949d558a76515f5754 Mon Sep 17 00:00:00 2001 From: Joseph Calderon Date: Tue, 11 Aug 2026 19:06:03 -0700 Subject: [PATCH 5/9] add rateless FEC test with cmake/waf/ci wiring --- .github/workflows/cmake.yml | 15 ++++ CMakeLists.txt | 31 +++++++++ examples/normRatelessTest.cpp | 126 ++++++++++++++++++++++++++-------- makefiles/Makefile.common | 3 +- wscript | 1 + 5 files changed, 147 insertions(+), 29 deletions(-) diff --git a/.github/workflows/cmake.yml b/.github/workflows/cmake.yml index 7d205f66..fa1b1fbe 100755 --- a/.github/workflows/cmake.yml +++ b/.github/workflows/cmake.yml @@ -181,3 +181,18 @@ jobs: message(FATAL_ERROR "Bad exit status") endif() execute_process(COMMAND ccache -s) + - name: Test + shell: cmake -P {0} + run: | + # NOTE: run from inside "build" rather than with "--test-dir", which the + # pinned CMake ($ENV{CMAKE_VERSION}) predates (it arrived in 3.20). + execute_process( + COMMAND ${{ steps.cmake_and_ninja.outputs.cmake_dir }}/ctest + --output-on-failure + -C $ENV{BUILD_TYPE} + WORKING_DIRECTORY build + RESULT_VARIABLE result + ) + if (NOT result EQUAL 0) + message(FATAL_ERROR "Tests failed") + endif() diff --git a/CMakeLists.txt b/CMakeLists.txt index c0126ad1..60650f12 100755 --- a/CMakeLists.txt +++ b/CMakeLists.txt @@ -12,6 +12,18 @@ 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") +# Self-checking tests build by default only when NORM is the top-level project, so +# projects that add_subdirectory() NORM don't grow an extra binary unasked. +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 of the self-checking tests in /examples." ${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) @@ -129,6 +141,12 @@ if(BUILD_SHARED_LIBS) set_target_properties(norm PROPERTIES SOVERSION 1.5.10) if(WIN32) target_compile_definitions(norm PUBLIC NORM_USE_DLL) + # NORM_USE_DLL alone makes NORM_API_LINKAGE "dllimport" everywhere, which is + # what consumers want but is wrong inside the DLL itself: the classes marked + # with it (NormEncoder/NormDecoder) define members in normEncoder.cpp, and + # MSVC rejects defining a member of a dllimport class. normApi.cpp defines + # this itself for the non-CMake builds; the whole library needs it. + target_compile_definitions(norm PRIVATE _NORM_API_BUILD) endif() endif() @@ -222,5 +240,18 @@ if(NORM_BUILD_EXAMPLES) add_executable(${example} examples/${example}.cpp examples/normSocket.cpp) target_link_libraries(${example} PRIVATE norm protokit::protokit) endforeach() + endif() +if(NORM_BUILD_TESTING) + add_executable(normRatelessTest examples/normRatelessTest.cpp) + target_link_libraries(normRatelessTest PRIVATE norm protokit::protokit) + add_test(NAME normRatelessTest COMMAND normRatelessTest) + add_test(NAME normCustomFecTest COMMAND normRatelessTest) + set_tests_properties(normCustomFecTest PROPERTIES ENVIRONMENT "NORM_FEC_ID=132") + add_test(NAME normRatelessExhaustionTest COMMAND normRatelessTest) + set_tests_properties(normRatelessExhaustionTest PROPERTIES ENVIRONMENT + "NORM_FORCE_DECODE_FAILURE=1;NORM_NDATA=4;NORM_NPARITY=4;NORM_DATA_LEN=2000;NORM_RX_LOSS=50") + set_tests_properties(normRatelessTest normCustomFecTest normRatelessExhaustionTest + PROPERTIES RUN_SERIAL TRUE TIMEOUT 75) +endif() diff --git a/examples/normRatelessTest.cpp b/examples/normRatelessTest.cpp index 8921c225..1e5c557a 100644 --- a/examples/normRatelessTest.cpp +++ b/examples/normRatelessTest.cpp @@ -5,12 +5,65 @@ #include #include #include -#include -#include + +// The NORM descriptor is a socket on unix but a HANDLE on win32, so poll for events +// instead of select()ing on it -- that keeps this test buildable on every platform. +#ifdef WIN32 +#include +static void SleepMsec(unsigned int msec) { Sleep(msec); } +#else +#include +static void SleepMsec(unsigned int msec) { usleep(msec * 1000); } +#endif // if/else WIN32 + +// Read an unsigned tunable from the environment, ignoring unset/empty/unparsable +// values so a mistyped override can't silently produce a degenerate run. +static unsigned long EnvULong(const char* name, unsigned long defaultValue) +{ + const char* text = getenv(name); + if ((NULL == text) || ('\0' == text[0])) return defaultValue; + char* end = NULL; + unsigned long value = strtoul(text, &end, 0); + if ((NULL == end) || ('\0' != *end) || (0 == value)) + { + fprintf(stderr, "normRatelessTest warning: ignoring bad %s=\"%s\"\n", name, text); + return defaultValue; + } + return value; +} // FEC ID for the (partially-specified) rateless code family. static const UINT8 RATELESS_FEC_ID = 131; +static void PackCustomPayloadId(UINT32* buffer, UINT32 blockId, UINT16 symbolId, UINT16) +{ + UINT16* payloadId = (UINT16*)buffer; + payloadId[0] = htons((UINT16)blockId); + payloadId[1] = htons(symbolId); +} + +static UINT32 UnpackCustomBlockId(const UINT32* buffer) +{ + return ntohs(((const UINT16*)buffer)[0]); +} + +static UINT16 UnpackCustomSymbolId(const UINT32* buffer) +{ + return ntohs(((const UINT16*)buffer)[1]); +} + +static UINT16 UnpackCustomBlockLength(const UINT32*) { return 0; } + +// Deliberately registers a stack object; the library must copy it. +static bool RegisterCustomLayout(NormInstanceHandle instance, UINT8 fecId) +{ + NormFecLayout layout = { + 4, 0x0000ffff, PackCustomPayloadId, UnpackCustomBlockId, + UnpackCustomSymbolId, UnpackCustomBlockLength + }; + return NormRegisterFecLayout(instance, fecId, &layout); +} + // 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) @@ -98,6 +151,14 @@ class MockRatelessDecoder : public NormDecoder for (unsigned int i = 0; i < numData; i++) if (erased[i]) sourceErasures++; + // Exercise NORM's explicit-source fallback after every advertised + // repair symbol has been consumed. + if (getenv("NORM_FORCE_DECODE_FAILURE") != NULL) + { + delete[] erased; + return (int)sourceErasures; + } + BuildRatelessPermutation(perm, numData); unsigned int recovered = 0; @@ -136,7 +197,23 @@ int main(int /*argc*/, char* /*argv*/[]) if (NORM_INSTANCE_INVALID == instance) return -1; if (getenv("NORM_DEBUG") != NULL) NormSetDebugLevel((unsigned int)atoi(getenv("NORM_DEBUG"))); - srand((unsigned int)time(NULL)); + // Keep simulated loss/backoff choices repeatable in automated runs. + srand(0x4e4f524dU); + + UINT8 fecId = (UINT8)EnvULong("NORM_FEC_ID", RATELESS_FEC_ID); + if ((RATELESS_FEC_ID != fecId) && !RegisterCustomLayout(instance, fecId)) + { + fprintf(stderr, "normRatelessTest error: failed to register custom FEC layout\n"); + return -1; + } + + // Register the mock rateless codec with the instance. The registry is shared by + // every session the instance owns, so this must happen before any are created. + if (!NormRegisterFecCoder(instance, fecId, CreateMockEncoder, CreateMockDecoder, true)) + { + fprintf(stderr, "normRatelessTest error: failed to register rateless FEC codec\n"); + return -1; + } // 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 @@ -149,14 +226,6 @@ int main(int /*argc*/, char* /*argv*/[]) 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(txSession, true); NormSetRxPortReuse(rxSession, true); NormSetMulticastLoopback(txSession, true); // let our data reach the rx session on this host @@ -173,15 +242,23 @@ int main(int /*argc*/, char* /*argv*/[]) return -1; } - // 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)) + // Shrinking the sender buffer (relative to the object) forces blocks out of the + // sender's block_buffer before their repairs finish, so repair requests have to be + // served from SenderRecoverBlock()-recovered blocks. That is a distinct code path + // from steady-state repair, so make it reachable: + // NORM_TX_BUFFER=131072 NORM_DATA_LEN=200000 ./normRatelessTest + UINT32 txBuffer = (UINT32)EnvULong("NORM_TX_BUFFER", 1024 * 1024); + + UINT16 numData = (UINT16)EnvULong("NORM_NDATA", 16); + UINT16 numParity = (UINT16)EnvULong("NORM_NPARITY", 64); + if (!NormStartSender(txSession, 1, txBuffer, 512, numData, numParity, fecId)) { 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; + const UINT32 dataLen = (UINT32)EnvULong("NORM_DATA_LEN", 20000); char* txData = new char[dataLen]; for (UINT32 i = 0; i < dataLen; i++) txData[i] = (char)((i * 31 + 7) & 0xff); @@ -192,28 +269,20 @@ int main(int /*argc*/, char* /*argv*/[]) fprintf(stderr, "normRatelessTest error: NormDataEnqueue() failed\n"); return -1; } - printf("normRatelessTest: sending %u bytes via mock rateless codec (10%% loss)...\n", dataLen); + printf("normRatelessTest: sending %u bytes via mock rateless codec (%.3g%% rx loss)...\n", + dataLen, lossPct); 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; + // Poll non-blocking so the loop can enforce a wall-clock watchdog (must never hang). + time_t deadline = time(NULL) + 60; 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; + bool gotEvent = false; while (running && NormGetNextEvent(instance, &event, false)) { + gotEvent = true; switch (event.type) { case NORM_RX_OBJECT_COMPLETED: @@ -241,6 +310,7 @@ int main(int /*argc*/, char* /*argv*/[]) break; } } + if (!gotEvent) SleepMsec(10); } if (running) fprintf(stderr, "normRatelessTest error: timed out waiting for object reception\n"); diff --git a/makefiles/Makefile.common b/makefiles/Makefile.common index c747a78d..49481fe1 100755 --- a/makefiles/Makefile.common +++ b/makefiles/Makefile.common @@ -234,7 +234,7 @@ normClient: $(CLIENT_OBJ) libnorm.a $(LIBPROTO) $(CC) $(CFLAGS) -o $@ $(CLIENT_OBJ) $(LDFLAGS) libnorm.a $(LIBPROTO) $(LIBS) mkdir -p ../bin cp $@ ../bin/$@ - + # (normRatelessTest) rateless fec tester RATELESS_SRC = $(EXAMPLE)/normRatelessTest.cpp RATELESS_OBJ = $(RATELESS_SRC:.cpp=.o) @@ -317,3 +317,4 @@ distclean: clean # DO NOT DELETE THIS LINE -- mkdep uses it. # DO NOT PUT ANYTHING AFTER THIS LINE, IT WILL GO AWAY. + diff --git a/wscript b/wscript index 2e21b07c..613aed79 100755 --- a/wscript +++ b/wscript @@ -232,6 +232,7 @@ def build(ctx): 'chant', 'normClient', 'normServer', + 'normRatelessTest', #'wintest' # Windows only (can uncomment on Windows) ): _make_simple_example(ctx, example) From ce0330cff47a29250e124760f1505dda8bc8c99e Mon Sep 17 00:00:00 2001 From: joseph calderon Date: Mon, 17 Aug 2026 09:32:01 -0700 Subject: [PATCH 6/9] mingw issues --- CMakeLists.txt | 11 +++++++++++ 1 file changed, 11 insertions(+) diff --git a/CMakeLists.txt b/CMakeLists.txt index 60650f12..32321f41 100755 --- a/CMakeLists.txt +++ b/CMakeLists.txt @@ -73,9 +73,20 @@ if(NORM_PROTOKIT_GIT_TAG) GIT_TAG ${NORM_PROTOKIT_GIT_TAG} ) FetchContent_MakeAvailable(protokit) + FetchContent_GetProperties(protokit SOURCE_DIR PROTOKIT_SRC_DIR) else() message(STATUS "Building protokit from ${NORM_CUSTOM_PROTOLIB_VERSION}") add_subdirectory(protolib) + set(PROTOKIT_SRC_DIR ${CMAKE_CURRENT_SOURCE_DIR}/protolib) +endif() + +if (MINGW AND TARGET protokit) + target_sources(protokit PRIVATE + ${PROTOKIT_SRC_DIR}/src/win32/win32Net.cpp + ${PROTOKIT_SRC_DIR}/src/win32/win32RouteMgr.cpp + ) + target_link_libraries(protokit PUBLIC ws2_32 iphlpapi user32 gdi32 Advapi32 ntdll) + target_compile_definitions(protokit PUBLIC _CRT_SECURE_NO_WARNINGS) endif() From a40cb831a1f72c127e2e684ef7076ffaa53c3056 Mon Sep 17 00:00:00 2001 From: joseph calderon Date: Mon, 17 Aug 2026 09:45:11 -0700 Subject: [PATCH 7/9] use explicit unicast for rateless loopback test --- examples/normRatelessTest.cpp | 12 +++++++----- 1 file changed, 7 insertions(+), 5 deletions(-) diff --git a/examples/normRatelessTest.cpp b/examples/normRatelessTest.cpp index 1e5c557a..11fbdfe1 100644 --- a/examples/normRatelessTest.cpp +++ b/examples/normRatelessTest.cpp @@ -219,17 +219,19 @@ int main(int /*argc*/, char* /*argv*/[]) // 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 char* groupAddr = "127.0.0.1"; 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; - 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 + // Use explicit unicast configuration to avoid SO_REUSEPORT load-balancing on modern OSes + // when using the same loopback port. + NormSetTxOnly(txSession, true); + NormSetTxPort(txSession, 6004, true); + NormSetDefaultUnicastNack(rxSession, true); + NormSetTxRate(txSession, 10.0e+06); // 10 Mbps for a quick loopback run // Induce loss on the receiver so it must repair via rateless parity. From f6b96d97a5df4f5e1db64af10755d6d9d3797300 Mon Sep 17 00:00:00 2001 From: joseph calderon Date: Wed, 19 Aug 2026 07:19:44 -0700 Subject: [PATCH 8/9] remove c++11 atomics and mutex in favor of thread suspension --- include/normApi.h | 8 ++++++++ include/normMessage.h | 24 ++++++++++++++---------- makefiles/Makefile.common | 2 +- src/common/normApi.cpp | 22 ++++++++++++++++------ src/common/normMessage.cpp | 29 ++++++++++++++++++++++------- 5 files changed, 61 insertions(+), 24 deletions(-) diff --git a/include/normApi.h b/include/normApi.h index ab38a8b1..8a676def 100755 --- a/include/normApi.h +++ b/include/normApi.h @@ -883,6 +883,14 @@ bool NormRegisterFecCoder(NormInstanceHandle instanceHandle, // fails. Re-registering an equivalent layout succeeds, so separate NORM instances in // one process may each register the codecs they use. The layout is copied into // library-owned storage and remains registered for process lifetime. +// +// Call this before creating any session on the instance: registration is refused +// once the instance has sessions, since a layout describes a wire format those +// sessions may already be parsing with. One case is deliberately not covered -- +// two instances in one process publishing *different* layouts for the same fecId +// at the same moment. Closing it needs a process-wide lock, and nothing under +// src/common or include may pull in the STL, so registration is a startup-time +// operation by contract rather than by enforcement. NORM_API_LINKAGE bool NormRegisterFecLayout(NormInstanceHandle instance, UINT8 fecId, const NormFecLayout* layout); diff --git a/include/normMessage.h b/include/normMessage.h index 6c4237e7..56bd58bb 100755 --- a/include/normMessage.h +++ b/include/normMessage.h @@ -9,7 +9,6 @@ #include // for memcpy(), etc #include #include // for rand(), etc -#include #ifdef _WIN32_WCE #include @@ -421,11 +420,11 @@ class NormPayloadId // Published layouts are copied into library-owned storage and remain valid for // process lifetime. A NULL layout succeeds only when the id is unclaimed. static bool SetFecLayout(UINT8 fecId, const NormFecLayout* layout); - static const NormFecLayout* GetFecLayout(UINT8 fecId) {return fec_layouts[fecId].load();} + static const NormFecLayout* GetFecLayout(UINT8 fecId) {return fec_layouts[fecId];} static bool IsValid(UINT8 fecId) { - const NormFecLayout* layout = fec_layouts[fecId].load(); + const NormFecLayout* layout = GetFecLayout(fecId); if (layout != NULL) return true; switch (fecId) @@ -446,7 +445,7 @@ class NormPayloadId static UINT16 GetLength(UINT8 fecId) { - const NormFecLayout* layout = fec_layouts[fecId].load(); + const NormFecLayout* layout = GetFecLayout(fecId); if (layout != NULL) return layout->payloadIdLength; switch (fecId) @@ -464,7 +463,7 @@ class NormPayloadId static UINT32 GetFecBlockMask(UINT8 fecId, UINT8 fecM) { - const NormFecLayout* layout = fec_layouts[fecId].load(); + const NormFecLayout* layout = GetFecLayout(fecId); if (layout != NULL) return layout->blockMask; switch (fecId) @@ -487,7 +486,7 @@ class NormPayloadId void SetFecPayloadId(UINT32 blockId, UINT16 symbolId, UINT16 blockLen) { - const NormFecLayout* layout = fec_layouts[fec_id].load(); + const NormFecLayout* layout = GetFecLayout(fec_id); if (layout != NULL && layout->packPayloadId != NULL) { layout->packPayloadId(buffer, blockId, symbolId, blockLen); @@ -533,7 +532,7 @@ class NormPayloadId // Message processing methods NormBlockId GetFecBlockId() const { - const NormFecLayout* layout = fec_layouts[fec_id].load(); + const NormFecLayout* layout = GetFecLayout(fec_id); if (layout != NULL && layout->unpackBlockId != NULL) { return layout->unpackBlockId(cbuffer); @@ -571,7 +570,7 @@ class NormPayloadId UINT16 GetFecSymbolId() const { - const NormFecLayout* layout = fec_layouts[fec_id].load(); + const NormFecLayout* layout = GetFecLayout(fec_id); if (layout != NULL && layout->unpackSymbolId != NULL) { return layout->unpackSymbolId(cbuffer); @@ -612,7 +611,7 @@ class NormPayloadId UINT16 GetFecBlockLength() const { - const NormFecLayout* layout = fec_layouts[fec_id].load(); + const NormFecLayout* layout = GetFecLayout(fec_id); if (layout != NULL && layout->unpackBlockLength != NULL) { return layout->unpackBlockLength(cbuffer); @@ -631,7 +630,12 @@ class NormPayloadId private: // Registered via NormRegisterFecLayout(); read on the rx thread, so atomic. - static std::atomic fec_layouts[256]; + // volatile, not atomic: a reader must not have its load hoisted or cached + // across calls, and a pointer-width aligned load or store is indivisible on + // every supported target. is deliberately not used -- nothing under + // src/common or include pulls in the STL, which is what lets the static + // library link with no C++ runtime. + static const NormFecLayout* volatile fec_layouts[256]; UINT8 fec_id; UINT8 fec_m; diff --git a/makefiles/Makefile.common b/makefiles/Makefile.common index 49481fe1..6f1753b0 100755 --- a/makefiles/Makefile.common +++ b/makefiles/Makefile.common @@ -38,7 +38,7 @@ NORM_SRC = $(COMMON)/normMessage.cpp $(COMMON)/normSession.cpp \ $(COMMON)/normSegment.cpp $(COMMON)/normEncoder.cpp \ $(COMMON)/normEncoderRS8.cpp $(COMMON)/normEncoderRS16.cpp \ $(COMMON)/normEncoderMDP.cpp $(COMMON)/galois.cpp \ - $(COMMON)/normFile.cpp $(COMMON)/normApi.cpp $(SYSTEM_SRC) + $(COMMON)/normFile.cpp $(COMMON)/normApi.cpp $(PROTOLIB)/src/common/protoThread.cpp $(SYSTEM_SRC) NORM_OBJ = $(NORM_SRC:.cpp=.o) diff --git a/src/common/normApi.cpp b/src/common/normApi.cpp index 67127fa5..293b3610 100755 --- a/src/common/normApi.cpp +++ b/src/common/normApi.cpp @@ -1710,12 +1710,22 @@ bool NormRegisterFecCoder(NormInstanceHandle instanceHandle, NORM_API_LINKAGE bool NormRegisterFecLayout(NormInstanceHandle instanceHandle, UINT8 fecId, const NormFecLayout* layout) { - // Validate before touching the process-wide table, so a bad handle can't leave a - // registration behind. - if (NULL == (NormInstance*)instanceHandle) return false; - // Layouts are keyed by fecId process-wide: the NormPayloadId call sites that need - // them (FTI/NACK parsing, block mask setup) have no instance context to consult. - return NormPayloadId::SetFecLayout(fecId, layout); + NormInstance* instance = (NormInstance*)instanceHandle; + if (instance && instance->dispatcher.SuspendThread()) + { + // Same rule as NormRegisterFecCoder(): a layout describes the payload id wire + // format its codec depends on, so refuse to publish one underneath sessions + // that may already be parsing with it. + if (instance->session_mgr.HasSessions()) + { + instance->dispatcher.ResumeThread(); + return false; + } + bool result = NormPayloadId::SetFecLayout(fecId, layout); + instance->dispatcher.ResumeThread(); + return result; + } + return false; } // end NormRegisterFecLayout() NORM_API_LINKAGE diff --git a/src/common/normMessage.cpp b/src/common/normMessage.cpp index abede052..4520c372 100755 --- a/src/common/normMessage.cpp +++ b/src/common/normMessage.cpp @@ -1,9 +1,22 @@ #include "normMessage.h" -#include -std::atomic NormPayloadId::fec_layouts[256] = {}; +const NormFecLayout* volatile NormPayloadId::fec_layouts[256] = {}; static NormFecLayout norm_fec_layout_storage[256] = {}; -static std::mutex norm_fec_layout_mutex; + +namespace { +// Publish barrier. Thread suspension orders this writer against *this instance's* +// protocol thread, but the layout table is process-wide, so another instance's +// thread can be reading fec_layouts[] while this one writes. The fence keeps the +// pointer store from being seen before the layout it points at. +inline void NormFecLayoutPublishBarrier() +{ +#ifdef WIN32 + MemoryBarrier(); +#else + __sync_synchronize(); +#endif // if/else WIN32/UNIX +} +} // namespace // Do two layouts describe the same payload id wire format? Compared field-wise // rather than by memcmp() so struct padding can't make equivalent layouts differ. @@ -24,7 +37,7 @@ bool NormPayloadId::SetFecLayout(UINT8 fecId, const NormFecLayout* layout) // Published layouts may be in use by any NORM instance in this process. // Keep them immutable for process lifetime so readers and callback targets // cannot be invalidated underneath packet processing. - return (NULL == fec_layouts[fecId].load()); + return (NULL == GetFecLayout(fecId)); } if ((0 == layout->payloadIdLength) || (0 != (layout->payloadIdLength & 0x03)) || (0 == layout->blockMask) || (NULL == layout->packPayloadId) || @@ -34,16 +47,18 @@ bool NormPayloadId::SetFecLayout(UINT8 fecId, const NormFecLayout* layout) return false; } - std::lock_guard lock(norm_fec_layout_mutex); - const NormFecLayout* current = fec_layouts[fecId].load(); + const NormFecLayout* current = GetFecLayout(fecId); if (NULL == current) { norm_fec_layout_storage[fecId] = *layout; - fec_layouts[fecId].store(&norm_fec_layout_storage[fecId]); + NormFecLayoutPublishBarrier(); + fec_layouts[fecId] = &norm_fec_layout_storage[fecId]; return true; } if (NormFecLayoutsEquivalent(current, layout)) + { return true; // same wire format, so nothing to disagree about + } PLOG(PL_ERROR, "NormPayloadId::SetFecLayout() error: fec_id %u is already registered " "with a different payload id layout\n", fecId); return false; From e5b89343a5eef9bfb1edc21201b9ef4522db69f7 Mon Sep 17 00:00:00 2001 From: Joseph Calderon Date: Wed, 19 Aug 2026 08:35:01 -0700 Subject: [PATCH 9/9] extend rateless mock tests --- CMakeLists.txt | 7 +++++ examples/normRatelessTest.cpp | 49 +++++++++++++++++++++++++++++++++++ 2 files changed, 56 insertions(+) diff --git a/CMakeLists.txt b/CMakeLists.txt index 32321f41..8a439097 100755 --- a/CMakeLists.txt +++ b/CMakeLists.txt @@ -263,6 +263,13 @@ if(NORM_BUILD_TESTING) add_test(NAME normRatelessExhaustionTest COMMAND normRatelessTest) set_tests_properties(normRatelessExhaustionTest PROPERTIES ENVIRONMENT "NORM_FORCE_DECODE_FAILURE=1;NORM_NDATA=4;NORM_NPARITY=4;NORM_DATA_LEN=2000;NORM_RX_LOSS=50") + add_test(NAME normRatelessReNackTest COMMAND normRatelessTest) + set_tests_properties(normRatelessReNackTest PROPERTIES ENVIRONMENT + "NORM_DECODE_SHORT_COUNT=1;NORM_NDATA=4;NORM_NPARITY=8;NORM_DATA_LEN=2000;NORM_RX_LOSS=30") + add_test(NAME normRatelessReNackMultiTest COMMAND normRatelessTest) + set_tests_properties(normRatelessReNackMultiTest PROPERTIES ENVIRONMENT + "NORM_DECODE_SHORT_COUNT=8;NORM_NDATA=8;NORM_NPARITY=32;NORM_DATA_LEN=20000;NORM_RX_LOSS=40") set_tests_properties(normRatelessTest normCustomFecTest normRatelessExhaustionTest + normRatelessReNackTest normRatelessReNackMultiTest PROPERTIES RUN_SERIAL TRUE TIMEOUT 75) endif() diff --git a/examples/normRatelessTest.cpp b/examples/normRatelessTest.cpp index 11fbdfe1..60d01987 100644 --- a/examples/normRatelessTest.cpp +++ b/examples/normRatelessTest.cpp @@ -78,6 +78,17 @@ static void BuildRatelessPermutation(unsigned int* perm, unsigned int n) } } +// Decoder instrumentation. NORM is supposed to honor a *partial* rateless decode +// by keeping the block pending and NACKing for more repair symbols, then decoding +// again once they arrive. NORM_DECODE_SHORT_COUNT forces the first N decode +// attempts to come up short (recovering nothing) so that recovery path gets +// exercised -- distinct from NORM_FORCE_DECODE_FAILURE, which never recovers and +// therefore only reaches the explicit-source-repair fallback. +static unsigned long mock_short_budget = 0; // forced short decodes remaining +static unsigned long mock_decode_calls = 0; // Decode() invocations +static unsigned long mock_short_returns = 0; // returns saying "still missing" +static unsigned long mock_parity_recoveries = 0; // decodes that fully repaired via parity + class MockRatelessEncoder : public NormEncoder { public: @@ -151,6 +162,8 @@ class MockRatelessDecoder : public NormDecoder for (unsigned int i = 0; i < numData; i++) if (erased[i]) sourceErasures++; + mock_decode_calls++; + // Exercise NORM's explicit-source fallback after every advertised // repair symbol has been consumed. if (getenv("NORM_FORCE_DECODE_FAILURE") != NULL) @@ -159,6 +172,16 @@ class MockRatelessDecoder : public NormDecoder return (int)sourceErasures; } + // Come up short on purpose: recover nothing and report the erasures as + // still missing, so NORM must request more symbols and decode again. + if ((mock_short_budget > 0) && (sourceErasures > 0)) + { + mock_short_budget--; + mock_short_returns++; + delete[] erased; + return (int)sourceErasures; + } + BuildRatelessPermutation(perm, numData); unsigned int recovered = 0; @@ -175,6 +198,7 @@ class MockRatelessDecoder : public NormDecoder } } unsigned int stillMissing = sourceErasures - recovered; + if ((recovered > 0) && (0 == stillMissing)) mock_parity_recoveries++; delete[] erased; return (int)stillMissing; } @@ -251,6 +275,8 @@ int main(int /*argc*/, char* /*argv*/[]) // NORM_TX_BUFFER=131072 NORM_DATA_LEN=200000 ./normRatelessTest UINT32 txBuffer = (UINT32)EnvULong("NORM_TX_BUFFER", 1024 * 1024); + mock_short_budget = EnvULong("NORM_DECODE_SHORT_COUNT", 0); + UINT16 numData = (UINT16)EnvULong("NORM_NDATA", 16); UINT16 numParity = (UINT16)EnvULong("NORM_NPARITY", 64); if (!NormStartSender(txSession, 1, txBuffer, 512, numData, numParity, fecId)) @@ -317,6 +343,29 @@ int main(int /*argc*/, char* /*argv*/[]) if (running) fprintf(stderr, "normRatelessTest error: timed out waiting for object reception\n"); + printf("normRatelessTest: decode attempts %lu, short returns %lu, parity recoveries %lu\n", + mock_decode_calls, mock_short_returns, mock_parity_recoveries); + + // When short decodes were requested, completing the transfer is not enough: NORM + // could have limped home on explicit source repair alone. Require evidence that + // it honored the partial decode (a short return actually occurred) and that a + // later decode of the same block then succeeded via parity. + if ((0 == result) && (0 != EnvULong("NORM_DECODE_SHORT_COUNT", 0))) + { + if (0 == mock_short_returns) + { + fprintf(stderr, "normRatelessTest: no short decode ever occurred " + "(loss too low to force a decode attempt?). FAIL\n"); + result = -1; + } + else if (0 == mock_parity_recoveries) + { + fprintf(stderr, "normRatelessTest: block never re-decoded after the short " + "return; repaired by explicit source symbols only. FAIL\n"); + result = -1; + } + } + NormStopSender(txSession); NormStopReceiver(rxSession); NormDestroyInstance(instance);