From fe495011fa86188087df147ba26fcd418dad5731 Mon Sep 17 00:00:00 2001 From: joseph calderon Date: Sat, 13 Jun 2026 22:45:23 -0700 Subject: [PATCH] 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 | 159 ++++++++++++++++++++-------------- src/common/normObject.cpp | 57 ++++++++++-- src/common/normSegment.cpp | 30 ++++++- src/common/normSession.cpp | 61 +++++++++++-- 12 files changed, 545 insertions(+), 92 deletions(-) create mode 100644 examples/normRatelessTest.cpp diff --git a/examples/normRatelessTest.cpp b/examples/normRatelessTest.cpp new file mode 100644 index 0000000..ba4b8a8 --- /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 0caa5c3..e29b8b2 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 3c87875..4216198 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 954c5e0..ddc8e95 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 cf88173..cb68527 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 38971c6..5ee2d14 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 fb18d78..c747a78 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 a151e94..2fc8146 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 63619f0..34e8c56 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 = 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; - } - 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() new NormDecoderRS8 error: %s\n", GetErrorString()); - Close(); - return false; - } - } - else - { - PLOG(PL_FATAL, "NormSenderNode::AllocateBuffers() error: unknown fecId=129 instanceId!\n"); - Close(); - return false; - } -#endif // if/else ASSUME_MDP_FEC - break; - default: - PLOG(PL_FATAL, "NormSenderNode::AllocateBuffers() error: unknown fecId>%d!\n", fecId); + if (NULL == (decoder = decoderFactory())) + { + PLOG(PL_FATAL, "NormSenderNode::AllocateBuffers() new custom decoder error\n"); Close(); - return false; + 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; + } + 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() new NormDecoderRS8 error: %s\n", GetErrorString()); + Close(); + return false; + } + } + else + { + PLOG(PL_FATAL, "NormSenderNode::AllocateBuffers() error: unknown fecId=129 instanceId!\n"); + 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; + } } 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 fdda3da..eca1717 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 e2a2fb7..d31a2da 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 6c73855..c06edd5 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