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

Merge 2d653628ce into 1686404314

pull/102/merge
joe-calderon-ditto 2026-07-13 03:23:56 +00:00 committed by GitHub
parent 1686404314 2d653628ce
commit 1f6584ffb3
No known key found for this signature in database
GPG Key ID: B5690EEEBB952194
12 changed files with 836 additions and 132 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

255
examples/normRatelessTest.cpp Normal file
View File

@ -0,0 +1,255 @@
#include <normApi.h>
#include <normEncoder.h>
#include <stdio.h>
#include <string.h>
#include <stdlib.h>
#include <time.h>
#include <sys/select.h>
#include <sys/time.h>
// FEC ID for the (partially-specified) rateless code family.
static const UINT8 RATELESS_FEC_ID = 131;
// Deterministic permutation of [0, n) shared by encoder and decoder.
// The seed depends only on "n", so both sides produce the identical ordering.
static void BuildRatelessPermutation(unsigned int* perm, unsigned int n)
{
for (unsigned int i = 0; i < n; i++) perm[i] = i;
unsigned int state = 0x9e3779b9u ^ n;
for (unsigned int i = n; i > 1; i--)
{
state = state * 1103515245u + 12345u;
unsigned int j = (state >> 8) % i;
unsigned int t = perm[i - 1]; perm[i - 1] = perm[j]; perm[j] = t;
}
}
class MockRatelessEncoder : public NormEncoder
{
public:
MockRatelessEncoder() : ndata(0), vec_size(0), perm(NULL) {}
virtual ~MockRatelessEncoder() { Destroy(); }
virtual bool Init(unsigned int numData, unsigned int /*numParity*/, UINT16 vectorSize)
{
Destroy();
ndata = numData;
vec_size = vectorSize;
perm = new unsigned int[ndata ? ndata : 1];
return (NULL != perm);
}
virtual void Destroy() { delete[] perm; perm = NULL; }
// Block-oriented encode is unused by this rateless surrogate; repair
// symbols are synthesized on demand in EncodeParity() below.
virtual void Encode(unsigned int, const char*, char**) {}
// Generate the "parityId"-th repair symbol as a copy of a source symbol
// selected by the shared permutation.
virtual void EncodeParity(unsigned int parityId, const char** sourceVectorList,
unsigned int numData, char* parityVector)
{
if (0 == numData) return;
BuildRatelessPermutation(perm, numData);
unsigned int k = perm[parityId % numData];
if ((k < numData) && (NULL != sourceVectorList[k]))
memcpy(parityVector, sourceVectorList[k], vec_size);
}
virtual bool IsRateless() const { return true; }
private:
unsigned int ndata;
UINT16 vec_size;
unsigned int* perm; // scratch buffer sized to the block size
};
class MockRatelessDecoder : public NormDecoder
{
public:
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;
unsigned int nparity;
UINT16 vec_size;
unsigned int* perm;
};
extern "C" {
NormEncoder* CreateMockEncoder() { return new MockRatelessEncoder(); }
NormDecoder* CreateMockDecoder() { return new MockRatelessDecoder(); }
}
int main(int /*argc*/, char* /*argv*/[])
{
NormInstanceHandle instance = NormCreateInstance();
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;
}
// 16 source symbols/block, up to 64 repair symbols/block, 512-byte segments.
if (!NormStartSender(txSession, 1, 1024 * 1024, 512, 16, 64, RATELESS_FEC_ID))
{
fprintf(stderr, "normRatelessTest error: NormStartSender() failed\n");
return -1;
}
// Build a known payload spanning several blocks (incl. a short final block).
const UINT32 dataLen = 20000;
char* txData = new char[dataLen];
for (UINT32 i = 0; i < dataLen; i++)
txData[i] = (char)((i * 31 + 7) & 0xff);
NormObjectHandle txObj = NormDataEnqueue(txSession, txData, dataLen);
if (NORM_OBJECT_INVALID == txObj)
{
fprintf(stderr, "normRatelessTest error: NormDataEnqueue() failed\n");
return -1;
}
printf("normRatelessTest: sending %u bytes via mock rateless codec (10%% loss)...\n", dataLen);
int result = -1;
// NormGetNextEvent() blocks, so drive it via select() on the NORM descriptor
// to enforce a wall-clock watchdog (the demo must never hang).
NormDescriptor normFd = NormGetDescriptor(instance);
time_t deadline = time(NULL) + 30;
bool running = true;
while (running && (time(NULL) <= deadline))
{
fd_set fdSet;
FD_ZERO(&fdSet);
FD_SET(normFd, &fdSet);
struct timeval timeout;
timeout.tv_sec = 1;
timeout.tv_usec = 0;
int n = select((int)normFd + 1, &fdSet, NULL, NULL, &timeout);
if (n <= 0) continue; // timeout tick (re-check deadline) or interrupted
NormEvent event;
while (running && NormGetNextEvent(instance, &event, false))
{
switch (event.type)
{
case NORM_RX_OBJECT_COMPLETED:
{
unsigned int rxLen = (unsigned int)NormObjectGetSize(event.object);
const char* rxData = NormDataAccessData(event.object);
if ((rxLen == dataLen) && (NULL != rxData) && (0 == memcmp(rxData, txData, dataLen)))
{
printf("normRatelessTest: received %u bytes, contents verified. PASS\n", rxLen);
result = 0;
}
else
{
fprintf(stderr, "normRatelessTest: received object mismatch (len %u vs %u). FAIL\n",
rxLen, dataLen);
}
running = false;
break;
}
case NORM_RX_OBJECT_ABORTED:
fprintf(stderr, "normRatelessTest error: NORM_RX_OBJECT_ABORTED\n");
running = false;
break;
default:
break;
}
}
}
if (running)
fprintf(stderr, "normRatelessTest error: timed out waiting for object reception\n");
NormStopSender(txSession);
NormStopReceiver(rxSession);
NormDestroyInstance(instance);
delete[] txData;
printf("normRatelessTest: %s\n", (0 == result) ? "SUCCESS" : "FAILURE");
return result;
}

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

