BodgeMaster
/
NORM-mirror
mirror of https://github.com/USNavalResearchLaboratory/norm.git
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

basic rateless codec support

pull/102/head
joseph calderon 2026-06-13 22:45:23 -07:00 committed by Joseph Calderon
parent 1686404314
commit fe495011fa
12 changed files with 545 additions and 92 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

88
examples/normRatelessTest.cpp Normal file
View File

@ -0,0 +1,88 @@
#include <normApi.h>
#include <normEncoder.h>
#include <stdio.h>
#include <string.h>
#include <stdlib.h>
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;
}

25
include/normApi.h
View File

@ -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 */

15
include/normEncoder.h
View File

@ -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

119
include/normMessage.h
View File

@ -3,6 +3,7 @@
// PROTOLIB includes
#include "protokit.h"
#include "normApi.h"
// standard includes
#include <string.h> // 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);
@ -555,7 +606,6 @@ class NormPayloadId
}
private:
UINT8 fec_id;
UINT8 fec_m;
@ -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
{

6
include/normSegment.h
View File

@ -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)

37
include/normSession.h
View File

@ -4,9 +4,9 @@
#include "normMessage.h"
#include "normObject.h"
#include "normNode.h"
#include "normEncoder.h"
#include "protokit.h"
#include "normEncoder.h"
#include <protoTree.h>
#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,11 +128,14 @@ 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};
static const UINT8 DEFAULT_TTL;
@ -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;

10
makefiles/Makefile.common
View File

@ -235,6 +235,15 @@ normClient: $(CLIENT_OBJ) libnorm.a $(LIBPROTO)
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.

30
src/common/normApi.cpp
View File

@ -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)

37
src/common/normNode.cpp
View File

@ -286,6 +286,18 @@ bool NormSenderNode::AllocateBuffers(unsigned int bufferSpace,
if (NULL != decoder) delete decoder;
if (0 != numParity)
{
NormDecoderFactory decoderFactory = session.GetDecoderFactory(fecId);
if (NULL != decoderFactory)
{
if (NULL == (decoder = decoderFactory()))
{
PLOG(PL_FATAL, "NormSenderNode::AllocateBuffers() new custom decoder error\n");
Close();
return false;
}
}
else
{
switch (fecId)
{
@ -324,14 +336,14 @@ bool NormSenderNode::AllocateBuffers(unsigned int bufferSpace,
}
break;
case 129:
#ifdef ASSUME_MDP_FEC
#ifdef ASSUME_MDP_FEC
if (NULL == (decoder = new NormDecoderMDP))
{
PLOG(PL_FATAL, "NormSenderNode::AllocateBuffers() new NormDecoderMDP error: %s\n", GetErrorString());
Close();
return false;
}
#else
#else
if (0 == fecInstanceId)
{
if (NULL == (decoder = new NormDecoderRS8))
@ -347,13 +359,14 @@ bool NormSenderNode::AllocateBuffers(unsigned int bufferSpace,
Close();
return false;
}
#endif // if/else ASSUME_MDP_FEC
#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()))
{
PLOG(PL_FATAL, "NormSenderNode::AllocateBuffers() decoder init error\n");
@ -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);

57
src/common/normObject.cpp
View File

@ -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)

30
src/common/normSegment.cpp
View File

@ -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;

57
src/common/normSession.cpp
View File

@ -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 &currentTime, 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,11 +4140,12 @@ void NormSession::SenderHandleNackMessage(const struct timeval &currentTime, Nor
{
// mark nack time for potential flow control
static_cast<NormStreamObject *>(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;
if (NormRepairRequest::ERASURES != requestForm)
lastLockId = MIN(lastLockId, lastSegmentId);
if (holdoff)
{
@ -4164,6 +4207,9 @@ void NormSession::SenderHandleNackMessage(const struct timeval &currentTime, 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);
if (holdoff)
{
@ -4238,11 +4284,16 @@ void NormSession::SenderHandleNackMessage(const struct timeval &currentTime, 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