26
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,15 @@ 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;
// 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; }
}; // end class NormEncoder
class NormDecoder
@ -50,6 +70,12 @@ class NormDecoder
virtual ~NormDecoder();
virtual bool Init(unsigned int numData, unsigned int numParity, UINT16 vectorSize) = 0;
virtual void Destroy() = 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

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
{

12
include/normSegment.h
View File

@ -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
{
@ -238,14 +243,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)
@ -277,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;

47
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,8 +72,12 @@ class NormController
class NormNode* node,
class NormObject* object) = 0;
const NormFecLayout* fec_layouts[256];
}; // end class NormController
typedef class NormEncoder* (*NormEncoderFactory)();
typedef class NormDecoder* (*NormDecoderFactory)();
class NormSessionMgr
{
friend class NormSession;
@ -112,6 +120,12 @@ class NormSessionMgr
NormDataObject::DataFreeFunctionHandle GetDataFreeFunction() const
{return data_free_func;}
// 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;
@ -121,13 +135,17 @@ class NormSessionMgr
class NormSession* top_session; // top of NormSession list
NormEncoderFactory encoder_factories[256];
NormDecoderFactory decoder_factories[256];
bool is_rateless_codec[256];
}; // end class NormSessionMgr
class NormSession
class NormSession : public NormTelemetryContext
{
friend class NormSessionMgr;
public:
enum {DEFAULT_MESSAGE_POOL_DEPTH = 16};
static const UINT8 DEFAULT_TTL;
@ -465,6 +483,14 @@ class NormSession
void SenderSetExtraParity(UINT16 extraParity)
{extra_parity = extraParity;}
// 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; }
INT32 Difference(NormBlockId a, NormBlockId b) const
{return NormBlockId::Difference(a, b, fec_block_mask);}
int Compare(NormBlockId a, NormBlockId b) const
@ -521,8 +547,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, const char** sourceVectorList, unsigned int numData, char* parityVector)
{encoder->EncodeParity(parityId, sourceVectorList, numData, parityVector);}
NormBlock* SenderGetFreeBlock(NormObjectId objectId, NormBlockId blockId);

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->GetSessionMgr().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)

31
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)
{
@ -354,6 +366,7 @@ bool NormSenderNode::AllocateBuffers(unsigned int bufferSpace,
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);

121
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)
@ -1605,9 +1609,18 @@ void NormObject::HandleObjectMessage(const NormObjectMsg& msg,
} // end if (nextErasure < numData)
} // end if (block->GetFirstPending(nextErasure))
bool ratelessDecodeIncomplete = false;
if (erasureCount)
{
sender->Decode(block->SegmentList(), numData, erasureCount);
// The decoder returns the number of source erasures it could NOT
// 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)
{
for (UINT16 i = 0; i < erasureCount; i++)
{
NormSegmentId sid = sender->GetErasureLoc(i);
@ -1634,13 +1647,28 @@ void NormObject::HandleObjectMessage(const NormObjectMsg& msg,
}
}
}
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++)
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);
}
} // if erasureCount <= parityCount (i.e., block complete)
// Notify application of new data available
// (TBD) this could be improved for stream objects
@ -1850,6 +1878,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 +1946,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 +1967,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);
@ -2050,6 +2094,25 @@ bool NormObject::NextSenderMsg(NormObjectMsg* msg)
block->UpdateSegSizeMax(payloadLength);
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
@ -2060,6 +2123,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, (const char**)block->SegmentList(), 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 +2287,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++)
@ -2218,6 +2303,25 @@ 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
{
@ -2238,7 +2342,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)

40
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),
@ -207,8 +208,14 @@ 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
repair_mask.XCopy(pending_mask);
@ -528,8 +535,37 @@ bool NormBlock::AppendRepairRequest(NormNackMsg& nack,
UINT16 numParity,
NormObjectId objectId,
bool pendingInfo,
UINT16 payloadMax)
UINT16 payloadMax,
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;

46
src/common/normSession.cpp
View File

@ -836,7 +836,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 +4047,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 +4118,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 +4185,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 +4262,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
@ -5790,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();