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

v1.3b9

pull/2/head
Jeff Weston 2019-09-11 11:55:26 -04:00
parent 098570e3ad
commit 276f7c4e20
19 changed files with 713 additions and 333 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

13
TODO.TXT
View File

@ -44,7 +44,20 @@
14) IMPORANT: double check ACKing in reponse to watermark polling and 14) IMPORANT: double check ACKing in reponse to watermark polling and
relation to cc feedback suppression, etc ... watermark acks should relation to cc feedback suppression, etc ... watermark acks should
be unicast and not effect cc feedback? be unicast and not effect cc feedback?
15) Implement TX_OBJECT_SENT event (COMPLETED)
16) Set NORM_DATA "NORM_FLAG_REPAIR" properly
17) Set objectId in TX_FLUSH_COMPLETED and TX_WATERMARK_COMPLETED events.
18) Add APIs for managing remote server state kept at receiver
19) Fix "activity_timer" interval setting (COMPLETED)
20) Double-check watermark check code around line 600 of normSession.cpp
(COMPLETED - reimplemented, adding tx_repair_pending index to use
instead of seeking each time)

30
VERSION.TXT
View File

@ -8,6 +8,36 @@ Version 1.3b9
message erroneously dereferenced a NormObject pointer as message erroneously dereferenced a NormObject pointer as
NormFileObject (led to run-time issue in optimized code build) NormFileObject (led to run-time issue in optimized code build)
(thanks to Charlie Davis once again) (thanks to Charlie Davis once again)
- Fixed places where NORM_CMD(FLUSH/SQUELCH) messages didn't have
FEC source block length properly set (thanks to Ron in 't Velt)
- Fixed misuse of "delete" operator where "delete[]" should have
been used instead in "normEncoder.cpp" (thanks to Jon of Totient?)
- Fixed error in NormQuantizeRate() and NormUnQuantizeRate() functions
in "normMessage.h". (Thanks to Julian Onions for getting it right
in the Ethereal NORM dissector he created).
- Fixed case in NormObject::NextServerMsg() where old recovered block
could cause assertion failure (thanks to John Wood for the stress testing)
- Fixed "normApi.cpp" code that was resulting in NormServerNode reference
counts not being updated properly (thanks to Jon Dyte for this)
- Fixed missed initialization of NormServerNode::nominal_packet_count and
NormLossEstimator2::init members (Thanks to Jon Dyte again).
- Fixed bug where NORM_DATA messages containing FEC parity content had
8 bytes of extraneous content appended in their payload (Thanks to
Jon Jannucci for finding this one).
- Fixed bug where receiver loss estimate returning to ZERO could result
in a floating point exception in NormSession::CalculateRate()
(thanks to Ian Downard for finding this one)
- Re-ordered NORM_DATA "payload_length" and "payload_msg_start" fields
to be consistent with revised NORM spec.
- Corrected the detection of "data loss" when application (or API)
fails to consume received stream data before it is discarded.
- Fixed bug where when small "txbuffersize" was set for sender, but
a relatively large FEC block size was kept, the sender app would
not be notified by stream "vacancy"
- Fixed bug with initialization of grtt_estimate when NormSetTransmitRate()
was called _before_ NormStartSender()
- Fixed intermediate issue where flush_timer mis-scheduling caused
sender to "hang" (Thanks to Scott Bates for quickly noticing this)
Version 1.3b8 Version 1.3b8
============= =============

6
common/normApi.cpp
View File

@ -459,7 +459,7 @@ bool NormInstance::GetNextEvent(NormEvent* theEvent)
// "Release" any previously-retained object or node handle // "Release" any previously-retained object or node handle
if (NORM_OBJECT_INVALID != previous_notification->event.object) if (NORM_OBJECT_INVALID != previous_notification->event.object)
((NormObject*)(previous_notification->event.object))->Release(); ((NormObject*)(previous_notification->event.object))->Release();
else if (NORM_NODE_INVALID != previous_notification->event.object) else if (NORM_NODE_INVALID != previous_notification->event.sender)
((NormServerNode*)(previous_notification->event.sender))->Release(); ((NormServerNode*)(previous_notification->event.sender))->Release();
notify_pool.Append(previous_notification); notify_pool.Append(previous_notification);
previous_notification = NULL; previous_notification = NULL;
@ -600,7 +600,7 @@ void NormInstance::Shutdown()
// Release any previously-retained object or node handles // Release any previously-retained object or node handles
if (NORM_OBJECT_INVALID != previous_notification->event.object) if (NORM_OBJECT_INVALID != previous_notification->event.object)
((NormObject*)(previous_notification->event.object))->Release(); ((NormObject*)(previous_notification->event.object))->Release();
else if (NORM_NODE_INVALID != previous_notification->event.object) else if (NORM_NODE_INVALID != previous_notification->event.sender)
((NormServerNode*)(previous_notification->event.sender))->Release(); ((NormServerNode*)(previous_notification->event.sender))->Release();
notify_pool.Append(previous_notification); notify_pool.Append(previous_notification);
previous_notification = NULL; previous_notification = NULL;
@ -629,7 +629,7 @@ void NormInstance::Shutdown()
} }
if (NORM_OBJECT_INVALID != n->event.object) if (NORM_OBJECT_INVALID != n->event.object)
((NormObject*)(n->event.object))->Release(); ((NormObject*)(n->event.object))->Release();
else if (NORM_NODE_INVALID != n->event.object) else if (NORM_NODE_INVALID != n->event.sender)
((NormServerNode*)(n->event.sender))->Release(); ((NormServerNode*)(n->event.sender))->Release();
delete n; delete n;
} }

17
common/normApi.h
View File

@ -501,6 +501,23 @@ bool NormNodeGetAddress(NormNodeHandle nodeHandle,
NORM_API_LINKAGE NORM_API_LINKAGE
double NormNodeGetGrtt(NormNodeHandle nodeHandle); double NormNodeGetGrtt(NormNodeHandle nodeHandle);
// The next 4 functions have not yet been implemented
// (work in progress)
NORM_API_LINKAGE
void NormNodeSetAutoDelete(NormNodeHandle nodeHandle,
bool autoDelete);
NORM_API_LINKAGE
void NormNodeDeleteSender(NormNodeHandle nodeHandle);
NORM_API_LINKAGE
bool NormNodeAllowSender(NormNodeId senderId);
NORM_API_LINKAGE
bool NormNodeDenySender(NormNodeId senderId);
NORM_API_LINKAGE NORM_API_LINKAGE
void NormNodeRetain(NormNodeHandle nodeHandle); void NormNodeRetain(NormNodeHandle nodeHandle);

143
common/normApp.cpp
View File

@ -40,6 +40,7 @@ class NormApp : public NormController, public ProtoApp
private: private:
void ShowHelp(); void ShowHelp();
void OnInputReady(); void OnInputReady();
bool AddAckingNodes(const char* ackingNodeList);
enum CmdType {CMD_INVALID, CMD_NOARG, CMD_ARG}; enum CmdType {CMD_INVALID, CMD_NOARG, CMD_ARG};
CmdType CommandType(const char* cmd); CmdType CommandType(const char* cmd);
@ -111,6 +112,8 @@ class NormApp : public NormController, public ProtoApp
double tx_repeat_interval; double tx_repeat_interval;
bool tx_repeat_clear; bool tx_repeat_clear;
ProtoTimer interval_timer; ProtoTimer interval_timer;
char* acking_node_list; // comma-delimited string
bool watermark_pending;
// NormSession client-only parameters // NormSession client-only parameters
unsigned long rx_buffer_size; // bytes unsigned long rx_buffer_size; // bytes
@ -144,6 +147,7 @@ NormApp::NormApp()
group_size(NormSession::DEFAULT_GSIZE_ESTIMATE), group_size(NormSession::DEFAULT_GSIZE_ESTIMATE),
tx_buffer_size(1024*1024), tx_one_shot(false), tx_buffer_size(1024*1024), tx_one_shot(false),
tx_object_interval(0.0), tx_repeat_count(0), tx_repeat_interval(2.0), tx_repeat_clear(true), tx_object_interval(0.0), tx_repeat_count(0), tx_repeat_interval(2.0), tx_repeat_clear(true),
acking_node_list(NULL), watermark_pending(false),
rx_buffer_size(1024*1024), rx_sock_buffer_size(0), rx_buffer_size(1024*1024), rx_sock_buffer_size(0),
rx_cache_path(NULL), post_processor(NULL), unicast_nacks(false), silent_client(false), rx_cache_path(NULL), post_processor(NULL), unicast_nacks(false), silent_client(false),
tracing(false), tx_loss(0.0), rx_loss(0.0) tracing(false), tx_loss(0.0), rx_loss(0.0)
@ -170,6 +174,7 @@ NormApp::~NormApp()
if (interface_name) delete[] interface_name; if (interface_name) delete[] interface_name;
if (rx_cache_path) delete[] rx_cache_path; if (rx_cache_path) delete[] rx_cache_path;
if (post_processor) delete post_processor; if (post_processor) delete post_processor;
if (acking_node_list) delete[] acking_node_list;
} }
// NOTE on message flushing mode: // NOTE on message flushing mode:
@ -216,6 +221,7 @@ const char* const NormApp::cmd_list[] =
"+rinterval", // Interval (sec) between file/directory list repeats "+rinterval", // Interval (sec) between file/directory list repeats
"-oneshot", // Transmit file(s), exiting upon TX_FLUSH_COMPLETED "-oneshot", // Transmit file(s), exiting upon TX_FLUSH_COMPLETED
"-updatesOnly", // only send updated files on repeat transmission "-updatesOnly", // only send updated files on repeat transmission
"+ackingNodes", // comma-delimited list of node id's to from which to collect acks
"+rxcachedir", // recv file cache directory "+rxcachedir", // recv file cache directory
"+segment", // payload segment size (bytes) "+segment", // payload segment size (bytes)
"+block", // User data packets per FEC coding block (blockSize) "+block", // User data packets per FEC coding block (blockSize)
@ -267,6 +273,7 @@ void NormApp::ShowHelp()
" +rinterval, // Interval (sec) between file/directory list repeats\n" " +rinterval, // Interval (sec) between file/directory list repeats\n"
" -oneshot, // Exit upon sender TX_FLUSH_COMPLETED event (sender exits after transmission)\n" " -oneshot, // Exit upon sender TX_FLUSH_COMPLETED event (sender exits after transmission)\n"
" -updatesOnly, // only send updated files on repeat transmission\n" " -updatesOnly, // only send updated files on repeat transmission\n"
" +ackingNodes, // comma-delimited list of node id's to from which to collect acks\n"
" +rxcachedir, // recv file cache directory\n" " +rxcachedir, // recv file cache directory\n"
" +segment, // payload segment size (bytes)\n" " +segment, // payload segment size (bytes)\n"
" +block, // User data packets per FEC coding block (blockSize)\n" " +block, // User data packets per FEC coding block (blockSize)\n"
@ -653,6 +660,31 @@ bool NormApp::OnCommand(const char* cmd, const char* val)
else if (!strncmp("updatesOnly", cmd, len)) else if (!strncmp("updatesOnly", cmd, len))
{ {
tx_file_list.InitUpdateTime(true); tx_file_list.InitUpdateTime(true);
}
else if (!strncmp("ackingNodes", cmd, len))
{
size_t length = strlen(val);
if (NULL != acking_node_list)
length += strlen(acking_node_list) + 1;
char* tempString = new char[length + 1];
if (NULL == tempString)
{
DMSG(0, "NormApp::OnCommand(ackingNodes) error: %s\n", GetErrorString());
return false;
}
if (NULL != acking_node_list)
{
strcpy(tempString, acking_node_list);
strcat(tempString, ",");
delete[] acking_node_list;
}
else
{
tempString[0] = '\0';
}
acking_node_list = tempString;
strcat(acking_node_list, val);
if (NULL != session) return AddAckingNodes(acking_node_list);
} }
else if (!strncmp("rxcachedir", cmd, len)) else if (!strncmp("rxcachedir", cmd, len))
{ {
@ -909,6 +941,70 @@ bool NormApp::ProcessCommands(int argc, const char*const* argv)
return true; return true;
} // end NormApp::ProcessCommands() } // end NormApp::ProcessCommands()
bool NormApp::AddAckingNodes(const char* ackingNodeList)
{
if (NULL == ackingNodeList) return true;
if (NULL != session)
{
const char* ptr = ackingNodeList;
while ((NULL != ptr) && ('\0' != *ptr))
{
const char* end = strchr(ptr, ',');
size_t len = (NULL != end) ? (end - ptr) : strlen(ptr);
if (len >= 256)
{
DMSG(0, "NormApp::AddAckingNodes() error: bad acking node list\n");
return false;
}
char nodeString[256];
strncpy(nodeString, ptr, len);
nodeString[len] = '\0';
bool isNumber = true;
for (size_t i = 0; i < len; i++)
{
if (0 == isdigit(nodeString[i]))
{
isNumber = false;
break;
}
}
unsigned long nodeId;
if (isNumber)
{
if (1 != sscanf(nodeString, "%lu", &nodeId))
{
DMSG(0, "NormApp::AddAckingNodes() error: bad acking node id: %s\n", nodeString);
return false;
}
}
else
{
// It's an address or host name
ProtoAddress nodeAddr;
if (!nodeAddr.ResolveFromString(nodeString))
{
DMSG(0, "NormApp::AddAckingNodes() error: bad acking node id: %s\n", nodeString);
return false;
}
nodeId = nodeAddr.EndIdentifier();
}
if (!session->ServerAddAckingNode((UINT32)nodeId))
{
DMSG(0, "NormApp::AddAckingNodes() error: couldn't add acking node \n");
return false;
}
ptr = (NULL != end) ? ++end : NULL;
}
return true;
}
else
{
DMSG(0, "NormApp::AddAckingNodes() error: no session instantiated\n");
return false;
}
} // end NormApp::AddAckingNodes()
void NormApp::DoInputReady(ProtoDispatcher::Descriptor /*descriptor*/, void NormApp::DoInputReady(ProtoDispatcher::Descriptor /*descriptor*/,
ProtoDispatcher::Event /*theEvent*/, ProtoDispatcher::Event /*theEvent*/,
const void* userData) const void* userData)
@ -1100,8 +1196,8 @@ void NormApp::Notify(NormController::Event event,
} }
break; break;
case TX_QUEUE_EMPTY: case TX_QUEUE_EMPTY:
// Write to stream as needed // Write to stream as needed
//DMSG(0, "NormApp::Notify(TX_QUEUE_EMPTY) ...\n"); DMSG(3, "NormApp::Notify(TX_QUEUE_EMPTY) ...\n");
if (NULL != object) if (NULL != object)
{ {
if (input && (object == tx_stream)) if (input && (object == tx_stream))
@ -1118,17 +1214,33 @@ void NormApp::Notify(NormController::Event event,
} }
break; break;
case TX_OBJECT_SENT:
DMSG(3, "NormApp::Notify(TX_OBJECT_SENT) ...\n");
break;
case TX_FLUSH_COMPLETED: case TX_FLUSH_COMPLETED:
DMSG(3, "NormApp::Notify(TX_FLUSH_COMPLETED) ...\n");
if (tx_one_shot) if (tx_one_shot)
{ {
DMSG(0, "norm: transmit flushing completed, exiting.\n"); DMSG(0, "norm: transmit flushing completed, exiting.\n");
Stop(); Stop();
} }
break; break;
case TX_WATERMARK_COMPLETED:
{
DMSG(3, "NormApp::Notify(TX_WATERMARK_COMPLETED) ...\n");
NormSession::AckingStatus status = session->ServerGetAckingStatus(NORM_NODE_ANY);
DMSG(0, "norm: positive ack collection completed (%s)\n",
(NormSession::ACK_SUCCESS == status) ? "succeed" : "failed");
watermark_pending = false; // enable new watermark to be set
break;
}
case RX_OBJECT_NEW: case RX_OBJECT_NEW:
{ {
//DMSG(0, "NormApp::Notify(RX_OBJECT_NEW) ...\n"); DMSG(3, "NormApp::Notify(RX_OBJECT_NEW) ...\n");
// It's up to the app to "accept" the object // It's up to the app to "accept" the object
switch (object->GetType()) switch (object->GetType())
{ {
@ -1230,7 +1342,7 @@ void NormApp::Notify(NormController::Event event,
} }
case RX_OBJECT_INFO: case RX_OBJECT_INFO:
//DMSG(0, "NormApp::Notify(RX_OBJECT_INFO) ...\n"); DMSG(3, "NormApp::Notify(RX_OBJECT_INFO) ...\n");
switch(object->GetType()) switch(object->GetType())
{ {
case NormObject::FILE: case NormObject::FILE:
@ -1271,7 +1383,7 @@ void NormApp::Notify(NormController::Event event,
break; break;
case RX_OBJECT_UPDATED: case RX_OBJECT_UPDATED:
//DMSG(0, "NormApp::Notify(RX_OBJECT_UPDATED) ...\n"); DMSG(3, "NormApp::Notify(RX_OBJECT_UPDATED) ...\n");
switch (object->GetType()) switch (object->GetType())
{ {
case NormObject::FILE: case NormObject::FILE:
@ -1414,7 +1526,7 @@ void NormApp::Notify(NormController::Event event,
// (TBD) if we're not archiving files we should // (TBD) if we're not archiving files we should
// manage our cache, deleting the cache // manage our cache, deleting the cache
// on shutdown ... // on shutdown ...
//DMSG(0, "NormApp::Notify(RX_OBJECT_COMPLETE) ...\n"); DMSG(3, "NormApp::Notify(RX_OBJECT_COMPLETED) ...\n");
switch(object->GetType()) switch(object->GetType())
{ {
case NormObject::FILE: case NormObject::FILE:
@ -1477,7 +1589,8 @@ bool NormApp::OnIntervalTimeout(ProtoTimer& /*theTimer*/)
char temp[PATH_MAX]; char temp[PATH_MAX];
strncpy(temp, fileNameInfo, len); strncpy(temp, fileNameInfo, len);
temp[len] = '\0'; temp[len] = '\0';
if (!session->QueueTxFile(fileName, fileNameInfo, (UINT16)len)) NormFileObject* obj;
if (NULL == (obj = session->QueueTxFile(fileName, fileNameInfo, (UINT16)len)))
{ {
DMSG(0, "NormApp::OnIntervalTimeout() Error queuing tx file: %s\n", DMSG(0, "NormApp::OnIntervalTimeout() Error queuing tx file: %s\n",
fileName); fileName);
@ -1487,6 +1600,15 @@ bool NormApp::OnIntervalTimeout(ProtoTimer& /*theTimer*/)
ActivateTimer(interval_timer); ActivateTimer(interval_timer);
return false; return false;
} }
if (!watermark_pending && (NULL != acking_node_list))
{
NormBlockId blockId = obj->GetFinalBlockId();
NormSegmentId segmentId = obj->GetBlockSize(blockId) - 1;
session->ServerSetWatermark(obj->GetId(),
blockId,
segmentId);
watermark_pending = true; // only allow one pending watermark at a time
}
//DMSG(0, "norm: File \"%s\" queued for transmission.\n", fileName); //DMSG(0, "norm: File \"%s\" queued for transmission.\n", fileName);
interval_timer.SetInterval(tx_object_interval); interval_timer.SetInterval(tx_object_interval);
} }
@ -1587,6 +1709,13 @@ bool NormApp::OnStartup(int argc, const char*const* argv)
session->SetBackoffFactor(backoff_factor); session->SetBackoffFactor(backoff_factor);
session->ServerSetGrtt(grtt_estimate); session->ServerSetGrtt(grtt_estimate);
session->ServerSetGroupSize(group_size); session->ServerSetGroupSize(group_size);
if (!AddAckingNodes(acking_node_list))
{
DMSG(0, "NormApp::OnStartup() error: bad acking node list\n");
session_mgr.Destroy();
return false;
}
// We also use the baseId as our server's "instance id" for illustrative purposes // We also use the baseId as our server's "instance id" for illustrative purposes
UINT16 instanceId = baseId; UINT16 instanceId = baseId;
if (!session->StartServer(instanceId, tx_buffer_size, segment_size, ndata, nparity, interface_name)) if (!session->StartServer(instanceId, tx_buffer_size, segment_size, ndata, nparity, interface_name))

32
common/normEncoder.cpp
View File

@ -103,7 +103,7 @@ bool NormEncoder::CreateGeneratorPolynomial()
{ {
unsigned char *tp, *tp1, *tp2; unsigned char *tp, *tp1, *tp2;
int degree = 2*npar; int degree = 2*npar;
if(gen_poly) delete gen_poly; if(gen_poly) delete[] gen_poly;
if(!(gen_poly = new unsigned char[npar+1])) if(!(gen_poly = new unsigned char[npar+1]))
{ {
@ -116,22 +116,22 @@ bool NormEncoder::CreateGeneratorPolynomial()
{ {
DMSG(0, "NormEncoder: Error allocating memory while computing gen_poly: %s\n", DMSG(0, "NormEncoder: Error allocating memory while computing gen_poly: %s\n",
GetErrorString()); GetErrorString());
delete gen_poly; delete[] gen_poly;
return false; return false;
} }
if(!(tp1 = new unsigned char[2*degree])) if(!(tp1 = new unsigned char[2*degree]))
{ {
delete tp; delete[] tp;
delete gen_poly; delete[] gen_poly;
DMSG(0, "NormEncoder: Error allocating memory while computing gen_poly: %s\n", DMSG(0, "NormEncoder: Error allocating memory while computing gen_poly: %s\n",
GetErrorString()); GetErrorString());
return false; return false;
} }
if(!(tp2 = new unsigned char[2*degree])) if(!(tp2 = new unsigned char[2*degree]))
{ {
delete tp1; delete[] tp1;
delete tp; delete[] tp;
delete gen_poly; delete[] gen_poly;
DMSG(0, "NormEncoder: Error allocating memory while computing gen_poly: %s\n", DMSG(0, "NormEncoder: Error allocating memory while computing gen_poly: %s\n",
GetErrorString()); GetErrorString());
return false; return false;
@ -161,9 +161,9 @@ bool NormEncoder::CreateGeneratorPolynomial()
memcpy(tp1, gen_poly, (npar+1)*sizeof(unsigned char)); memcpy(tp1, gen_poly, (npar+1)*sizeof(unsigned char));
memset(&tp1[npar+1], 0, (2*degree)-(npar+1)); memset(&tp1[npar+1], 0, (2*degree)-(npar+1));
} }
delete tp2; delete[] tp2;
delete tp1; delete[] tp1;
delete tp; delete[] tp;
return true; return true;
} // end NormEncoder::CreateGeneratorPolynomial() } // end NormEncoder::CreateGeneratorPolynomial()
@ -303,27 +303,27 @@ void NormDecoder::Destroy()
{ {
if (scratch) if (scratch)
{ {
delete scratch; delete[] scratch;
scratch = NULL; scratch = NULL;
} }
if(o_vec) if(o_vec)
{ {
for(int i=0; i<npar; i++) for(int i=0; i<npar; i++)
if (o_vec[i]) delete o_vec[i]; if (o_vec[i]) delete[] o_vec[i];
delete o_vec; delete[] o_vec;
o_vec = NULL; o_vec = NULL;
} }
if(s_vec) if(s_vec)
{ {
for(int i = 0; i < npar; i++) for(int i = 0; i < npar; i++)
if (s_vec[i]) delete s_vec[i]; if (s_vec[i]) delete[] s_vec[i];
delete s_vec; delete[] s_vec;
s_vec = NULL; s_vec = NULL;
} }
if (lambda) if (lambda)
{ {
delete lambda; delete[] lambda;
lambda = NULL; lambda = NULL;
} }
} // end NormDecoder::Destroy() } // end NormDecoder::Destroy()

42
common/normMessage.h
View File

@ -92,14 +92,14 @@ inline double NormUnquantizeLoss(unsigned short lossQuantized)
inline unsigned short NormQuantizeRate(double rate) inline unsigned short NormQuantizeRate(double rate)
{ {
if (rate <= 0.0) return 0x01; // rate = 0.0 if (rate <= 0.0) return 0x01; // rate = 0.0
unsigned char exponent = (unsigned char)log10(rate); unsigned short exponent = (unsigned short)log10(rate);
unsigned short mantissa = (unsigned short)((256.0/10.0) * rate / pow(10.0, (double)exponent)); unsigned short mantissa = (unsigned short)((4096.0/10.0) * (rate / pow(10.0, (double)exponent)) + 0.5);
return ((mantissa << 8) | exponent); return ((mantissa << 4) | exponent);
} }
inline double NormUnquantizeRate(unsigned short rate) inline double NormUnquantizeRate(unsigned short rate)
{ {
double mantissa = ((double)(rate >> 8)) * (10.0/256.0); double mantissa = ((double)(rate >> 4)) * (10.0/4096.0);
double exponent = (double)(rate & 0xff); double exponent = (double)(rate & 0x000f);
return mantissa * pow(10.0, exponent); return mantissa * pow(10.0, exponent);
} }
@ -663,41 +663,39 @@ class NormDataMsg : public NormObjectMsg
{ {
((UINT8*)payload)[PAYLOAD_FLAGS_OFFSET] |= flag; ((UINT8*)payload)[PAYLOAD_FLAGS_OFFSET] |= flag;
}*/ }*/
static void WriteStreamPayloadMsgStart(char* payload, UINT16 msgStartOffset)
{
UINT16 temp16 = htons(msgStartOffset);
memcpy(payload+PAYLOAD_MSG_START_OFFSET, &temp16, 2);
}
static void WriteStreamPayloadLength(char* payload, UINT16 len) static void WriteStreamPayloadLength(char* payload, UINT16 len)
{ {
UINT16 temp16 = htons(len); UINT16 temp16 = htons(len);
memcpy(payload+PAYLOAD_LENGTH_OFFSET, &temp16, 2); memcpy(payload+PAYLOAD_LENGTH_OFFSET, &temp16, 2);
} }
static void WriteStreamPayloadMsgStart(char* payload, UINT16 msgStartOffset)
{
UINT16 temp16 = htons(msgStartOffset);
memcpy(payload+PAYLOAD_MSG_START_OFFSET, &temp16, 2);
}
static void WriteStreamPayloadOffset(char* payload, UINT32 offset) static void WriteStreamPayloadOffset(char* payload, UINT32 offset)
{ {
UINT32 temp32 = htonl(offset); UINT32 temp32 = htonl(offset);
memcpy(payload+PAYLOAD_OFFSET_OFFSET, &temp32, 4); memcpy(payload+PAYLOAD_OFFSET_OFFSET, &temp32, 4);
} }
static UINT16 ReadStreamPayloadMsgStart(const char* payload)
{
UINT16 temp16;
memcpy(&temp16, payload+PAYLOAD_MSG_START_OFFSET, 2);
return (ntohs(temp16));
}
static UINT16 ReadStreamPayloadLength(const char* payload) static UINT16 ReadStreamPayloadLength(const char* payload)
{ {
UINT16 temp16; UINT16 temp16;
memcpy(&temp16, payload+PAYLOAD_LENGTH_OFFSET, 2); memcpy(&temp16, payload+PAYLOAD_LENGTH_OFFSET, 2);
return (ntohs(temp16)); return (ntohs(temp16));
} }
static UINT16 ReadStreamPayloadMsgStart(const char* payload)
{
UINT16 temp16;
memcpy(&temp16, payload+PAYLOAD_MSG_START_OFFSET, 2);
return (ntohs(temp16));
}
static UINT32 ReadStreamPayloadOffset(const char* payload) static UINT32 ReadStreamPayloadOffset(const char* payload)
{ {
UINT32 temp32; UINT32 temp32;
memcpy(&temp32, payload+PAYLOAD_OFFSET_OFFSET, 4); memcpy(&temp32, payload+PAYLOAD_OFFSET_OFFSET, 4);
return (ntohl(temp32)); return (ntohl(temp32));
} }
private: private:
enum enum
@ -710,9 +708,9 @@ class NormDataMsg : public NormObjectMsg
enum enum
{ {
//PAYLOAD_FLAGS_OFFSET = 0, // deprecated //PAYLOAD_FLAGS_OFFSET = 0, // deprecated
PAYLOAD_MSG_START_OFFSET = 0, PAYLOAD_LENGTH_OFFSET = 0,
PAYLOAD_LENGTH_OFFSET = PAYLOAD_MSG_START_OFFSET+2, PAYLOAD_MSG_START_OFFSET = PAYLOAD_LENGTH_OFFSET+2,
PAYLOAD_OFFSET_OFFSET = PAYLOAD_LENGTH_OFFSET+2, PAYLOAD_OFFSET_OFFSET = PAYLOAD_MSG_START_OFFSET+2,
PAYLOAD_DATA_OFFSET = PAYLOAD_OFFSET_OFFSET+4 PAYLOAD_DATA_OFFSET = PAYLOAD_OFFSET_OFFSET+4
}; };
}; // end class NormDataMsg }; // end class NormDataMsg
@ -1578,7 +1576,7 @@ class NormAckFlushMsg : public NormAckMsg
} }
private: private:
// These are the payload offsets for "fec_id" = 129 // Note - These are the payload offsets for "fec_id" = 129
// "fec_payload_id" field // "fec_payload_id" field
enum enum
{ {

209
common/normNode.cpp
View File

@ -52,7 +52,7 @@ NormServerNode::NormServerNode(class NormSession& theSession, NormNodeId nodeId)
cc_sequence(0), cc_enable(false), cc_rate(0.0), cc_sequence(0), cc_enable(false), cc_rate(0.0),
rtt_confirmed(false), is_clr(false), is_plr(false), rtt_confirmed(false), is_clr(false), is_plr(false),
slow_start(true), send_rate(0.0), recv_rate(0.0), recv_accumulator(0), slow_start(true), send_rate(0.0), recv_rate(0.0), recv_accumulator(0),
recv_total(0), recv_goodput(0), resync_count(0), nominal_packet_size(0), recv_total(0), recv_goodput(0), resync_count(0),
nack_count(0), suppress_count(0), completion_count(0), failure_count(0) nack_count(0), suppress_count(0), completion_count(0), failure_count(0)
{ {
@ -67,7 +67,7 @@ NormServerNode::NormServerNode(class NormSession& theSession, NormNodeId nodeId)
repair_timer.SetRepeat(1); repair_timer.SetRepeat(1);
activity_timer.SetListener(this, &NormServerNode::OnActivityTimeout); activity_timer.SetListener(this, &NormServerNode::OnActivityTimeout);
double activityInterval = NormSession::DEFAULT_GRTT_ESTIMATE*NORM_ROBUST_FACTOR; double activityInterval = 2*NormSession::DEFAULT_GRTT_ESTIMATE*NORM_ROBUST_FACTOR;
if (activityInterval < ACTIVITY_INTERVAL_MIN) activityInterval = ACTIVITY_INTERVAL_MIN; if (activityInterval < ACTIVITY_INTERVAL_MIN) activityInterval = ACTIVITY_INTERVAL_MIN;
activity_timer.SetInterval(activityInterval); activity_timer.SetInterval(activityInterval);
activity_timer.SetRepeat(NORM_ROBUST_FACTOR); activity_timer.SetRepeat(NORM_ROBUST_FACTOR);
@ -97,8 +97,6 @@ NormServerNode::NormServerNode(class NormSession& theSession, NormNodeId nodeId)
prev_update_time.tv_sec = 0; prev_update_time.tv_sec = 0;
prev_update_time.tv_usec = 0; prev_update_time.tv_usec = 0;
recv_rate_quantized = NormQuantizeRate(recv_rate);
} }
NormServerNode::~NormServerNode() NormServerNode::~NormServerNode()
@ -299,29 +297,27 @@ void NormServerNode::FreeBuffers()
segment_size = ndata = nparity = 0; segment_size = ndata = nparity = 0;
} // end NormServerNode::FreeBuffers() } // end NormServerNode::FreeBuffers()
void NormServerNode::UpdateGrttEstimate(UINT8 grttQuantized)
{
grtt_quantized = grttQuantized;
grtt_estimate = NormUnquantizeRtt(grttQuantized);
DMSG(4, "NormServerNode::UpdateGrttEstimate() node>%lu server>%lu new grtt: %lf sec\n",
LocalNodeId(), GetId(), grtt_estimate);
// activity timer depends upon sender's grtt estimate
double activityInterval = 2*NORM_ROBUST_FACTOR*grtt_estimate;
if (activityInterval < ACTIVITY_INTERVAL_MIN) activityInterval = ACTIVITY_INTERVAL_MIN;
activity_timer.SetInterval(activityInterval);
if (activity_timer.IsActive()) activity_timer.Reschedule();
// (TBD) Scale/reschedule repair_timer and/or cc_timer???
session.Notify(NormController::GRTT_UPDATED, this, (NormObject*)NULL);
} // end NormServerNode::UpdateGrttEstimate()
void NormServerNode::HandleCommand(const struct timeval& currentTime, void NormServerNode::HandleCommand(const struct timeval& currentTime,
const NormCmdMsg& cmd) const NormCmdMsg& cmd)
{ {
UINT8 grttQuantized = cmd.GetGrtt(); UINT8 grttQuantized = cmd.GetGrtt();
if (grttQuantized != grtt_quantized) if (grttQuantized != grtt_quantized) UpdateGrttEstimate(grttQuantized);
{
grtt_quantized = grttQuantized;
grtt_estimate = NormUnquantizeRtt(grttQuantized);
DMSG(4, "NormServerNode::HandleCommand() node>%lu server>%lu new grtt: %lf sec\n",
LocalNodeId(), GetId(), grtt_estimate);
// Possible change in activity timeout
double nominalSize = (nominal_packet_size > segment_size) ? nominal_packet_size : segment_size;
double activityInterval =
(0.0 != recv_rate) ? (nominalSize / recv_rate) : DEFAULT_NOMINAL_INTERVAL;
if (grtt_estimate > activityInterval) activityInterval = grtt_estimate;
activityInterval *= NORM_ROBUST_FACTOR;
if (activityInterval < ACTIVITY_INTERVAL_MIN)
activityInterval = ACTIVITY_INTERVAL_MIN;
activity_timer.SetInterval(activityInterval);
if (activity_timer.IsActive()) activity_timer.Reschedule();
session.Notify(NormController::GRTT_UPDATED, this, (NormObject*)NULL);
}
UINT8 gsizeQuantized = cmd.GetGroupSize(); UINT8 gsizeQuantized = cmd.GetGroupSize();
if (gsizeQuantized != gsize_quantized) if (gsizeQuantized != gsize_quantized)
{ {
@ -421,16 +417,16 @@ void NormServerNode::HandleCommand(const struct timeval& currentTime,
ExponentialRand(maxBackoff, gsize_estimate) : 0.0; ExponentialRand(maxBackoff, gsize_estimate) : 0.0;
// Bias backoff timeout based on our rate // Bias backoff timeout based on our rate
double r; double r;
if (slow_start) double ccLoss = slow_start ? 0.0 : LossEstimate();
if (0.0 == ccLoss)
{ {
r = recv_rate / send_rate; r = recv_rate / send_rate;
cc_rate = 2.0 * recv_rate; cc_rate = 2.0 * recv_rate;
} }
else else
{ {
cc_rate = NormSession::CalculateRate(nominal_packet_size, double nominalSize = nominal_packet_size ? nominal_packet_size : segment_size;
rtt_estimate, cc_rate = NormSession::CalculateRate(nominalSize, rtt_estimate, ccLoss);
LossEstimate());
r = cc_rate / send_rate; r = cc_rate / send_rate;
r = MIN(r, 0.9); r = MIN(r, 0.9);
r = MAX(r, 0.5); r = MAX(r, 0.5);
@ -551,11 +547,13 @@ void NormServerNode::HandleCCFeedback(UINT8 ccFlags, double ccRate)
{ {
// We're suppressed by non-CLR receivers with no RTT confirmed // We're suppressed by non-CLR receivers with no RTT confirmed
// and/or lower rate // and/or lower rate
double localRate = slow_start ? double nominalSize = nominal_packet_size ? nominal_packet_size : segment_size;
(2.0*recv_rate) : double ccLoss = slow_start ? 0.0 : LossEstimate();
NormSession::CalculateRate(nominal_packet_size, double localRate = (0.0 == ccLoss) ?
rtt_estimate, (2.0*recv_rate) :
LossEstimate()); NormSession::CalculateRate(nominalSize,
rtt_estimate,
ccLoss);
// This increases our chance of being suppressed // This increases our chance of being suppressed
// (but is it a good idea?) // (but is it a good idea?)
localRate = MAX(localRate, cc_rate); localRate = MAX(localRate, cc_rate);
@ -890,27 +888,7 @@ char* NormServerNode::GetFreeSegment(NormObjectId objectId, NormBlockId blockId)
void NormServerNode::HandleObjectMessage(const NormObjectMsg& msg) void NormServerNode::HandleObjectMessage(const NormObjectMsg& msg)
{ {
UINT8 grttQuantized = msg.GetGrtt(); UINT8 grttQuantized = msg.GetGrtt();
if (grttQuantized != grtt_quantized) if (grttQuantized != grtt_quantized) UpdateGrttEstimate(grttQuantized);
{
grtt_quantized = grttQuantized;
grtt_estimate = NormUnquantizeRtt(grttQuantized);
DMSG(4, "NormServerNode::HandleObjectMessage() node>%lu server>%lu new grtt: %lf sec\n",
LocalNodeId(), GetId(), grtt_estimate);
// Scale and reschedule pertinent timeouts repair_timer
// Possible change in sender activity timeout
double nominalSize = (nominal_packet_size > segment_size) ? nominal_packet_size : segment_size;
double activityInterval =
(0.0 != recv_rate) ? (nominalSize / recv_rate) : DEFAULT_NOMINAL_INTERVAL;
if (grtt_estimate > activityInterval) activityInterval = grtt_estimate;
activityInterval *= NORM_ROBUST_FACTOR;
if (activityInterval < ACTIVITY_INTERVAL_MIN)
activityInterval = ACTIVITY_INTERVAL_MIN;
activity_timer.SetInterval(activityInterval);
if (activity_timer.IsActive()) activity_timer.Reschedule();
session.Notify(NormController::GRTT_UPDATED, this, (NormObject*)NULL);
}
UINT8 gsizeQuantized = msg.GetGroupSize(); UINT8 gsizeQuantized = msg.GetGroupSize();
if (gsizeQuantized != gsize_quantized) if (gsizeQuantized != gsize_quantized)
{ {
@ -1123,9 +1101,11 @@ void NormServerNode::HandleObjectMessage(const NormObjectMsg& msg)
#endif // !SIMULATE #endif // !SIMULATE
if (NormObject::STREAM != obj->GetType()) if (NormObject::STREAM != obj->GetType())
{ {
// Streams never complete // Streams never complete unless they are "closed" by sender
// and this handled within stream control code in "normObject.cpp"
session.Notify(NormController::RX_OBJECT_COMPLETED, this, obj); session.Notify(NormController::RX_OBJECT_COMPLETED, this, obj);
DeleteObject(obj); DeleteObject(obj);
obj = NULL;
completion_count++; completion_count++;
} }
} }
@ -1141,7 +1121,7 @@ void NormServerNode::HandleObjectMessage(const NormObjectMsg& msg)
// Optional "object boundary repair check (non-streams only!) // Optional "object boundary repair check (non-streams only!)
// (checks for repair needs for objects _prior_ to current objectId) // (checks for repair needs for objects _prior_ to current objectId)
// (also requests "info" for current objectId) // (also requests "info" for current objectId)
if (obj && (NormObject::STREAM == obj->GetType())) if (obj && obj->IsStream())
RepairCheck(NormObject::TO_BLOCK, objectId, blockId, segmentId); RepairCheck(NormObject::TO_BLOCK, objectId, blockId, segmentId);
else else
RepairCheck(NormObject::THRU_INFO, objectId, blockId, segmentId); RepairCheck(NormObject::THRU_INFO, objectId, blockId, segmentId);
@ -1368,7 +1348,8 @@ void NormServerNode::RepairCheck(NormObject::CheckLevel checkLevel,
NormSegmentId segmentId) NormSegmentId segmentId)
{ {
ASSERT(synchronized); ASSERT(synchronized);
if (objectId > max_pending_object) max_pending_object = objectId; if (objectId > max_pending_object)
max_pending_object = objectId;
if (!repair_timer.IsActive()) if (!repair_timer.IsActive())
{ {
// repair timer inactive // repair timer inactive
@ -1507,17 +1488,19 @@ bool NormServerNode::OnRepairTimeout(ProtoTimer& /*theTimer*/)
if (rtt_confirmed) if (rtt_confirmed)
ext.SetCCFlag(NormCC::RTT); ext.SetCCFlag(NormCC::RTT);
ext.SetCCRtt(rtt_quantized); ext.SetCCRtt(rtt_quantized);
double ccLoss = LossEstimate(); double ccLoss = slow_start ? 0.0 : LossEstimate();
UINT16 lossQuantized = NormQuantizeLoss(ccLoss); UINT16 lossQuantized = NormQuantizeLoss(ccLoss);
ext.SetCCLoss(lossQuantized); ext.SetCCLoss(lossQuantized);
if (slow_start) if (0.0 == ccLoss)
{ {
ext.SetCCFlag(NormCC::START); //if (slow_start) // (TBD) should we only set flag on actual slow_start?
ext.SetCCFlag(NormCC::START);
ext.SetCCRate(NormQuantizeRate(2.0 * recv_rate)); ext.SetCCRate(NormQuantizeRate(2.0 * recv_rate));
} }
else else
{ {
double ccRate = NormSession::CalculateRate(nominal_packet_size, double nominalSize = (nominal_packet_size > segment_size) ? nominal_packet_size : segment_size;
double ccRate = NormSession::CalculateRate(nominalSize,
rtt_estimate, rtt_estimate,
ccLoss); ccLoss);
ext.SetCCRate(NormQuantizeRate(ccRate)); ext.SetCCRate(NormQuantizeRate(ccRate));
@ -1735,10 +1718,8 @@ void NormServerNode::UpdateRecvRate(const struct timeval& currentTime, unsigned
// Here, we put a NORM_TICK_MIN sec lower bound on our measurementInterval for the // Here, we put a NORM_TICK_MIN sec lower bound on our measurementInterval for the
// recv_rate because of the typical limited granularity of our system clock // recv_rate because of the typical limited granularity of our system clock
// (Note this can limit our ramp up of data rate during slow start) // (Note this can limit our ramp up of data rate during slow start)
if (measurementInterval < NORM_TICK_MIN) if (measurementInterval < NORM_TICK_MIN) measurementInterval = NORM_TICK_MIN;
measurementInterval = NORM_TICK_MIN;
recv_accumulator += msgSize; recv_accumulator += msgSize;
unsigned short oldRateQuantized = recv_rate_quantized;
if (interval > 0.0) if (interval > 0.0)
{ {
double currentRecvRate = ((double)(recv_accumulator)) / interval; double currentRecvRate = ((double)(recv_accumulator)) / interval;
@ -1759,36 +1740,17 @@ void NormServerNode::UpdateRecvRate(const struct timeval& currentTime, unsigned
{ {
recv_rate = currentRecvRate; recv_rate = currentRecvRate;
} }
recv_rate_quantized = NormQuantizeRate(recv_rate);
} }
else if (0.0 == recv_rate) else if (0.0 == recv_rate)
{ {
// Approximate initial recv_rate when initial packets arrive in a burst // Approximate initial recv_rate when initial packets arrive in a burst
recv_rate = ((double)(recv_accumulator)) / NORM_TICK_MIN; recv_rate = ((double)(recv_accumulator)) / NORM_TICK_MIN;
recv_rate_quantized = NormQuantizeRate(recv_rate);
}
nominal_packet_size += 0.05 * (((double)msgSize) - nominal_packet_size);
if (recv_rate_quantized != oldRateQuantized)
{
double nominalSize = (nominal_packet_size > segment_size) ? nominal_packet_size : segment_size;
double activityInterval =
(0.0 != recv_rate) ? (nominalSize / recv_rate) : DEFAULT_NOMINAL_INTERVAL;
if ((activity_timer.GetInterval() > (NORM_ROBUST_FACTOR*grtt_estimate)) ||
(activityInterval > grtt_estimate))
{
if (grtt_estimate > activityInterval) activityInterval = grtt_estimate;
activityInterval *= NORM_ROBUST_FACTOR;
if (activityInterval < ACTIVITY_INTERVAL_MIN)
activityInterval = ACTIVITY_INTERVAL_MIN;
activity_timer.SetInterval(activityInterval);
if (activity_timer.IsActive()) activity_timer.Reschedule();
}
} }
nominal_packet_size += 0.05 * (((double)msgSize) - nominal_packet_size);
} }
else else
{ {
recv_rate = 0.0; recv_rate = 0.0;
recv_rate_quantized = NormQuantizeRate(0.0);
prev_update_time = currentTime; prev_update_time = currentTime;
recv_accumulator = 0; recv_accumulator = 0;
nominal_packet_size = msgSize; nominal_packet_size = msgSize;
@ -1800,13 +1762,8 @@ void NormServerNode::Activate(bool isObjectMsg)
{ {
if (!activity_timer.IsActive()) if (!activity_timer.IsActive())
{ {
double nominalSize = (nominal_packet_size > segment_size) ? nominal_packet_size : segment_size; double activityInterval = 2*NORM_ROBUST_FACTOR*grtt_estimate;
double activityInterval = if (activityInterval < ACTIVITY_INTERVAL_MIN) activityInterval = ACTIVITY_INTERVAL_MIN;
(0.0 != recv_rate) ? (nominalSize / recv_rate) : DEFAULT_NOMINAL_INTERVAL;
if (grtt_estimate > activityInterval) activityInterval = grtt_estimate;
activityInterval *= NORM_ROBUST_FACTOR;
if (activityInterval < ACTIVITY_INTERVAL_MIN)
activityInterval = ACTIVITY_INTERVAL_MIN;
activity_timer.SetInterval(activityInterval); activity_timer.SetInterval(activityInterval);
session.ActivateTimer(activity_timer); session.ActivateTimer(activity_timer);
server_active = false; server_active = false;
@ -1834,7 +1791,7 @@ bool NormServerNode::OnActivityTimeout(ProtoTimer& /*theTimer*/)
} }
else else
{ {
DMSG(4, "NormServerNode::OnActivityTimeout() node>%lu for server>%lu\n", DMSG(2, "NormServerNode::OnActivityTimeout() node>%lu for server>%lu\n",
LocalNodeId(), GetId()); LocalNodeId(), GetId());
struct timeval currentTime; struct timeval currentTime;
::ProtoSystemTime(currentTime); ::ProtoSystemTime(currentTime);
@ -1858,6 +1815,19 @@ bool NormServerNode::OnActivityTimeout(ProtoTimer& /*theTimer*/)
// Let's try this instead // Let's try this instead
RepairCheck(NormObject::THRU_OBJECT, // (TBD) thru object??? RepairCheck(NormObject::THRU_OBJECT, // (TBD) thru object???
max_pending_object, 0, 0); max_pending_object, 0, 0);
// (TBD) What should we really do here? Our current NormNode::RepairCheck() and
// NormObject::ClientRepairCheck() methods update the "max_pending" indices
// so we _could_ make ourselves NACK for more repair than we should
// when the inactivity timeout kicks in ??? But if we don't NACK "thru object"
// we may miss something at end-of-transmission by not not NACKing? I guess
// the reliability really is in the flush process and our activity timeout NACK
// is "iffy, at best" ... Perhaps we need to have some sort of "wildcard" NACK,
// but then _everyone_ would NACK at EOT all the time, often for nothing ... so
// I guess the activity timeout NACK isn't perfect ... but could help some
// so we leave it as it is for the moment ("THRU_OBJECT") ... perhaps we could
// add a parameter so NormObject::ClientRepairCheck() doesn't update its
// "max_pending" indices - or would this break NACK building
} }
else else
{ {
@ -1876,24 +1846,30 @@ bool NormServerNode::UpdateLossEstimate(const struct timeval& currentTime,
unsigned short seq, unsigned short seq,
bool ecn) bool ecn)
{ {
bool result = loss_estimator.Update(currentTime, seq, ecn); if (loss_estimator.Update(currentTime, seq, ecn))
if (result && slow_start)
{ {
// Calculate loss initialization based on current receive rate if (slow_start)
// and rtt estimation {
double lossInit = (recv_rate * rtt_estimate) / // Calculate loss initialization based on current receive rate
(nominal_packet_size*sqrt(3.0/2.0)); // and rtt estimation
lossInit = (lossInit*lossInit); double nominalSize = (nominal_packet_size > segment_size) ? nominal_packet_size : segment_size;
// At this point, "LastLossInterval()" will always be non-zero double lossInit = (recv_rate * rtt_estimate) / (nominalSize*sqrt(3.0/2.0));
double altLoss = (double)loss_estimator.LastLossInterval(); lossInit = (lossInit*lossInit);
double altInit = (altLoss > 0.0) ? (1.0 / altLoss) : 0.5; // At this point, "LastLossInterval()" will always be non-zero!
if (altInit < lossInit) double altLoss = (double)loss_estimator.LastLossInterval();
loss_estimator.SetInitialLoss(altInit); double altInit = (altLoss > 0.0) ? (1.0 / altLoss) : 0.5;
else if (altInit < lossInit)
loss_estimator.SetInitialLoss(lossInit); loss_estimator.SetInitialLoss(altInit);
slow_start = false; else
loss_estimator.SetInitialLoss(lossInit);
slow_start = false;
}
return true;
}
else
{
return false;
} }
return result;
} // end NormServerNode::UpdateLossEstimate() } // end NormServerNode::UpdateLossEstimate()
void NormServerNode::AttachCCFeedback(NormAckMsg& ack) void NormServerNode::AttachCCFeedback(NormAckMsg& ack)
@ -1910,18 +1886,18 @@ void NormServerNode::AttachCCFeedback(NormAckMsg& ack)
if (rtt_confirmed) if (rtt_confirmed)
ext.SetCCFlag(NormCC::RTT); ext.SetCCFlag(NormCC::RTT);
ext.SetCCRtt(rtt_quantized); ext.SetCCRtt(rtt_quantized);
double ccLoss = LossEstimate(); double ccLoss = slow_start ? 0.0 : LossEstimate();
UINT16 lossQuantized = NormQuantizeLoss(ccLoss); UINT16 lossQuantized = NormQuantizeLoss(ccLoss);
ext.SetCCLoss(lossQuantized); ext.SetCCLoss(lossQuantized);
if (slow_start) if (0.0 == ccLoss)
{ {
ext.SetCCFlag(NormCC::START); ext.SetCCFlag(NormCC::START);
ext.SetCCRate(NormQuantizeRate(2.0 * recv_rate)); ext.SetCCRate(NormQuantizeRate(2.0 * recv_rate));
} }
else else
{ {
double ccRate = NormSession::CalculateRate(nominal_packet_size, double nominalSize = (nominal_packet_size > segment_size) ? nominal_packet_size : segment_size;
rtt_estimate, ccLoss); double ccRate = NormSession::CalculateRate(nominalSize, rtt_estimate, ccLoss);
ext.SetCCRate(NormQuantizeRate(ccRate)); ext.SetCCRate(NormQuantizeRate(ccRate));
} }
//DMSG(0, "NormServerNode::AttachCCFeedback() node>%lu sending ACK rate:%lf kbps (rtt:%lf loss:%lf s:%lf recvRate:%lf) slow_start:%d\n", //DMSG(0, "NormServerNode::AttachCCFeedback() node>%lu sending ACK rate:%lf kbps (rtt:%lf loss:%lf s:%lf recvRate:%lf) slow_start:%d\n",
@ -2006,7 +1982,16 @@ bool NormServerNode::OnAckTimeout(ProtoTimer& /*theTimer*/)
AttachCCFeedback(*ack); AttachCCFeedback(*ack);
ack->SetObjectId(watermark_object_id); ack->SetObjectId(watermark_object_id);
ack->SetFecBlockId(watermark_block_id); ack->SetFecBlockId(watermark_block_id);
ack->SetFecBlockLen(ndata); // yuk // _Attempt_ to set the fec_payload_id source block length field appropriately
UINT16 blockLen;
NormObject* obj = rx_table.Find(watermark_object_id);
if (NULL != obj)
blockLen = obj->GetBlockSize(watermark_block_id);
else if (watermark_segment_id < ndata)
blockLen = ndata;
else
blockLen = watermark_segment_id;
ack->SetFecBlockLen(blockLen); // yuk
ack->SetFecSymbolId(watermark_segment_id); ack->SetFecSymbolId(watermark_segment_id);
if (unicast_nacks) if (unicast_nacks)
ack->SetDestination(GetAddress()); ack->SetDestination(GetAddress());
@ -2424,7 +2409,7 @@ double NormLossEstimator::LossFraction()
NormLossEstimator2::NormLossEstimator2() NormLossEstimator2::NormLossEstimator2()
: lag_mask(0xffffffff), lag_depth(0), lag_test_bit(0x01), : init(false), lag_mask(0xffffffff), lag_depth(0), lag_test_bit(0x01),
event_window(0), event_index(0), event_window(0), event_index(0),
event_window_time(0.0), event_index_time(0.0), event_window_time(0.0), event_index_time(0.0),
seeking_loss_event(true), seeking_loss_event(true),

7
common/normNode.h
View File

@ -148,7 +148,7 @@ class NormAckingNode : public NormNode
NormAckingNode(class NormSession& theSession, NormNodeId nodeId); NormAckingNode(class NormSession& theSession, NormNodeId nodeId);
~NormAckingNode(); ~NormAckingNode();
bool IsPending() const bool IsPending() const
{return (!ack_received && ( req_count > 0));} {return (!ack_received && (req_count > 0));}
void Reset(unsigned int maxAttempts = NORM_ROBUST_FACTOR) void Reset(unsigned int maxAttempts = NORM_ROBUST_FACTOR)
{ {
ack_received = false; ack_received = false;
@ -238,7 +238,9 @@ class NormServerNode : public NormNode
bool UnicastNacks() {return unicast_nacks;} bool UnicastNacks() {return unicast_nacks;}
void SetUnicastNacks(bool state) {unicast_nacks = state;} void SetUnicastNacks(bool state) {unicast_nacks = state;}
double GetGrttEstimate() {return grtt_estimate;}
void UpdateGrttEstimate(UINT8 grttQuantized);
double GetGrttEstimate() const {return grtt_estimate;}
bool UpdateLossEstimate(const struct timeval& currentTime, bool UpdateLossEstimate(const struct timeval& currentTime,
unsigned short theSequence, unsigned short theSequence,
@ -445,7 +447,6 @@ class NormServerNode : public NormNode
bool slow_start; bool slow_start;
double send_rate; // sender advertised rate double send_rate; // sender advertised rate
double recv_rate; // measured recv rate double recv_rate; // measured recv rate
unsigned short recv_rate_quantized;
struct timeval prev_update_time; // for recv_rate measurement struct timeval prev_update_time; // for recv_rate measurement
unsigned long recv_accumulator; // for recv_rate measurement unsigned long recv_accumulator; // for recv_rate measurement
double nominal_packet_size; double nominal_packet_size;

80
common/normObject.cpp
View File

@ -15,7 +15,7 @@ NormObject::NormObject(NormObject::Type theType,
transport_id(transportId), segment_size(0), pending_info(false), repair_info(false), transport_id(transportId), segment_size(0), pending_info(false), repair_info(false),
current_block_id(0), next_segment_id(0), current_block_id(0), next_segment_id(0),
max_pending_block(0), max_pending_segment(0), max_pending_block(0), max_pending_segment(0),
info_ptr(NULL), info_len(0), accepted(false), notify_on_update(true) info_ptr(NULL), info_len(0), first_pass(true), accepted(false), notify_on_update(true)
{ {
if (theServer) if (theServer)
{ {
@ -48,7 +48,9 @@ void NormObject::Release()
{ {
if (server) server->Release(); if (server) server->Release();
if (reference_count) if (reference_count)
{
reference_count--; reference_count--;
}
else else
{ {
DMSG(0, "NormObject::Release() releasing non-retained object?!\n"); DMSG(0, "NormObject::Release() releasing non-retained object?!\n");
@ -323,6 +325,7 @@ bool NormObject::TxReset(NormBlockId firstBlock, bool requeue)
if (requeue) block->ClearFlag(NormBlock::IN_REPAIR); // since we're requeuing if (requeue) block->ClearFlag(NormBlock::IN_REPAIR); // since we're requeuing
} }
} }
if (requeue) first_pass = true;
return increasedRepair; return increasedRepair;
} // end NormObject::TxReset() } // end NormObject::TxReset()
@ -544,6 +547,7 @@ bool NormObject::AppendRepairAdv(NormCmdRepairAdvMsg& cmd)
// This is used by sender for watermark check // This is used by sender for watermark check
bool NormObject::FindRepairIndex(NormBlockId& blockId, NormSegmentId& segmentId) const bool NormObject::FindRepairIndex(NormBlockId& blockId, NormSegmentId& segmentId) const
{ {
ASSERT(!server);
if (repair_info) if (repair_info)
{ {
blockId = 0; blockId = 0;
@ -1504,7 +1508,10 @@ bool NormObject::NextServerMsg(NormObjectMsg* msg)
NormBlockId blockId; NormBlockId blockId;
if (!GetFirstPending(blockId)) if (!GetFirstPending(blockId))
{ {
if (!IsStream()) // Attempt to advance stream (probably was repair-delayed)
if (IsStream())
static_cast<NormStreamObject*>(this)->StreamAdvance();
else
DMSG(0, "NormObject::NextServerMsg() pending object w/ no pending blocks?\n"); DMSG(0, "NormObject::NextServerMsg() pending object w/ no pending blocks?\n");
return false; return false;
} }
@ -1544,17 +1551,27 @@ bool NormObject::NextServerMsg(NormObjectMsg* msg)
} }
block->TxInit(blockId, numData, session.ServerAutoParity()); block->TxInit(blockId, numData, session.ServerAutoParity());
if (!block_buffer.Insert(block)) while (!block_buffer.Insert(block))
{ {
ASSERT(STREAM == type); ASSERT(STREAM == type);
if (blockId > block_buffer.RangeLo()) if (blockId > block_buffer.RangeLo())
{ {
NormBlock* b = block_buffer.Find(block_buffer.RangeLo()); NormBlock* b = block_buffer.Find(block_buffer.RangeLo());
ASSERT(b); ASSERT(NULL != b);
block_buffer.Remove(b); bool push = static_cast<NormStreamObject*>(this)->GetPushMode();
session.ServerPutFreeBlock(b); if (!push && (b->IsRepairPending() || IsRepairSet(b->GetId())))
bool success = block_buffer.Insert(block); {
ASSERT(success); // Pending repairs delaying stream advance
DMSG(0, "NormObject::NextServerMsg() node>%lu pending repairs delaying stream progress\n", LocalNodeId());
return false;
}
else
{
// Prune old non-pending block (or even pending if "push" enabled stream)
block_buffer.Remove(b);
session.ServerPutFreeBlock(b);
continue;
}
} }
else if (IsStream()) else if (IsStream())
{ {
@ -1574,6 +1591,7 @@ bool NormObject::NextServerMsg(NormObjectMsg* msg)
else else
{ {
ASSERT(0); ASSERT(0);
DMSG(0, "NormObject::NextServerMsg() invalid non-stream state!\n");
return false; return false;
} }
} }
@ -1629,9 +1647,9 @@ bool NormObject::NextServerMsg(NormObjectMsg* msg)
ASSERT(block->ParityReady(numData)); ASSERT(block->ParityReady(numData));
char* segment = block->GetSegment(segmentId); char* segment = block->GetSegment(segmentId);
ASSERT(segment); ASSERT(segment);
UINT16 payloadLength = block->GetSegSizeMax(); // segment_size; // We only need to send FEC content to cover the biggest segment
if (IsStream()) payloadLength += NormDataMsg::GetStreamPayloadHeaderLength(); // sent for the block.
data->SetPayload(segment, payloadLength); data->SetPayload(segment, block->GetSegSizeMax());
} }
block->UnsetPending(segmentId); block->UnsetPending(segmentId);
if (block->InRepair()) if (block->InRepair())
@ -1645,16 +1663,29 @@ bool NormObject::NextServerMsg(NormObjectMsg* msg)
pending_mask.Unset(blockId); pending_mask.Unset(blockId);
} }
// This lets NORM_STREAM objects continue indefinitely if (!pending_mask.IsSet())
if (IsStream() && !pending_mask.IsSet()) {
static_cast<NormStreamObject*>(this)->StreamAdvance(); if (IsStream())
{
// This lets NORM_STREAM objects continue indefinitely
static_cast<NormStreamObject*>(this)->StreamAdvance();
// (TBD) Is there a case where posting TX_OBJECT_SENT for a stream
// makes sense? E.g., so we could use NormRequeueObject() for streams?
}
else if (first_pass)
{
// "First pass" transmission of object (and any auto parity) has completed
first_pass = false;
session.Notify(NormController::TX_OBJECT_SENT, NULL, this);
}
}
return true; return true;
} // end NormObject::NextServerMsg() } // end NormObject::NextServerMsg()
void NormStreamObject::StreamAdvance() void NormStreamObject::StreamAdvance()
{ {
NormBlockId nextBlockId = stream_next_id; NormBlockId nextBlockId = stream_next_id;
// Make sure we won't prevent about any pending repairs // Make sure we won't prevent any pending repairs
if (repair_mask.CanSet(nextBlockId)) if (repair_mask.CanSet(nextBlockId))
{ {
if (block_buffer.CanInsert(nextBlockId)) if (block_buffer.CanInsert(nextBlockId))
@ -1671,6 +1702,7 @@ void NormStreamObject::StreamAdvance()
if (!block->IsTransmitPending()) if (!block->IsTransmitPending())
{ {
// ??? (TBD) Should this block be returned to the pool right now??? // ??? (TBD) Should this block be returned to the pool right now???
// especially for "push" enabled streams
if (pending_mask.Set(nextBlockId)) if (pending_mask.Set(nextBlockId))
stream_next_id++; stream_next_id++;
else else
@ -1747,7 +1779,7 @@ NormBlock* NormObject::ServerRecoverBlock(NormBlockId blockId)
return (NormBlock*)NULL; return (NormBlock*)NULL;
} }
} }
// Attempt to re-generate parity // Attempt to re-generate parity for the block
if (CalculateBlockParity(block)) if (CalculateBlockParity(block))
{ {
if (!block_buffer.Insert(block)) if (!block_buffer.Insert(block))
@ -2582,7 +2614,7 @@ UINT16 NormStreamObject::ReadSegment(NormBlockId blockId,
//if ((UINT32)offsetDelta < object_size.LSB()) //if ((UINT32)offsetDelta < object_size.LSB())
ASSERT(tx_index.block <= write_index.block); ASSERT(tx_index.block <= write_index.block);
UINT32 blockDelta = write_index.block - tx_index.block; UINT32 blockDelta = write_index.block - tx_index.block;
if (blockDelta < (block_pool.GetTotal() >> 1)) if (blockDelta <= (block_pool.GetTotal() >> 1))
{ {
NormBlock* b = stream_buffer.Find(stream_buffer.RangeLo()); NormBlock* b = stream_buffer.Find(stream_buffer.RangeLo());
if (b && !b->IsPending()) if (b && !b->IsPending())
@ -2666,21 +2698,24 @@ bool NormStreamObject::WriteSegment(NormBlockId blockId,
block->GetFirstPending(read_index.segment); block->GetFirstPending(read_index.segment);
NormBlock* tempBlock = block; NormBlock* tempBlock = block;
UINT32 tempOffset = read_offset; UINT32 tempOffset = read_offset;
NormStreamObject::Index tempIndex = read_index;
// (TBD) uncomment the code so that only a single // (TBD) uncomment the code so that only a single
// UPDATED notification is posted??? // UPDATED notification is posted???
if (notify_on_update) if (notify_on_update)
{ {
notify_on_update = false; notify_on_update = false;
session.Notify(NormController::RX_OBJECT_UPDATED, server, this); session.Notify(NormController::RX_OBJECT_UPDATED, server, this);
} }
block = stream_buffer.Find(stream_buffer.RangeLo()); block = stream_buffer.Find(stream_buffer.RangeLo());
if (tempBlock == block) if (tempBlock == block)
{ {
if (tempOffset == read_offset) if ((tempOffset == read_offset) &&
(tempIndex.block == read_index.block) &&
(tempIndex.segment == read_index.segment))
{ {
// App didn't want any data here, purge segment // App didn't want any data here, purge segment
//ASSERT(0); //ASSERT(0);
dataLost = true; dataLost = true;
block->UnsetPending(read_index.segment++); block->UnsetPending(read_index.segment++);
if (read_index.segment >= ndata) if (read_index.segment >= ndata)
{ {
@ -3185,12 +3220,15 @@ UINT32 NormStreamObject::Write(const char* buffer, UINT32 len, bool eom)
} }
break; break;
} }
// This old code detected buffer "fullness" by offset instead of segment index
// but, the problem there was when apps wrote & flushed messages smaller than
// the segment_size, the buffer used up before this detected it.
//INT32 deltaOffset = write_offset - tx_offset; // (TBD) deprecate tx_offset //INT32 deltaOffset = write_offset - tx_offset; // (TBD) deprecate tx_offset
//ASSERT(deltaOffset >= 0); //ASSERT(deltaOffset >= 0);
//if (deltaOffset >= (INT32)object_size.LSB()) //if (deltaOffset >= (INT32)object_size.LSB())
ASSERT(write_index.block >= tx_index.block); ASSERT(write_index.block >= tx_index.block);
UINT32 deltaBlock = write_index.block - tx_index.block; UINT32 deltaBlock = write_index.block - tx_index.block;
if (deltaBlock >= (block_pool.GetTotal() >> 1)) if (deltaBlock > (block_pool.GetTotal() >> 1))
{ {
write_vacancy = false; write_vacancy = false;
DMSG(8, "NormStreamObject::Write() stream buffer full (1)\n", len, eom); DMSG(8, "NormStreamObject::Write() stream buffer full (1)\n", len, eom);

5
common/normObject.h
View File

@ -167,6 +167,9 @@ class NormObject
NormBlock* ServerRecoverBlock(NormBlockId blockId); NormBlock* ServerRecoverBlock(NormBlockId blockId);
bool CalculateBlockParity(NormBlock* block); bool CalculateBlockParity(NormBlock* block);
/*bool IsFirstPass() {return first_pass;}
void ClearFirstPass() {first_pass = false};*/
bool TxReset(NormBlockId firstBlock = NormBlockId(0), bool requeue = false); bool TxReset(NormBlockId firstBlock = NormBlockId(0), bool requeue = false);
bool TxResetBlocks(NormBlockId nextId, NormBlockId lastId); bool TxResetBlocks(NormBlockId nextId, NormBlockId lastId);
bool TxUpdateBlock(NormBlock* theBlock, bool TxUpdateBlock(NormBlock* theBlock,
@ -265,6 +268,7 @@ class NormObject
// Here are some members used to let us know // Here are some members used to let us know
// our status with respect to the rest of the world // our status with respect to the rest of the world
bool first_pass; // for sender objects
bool accepted; bool accepted;
bool notify_on_update; bool notify_on_update;
@ -390,6 +394,7 @@ class NormStreamObject : public NormObject
SetFlushMode(oldFlushMode); SetFlushMode(oldFlushMode);
} }
void SetPushMode(bool state) {push_mode = state;} void SetPushMode(bool state) {push_mode = state;}
bool GetPushMode() const {return push_mode;}
bool IsClosing() {return stream_closing;} bool IsClosing() {return stream_closing;}
bool HasVacancy() bool HasVacancy()

1
common/normSegment.cpp
View File

@ -243,6 +243,7 @@ bool NormBlock::TxReset(UINT16 numData,
ptr++; ptr++;
} }
erasure_count = 0; erasure_count = 0;
seg_size_max = 0;
} }
} }
return increasedRepair; return increasedRepair;

402
common/normSession.cpp
View File

@ -26,7 +26,7 @@ NormSession::NormSession(NormSessionMgr& sessionMgr, NormNodeId localNodeId)
tx_cache_size_max((UINT32)20*1024*1024), tx_cache_size_max((UINT32)20*1024*1024),
flush_count(NORM_ROBUST_FACTOR+1), flush_count(NORM_ROBUST_FACTOR+1),
posted_tx_queue_empty(false), posted_tx_queue_empty(false),
acking_node_count(0), watermark_pending(false), advertise_repairs(false), acking_node_count(0), watermark_pending(false), tx_repair_pending(false), advertise_repairs(false),
suppress_nonconfirmed(false), suppress_rate(-1.0), suppress_rtt(-1.0), suppress_nonconfirmed(false), suppress_rate(-1.0), suppress_rtt(-1.0),
probe_proactive(true), probe_pending(false), probe_reset(true), probe_data_check(false), probe_proactive(true), probe_pending(false), probe_reset(true), probe_data_check(false),
grtt_interval(0.5), grtt_interval(0.5),
@ -68,7 +68,7 @@ NormSession::NormSession(NormSessionMgr& sessionMgr, NormNodeId localNodeId)
grtt_quantized = NormQuantizeRtt(DEFAULT_GRTT_ESTIMATE); grtt_quantized = NormQuantizeRtt(DEFAULT_GRTT_ESTIMATE);
grtt_measured = grtt_advertised = NormUnquantizeRtt(grtt_quantized); grtt_measured = grtt_advertised = NormUnquantizeRtt(grtt_quantized);
gsize_measured = DEFAULT_GSIZE_ESTIMATE; gsize_measured = DEFAULT_GSIZE_ESTIMATE;
gsize_quantized = NormQuantizeGroupSize(DEFAULT_GSIZE_ESTIMATE); gsize_quantized = NormQuantizeGroupSize(DEFAULT_GSIZE_ESTIMATE);
gsize_advertised = NormUnquantizeGroupSize(gsize_quantized); gsize_advertised = NormUnquantizeGroupSize(gsize_quantized);
@ -249,6 +249,11 @@ bool NormSession::SetMulticastInterface(const char* interfaceName)
void NormSession::SetTxRateInternal(double txRate) void NormSession::SetTxRateInternal(double txRate)
{ {
if (!is_server)
{
tx_rate = txRate;
return;
}
if (txRate < 0.0) if (txRate < 0.0)
{ {
DMSG(0, "NormSession::SetTxRateInternal() invalid transmit rate!\n"); DMSG(0, "NormSession::SetTxRateInternal() invalid transmit rate!\n");
@ -287,6 +292,7 @@ void NormSession::SetTxRateInternal(double txRate)
grtt_quantized = NormQuantizeRtt(grtt_measured); grtt_quantized = NormQuantizeRtt(grtt_measured);
grtt_advertised = NormUnquantizeRtt(grtt_quantized); grtt_advertised = NormUnquantizeRtt(grtt_quantized);
// What do we do when "pktInterval" > "grtt_max"? // What do we do when "pktInterval" > "grtt_max"?
// We will take our lumps with some extra activity timeout NACKs when they happen? // We will take our lumps with some extra activity timeout NACKs when they happen?
if (grtt_advertised > grtt_max) if (grtt_advertised > grtt_max)
@ -294,7 +300,6 @@ void NormSession::SetTxRateInternal(double txRate)
grtt_quantized = NormQuantizeRtt(grtt_max); grtt_quantized = NormQuantizeRtt(grtt_max);
grtt_advertised = NormUnquantizeRtt(grtt_quantized); grtt_advertised = NormUnquantizeRtt(grtt_quantized);
} }
if (grttQuantizedOld != grtt_quantized) if (grttQuantizedOld != grtt_quantized)
{ {
DMSG(4, "NormSession::SetTxRateInternal() node>%lu %s to new grtt to: %lf sec\n", DMSG(4, "NormSession::SetTxRateInternal() node>%lu %s to new grtt to: %lf sec\n",
@ -439,6 +444,10 @@ bool NormSession::StartServer(UINT16 instanceId,
//tx_rate = txRate; // keep grtt at initial //tx_rate = txRate; // keep grtt at initial
SetTxRateInternal(txRate); // adjusts grtt_advertised as needed SetTxRateInternal(txRate); // adjusts grtt_advertised as needed
} }
else
{
SetTxRateInternal(tx_rate); // takes segment size into account, etc on server start
}
cc_slow_start = true; cc_slow_start = true;
cc_active = false; cc_active = false;
@ -451,6 +460,7 @@ bool NormSession::StartServer(UINT16 instanceId,
OnProbeTimeout(probe_timer); OnProbeTimeout(probe_timer);
ActivateTimer(probe_timer); ActivateTimer(probe_timer);
} }
return true; return true;
} // end NormSession::StartServer() } // end NormSession::StartServer()
@ -462,6 +472,11 @@ void NormSession::StopServer()
probe_timer.Deactivate(); probe_timer.Deactivate();
probe_reset = true; probe_reset = true;
} }
if (repair_timer.IsActive())
{
repair_timer.Deactivate();
tx_repair_pending = false;
}
encoder.Destroy(); encoder.Destroy();
acking_node_tree.Destroy(); acking_node_tree.Destroy();
cc_node_list.Destroy(); cc_node_list.Destroy();
@ -525,8 +540,9 @@ void NormSession::Serve()
obj = tx_table.Find(objectId); obj = tx_table.Find(objectId);
ASSERT(obj); ASSERT(obj);
} }
if (watermark_pending) // (TBD) Work on this whole watermark flush check logic!
if (watermark_pending && !flush_timer.IsActive())
{ {
// Determine next message (objectId::blockId::segmentId) to be sent // Determine next message (objectId::blockId::segmentId) to be sent
NormObject* nextObj; NormObject* nextObj;
@ -535,7 +551,7 @@ void NormSession::Serve()
NormSegmentId nextSegmentId = 0; NormSegmentId nextSegmentId = 0;
if (obj) if (obj)
{ {
// Use current transmit pending object // Get index (objectId::blockId::segmentId) of next transmit pending segment
nextObj = obj; nextObj = obj;
nextObjectId = objectId; nextObjectId = objectId;
if (nextObj->IsPending()) if (nextObj->IsPending())
@ -551,9 +567,8 @@ void NormSession::Serve()
block->GetFirstPending(nextSegmentId); block->GetFirstPending(nextSegmentId);
#endif // if/else PROTO_DEBUG #endif // if/else PROTO_DEBUG
// Adjust so watermark segmentId < block length // Adjust so watermark segmentId < block length
if (nextSegmentId >= nextObj->GetBlockSize(nextBlockId)) UINT16 nextBlockSize = nextObj->GetBlockSize(nextBlockId);
nextSegmentId = nextObj->GetBlockSize(nextBlockId) - 1; if (nextSegmentId >= nextBlockSize) nextSegmentId = nextBlockSize - 1;
} }
} }
else else
@ -564,56 +579,77 @@ void NormSession::Serve()
else else
{ {
// Must be an active, but non-pending stream object // Must be an active, but non-pending stream object
ASSERT(nextObj->IsStream());
nextBlockId = static_cast<NormStreamObject*>(nextObj)->GetNextBlockId(); nextBlockId = static_cast<NormStreamObject*>(nextObj)->GetNextBlockId();
nextSegmentId = static_cast<NormStreamObject*>(nextObj)->GetNextSegmentId(); nextSegmentId = static_cast<NormStreamObject*>(nextObj)->GetNextSegmentId();
} }
} }
else
if (tx_repair_pending)
{ {
// Nothing transmit pending, check for repair pending object
if ((tx_repair_object_min < nextObjectId) ||
((tx_repair_object_min == nextObjectId) &&
((tx_repair_block_min < nextBlockId) ||
((tx_repair_block_min == nextBlockId) &&
(tx_repair_segment_min < nextSegmentId)))))
{
nextObjectId = tx_repair_object_min;
nextBlockId = tx_repair_block_min;
nextSegmentId = tx_repair_segment_min;
DMSG(8, "watermark>%hu:%lu:%hu check against repair index>%hu:%lu:%hu\n",
(UINT16)watermark_object_id, (UINT32)watermark_block_id, (UINT16)watermark_segment_id,
(UINT16)nextObjectId, (UINT32)nextBlockId, (UINT16)nextSegmentId);
}
/*
// Get index (objectId::blockId::segmentId) of next repair pending segment
// (This seems like alot of work to do for each call to NormSession::Serve(). Yuk!
// (well when a watermark is pending ... which could be alot of the time
// i.e. iterating through the tx_table, looking for repair pending objects ...
// (TBD) Is there a better way such as keeping state elsewhere on minimum
// pending repairs to save doing the work here?jo
NormObjectId repairObjectId = next_tx_object_id;
NormBlockId repairBlockId = 0;
NormSegmentId repairSegmentId = 0;
if (!ServerGetFirstRepairPending(repairObjectId))
repairObjectId = next_tx_object_id;
NormObjectTable::Iterator iterator(tx_table); NormObjectTable::Iterator iterator(tx_table);
if (ServerGetFirstRepairPending(nextObjectId)) while ((nextObj = iterator.GetNextObject()))
{ {
while ((nextObj = iterator.GetNextObject())) if (repairObjectId < nextObj->GetId())
{ {
if (nextObjectId < nextObj->GetId()) nextObj = tx_table.Find(repairObjectId);
{ break;
nextObj = tx_table.Find(nextObjectId); }
break; else if (nextObj->IsRepairPending())
}
else if (nextObj->IsRepairPending())
{
nextObjectId = nextObj->GetId();
break;
}
}
}
else
{
nextObjectId = next_tx_object_id;
while ((nextObj = iterator.GetNextObject()))
{ {
if (nextObj->IsRepairPending()) repairObjectId = nextObj->GetId();
{ break;
nextObjectId = nextObj->GetId(); }
break; }
} if (nextObj) nextObj->FindRepairIndex(repairBlockId, repairSegmentId);
}
} // Get min of next transmit pending or repair pending segment
if (nextObj) if ((repairObjectId < nextObjectId) ||
((repairObjectId == nextObjectId) &&
((repairBlockId < nextBlockId) ||
((repairBlockId == nextBlockId) &&
(repairSegmentId < nextSegmentId)))))
{ {
#ifdef PROTO_DEBUG nextObjectId = repairObjectId;
ASSERT(nextObj->FindRepairIndex(nextBlockId, nextSegmentId)); nextBlockId = repairBlockId;
#else nextSegmentId = repairSegmentId;
nextObj->FindRepairIndex(nextBlockId, nextSegmentId);
#endif
} }
} */
} // end if (tx_repair_pending)
if ((nextObjectId > watermark_object_id) || if ((nextObjectId > watermark_object_id) ||
((nextObjectId == watermark_object_id) && ((nextObjectId == watermark_object_id) &&
((nextBlockId > watermark_block_id) || ((nextBlockId > watermark_block_id) ||
(((nextBlockId == watermark_block_id) && ((nextBlockId == watermark_block_id) &&
(nextSegmentId > watermark_segment_id)))))) (nextSegmentId > watermark_segment_id)))))
{ {
if (ServerQueueWatermarkFlush()) if (ServerQueueWatermarkFlush())
{ {
@ -637,8 +673,7 @@ void NormSession::Serve()
NormAckingNode* next; NormAckingNode* next;
while ((next = static_cast<NormAckingNode*>(iterator.GetNextNode()))) while ((next = static_cast<NormAckingNode*>(iterator.GetNextNode())))
{ {
//if (next->IsPending()) next->ResetReqCount();
next->ResetReqCount();
} }
} }
} }
@ -674,21 +709,15 @@ void NormSession::Serve()
msg->SetGroupSize(gsize_quantized); msg->SetGroupSize(gsize_quantized);
QueueMessage(msg); QueueMessage(msg);
flush_count = 0; flush_count = 0;
//if (flush_timer.IsActive()) flush_timer.Deactivate(); // (TBD) ??? should streams every allowed to be non-pending?
if (!obj->IsPending()) // we _could_ re-architect streams a little bit and allow
{ // for this by having NormStreamObject::Write() control
if (obj->IsStream()) // stream advancement ... I think it would be cleaner.
posted_tx_queue_empty = true; // repair-delayed stream advance // (mod NormStreamObject::StreamAdvance() to depend upon
else // what has been written and conversely set some pending
tx_pending_mask.Unset(obj->GetId()); // state as calls to NormStreamObject::Write() are made.
// (TBD) post TX_OBJECT_SENT notification if (!obj->IsPending() && !obj->IsStream())
tx_pending_mask.Unset(obj->GetId());
}
else if (obj->IsStream())
{
// Is there room write to the stream
// should we post TX_QUEUE_VACANCY here?
}
} }
else else
{ {
@ -720,7 +749,7 @@ void NormSession::Serve()
} }
} }
} }
if (!posted_tx_queue_empty && !stream->IsClosing()) if (!posted_tx_queue_empty && stream->IsPending() && !stream->IsClosing())
{ {
//data_active = false; //data_active = false;
posted_tx_queue_empty = true; posted_tx_queue_empty = true;
@ -758,7 +787,11 @@ void NormSession::Serve()
if (flush_count < NORM_ROBUST_FACTOR) if (flush_count < NORM_ROBUST_FACTOR)
{ {
// Queue flush message // Queue flush message
ServerQueueFlush(); if (!tx_repair_pending) // don't queue flush if repair pending
ServerQueueFlush();
else
DMSG(8, "NormSession::Serve() node>%lu NORM_CMD(FLUSH) deferred by pending repairs ...\n",
LocalNodeId());
} }
else if (flush_count == NORM_ROBUST_FACTOR) else if (flush_count == NORM_ROBUST_FACTOR)
{ {
@ -878,6 +911,16 @@ bool NormSession::ServerQueueWatermarkFlush()
flush->SetGroupSize(gsize_quantized); flush->SetGroupSize(gsize_quantized);
flush->SetObjectId(watermark_object_id); flush->SetObjectId(watermark_object_id);
flush->SetFecBlockId(watermark_block_id); flush->SetFecBlockId(watermark_block_id);
// _Attempt_ to set the fec_payload_id source block length field appropriately
UINT16 blockLen;
NormObject* obj = tx_table.Find(watermark_object_id);
if (NULL != obj)
blockLen = obj->GetBlockSize(watermark_block_id);
else if (watermark_segment_id < ndata)
blockLen = ndata;
else
blockLen = watermark_segment_id;
flush->SetFecBlockLen(blockLen);
flush->SetFecSymbolId(watermark_segment_id); flush->SetFecSymbolId(watermark_segment_id);
NormNodeTreeIterator iterator(acking_node_tree); NormNodeTreeIterator iterator(acking_node_tree);
NormAckingNode* next; NormAckingNode* next;
@ -960,7 +1003,7 @@ bool NormSession::ServerQueueWatermarkFlush()
void NormSession::ServerQueueFlush() void NormSession::ServerQueueFlush()
{ {
// (TBD) Deal with EOT or pre-queued squelch on squelch case // (TBD) Don't enqueue a new flush if there is already one in our tx_queue!
if (flush_timer.IsActive()) return; if (flush_timer.IsActive()) return;
NormObject* obj = tx_table.Find(tx_table.RangeHi()); NormObject* obj = tx_table.Find(tx_table.RangeHi());
NormObjectId objectId; NormObjectId objectId;
@ -981,17 +1024,43 @@ void NormSession::ServerQueueFlush()
blockId = obj->GetFinalBlockId(); blockId = obj->GetFinalBlockId();
segmentId = obj->GetBlockSize(blockId) - 1; segmentId = obj->GetBlockSize(blockId) - 1;
} }
NormCmdFlushMsg* flush = (NormCmdFlushMsg*)GetMessageFromPool();
if (flush)
{
flush->Init();
flush->SetDestination(address);
flush->SetGrtt(grtt_quantized);
flush->SetBackoffFactor((unsigned char)backoff_factor);
flush->SetGroupSize(gsize_quantized);
flush->SetObjectId(objectId);
flush->SetFecBlockId(blockId);
flush->SetFecBlockLen(obj->GetBlockSize(blockId));
flush->SetFecSymbolId(segmentId);
QueueMessage(flush);
if (flush_count < NORM_ROBUST_FACTOR) flush_count++;
DMSG(4, "NormSession::ServerQueueFlush() node>%lu, flush queued (flush_count:%u)...\n",
LocalNodeId(), flush_count);
}
else
{
DMSG(0, "NormSession::ServerQueueFlush() node>%lu message_pool exhausted! (couldn't flush)\n",
LocalNodeId());
}
} }
else else
{ {
// Why did I do this? - Brian // Because a squelch keeps the receivers from NACKing in futility // Why did I do this? - Brian // Because a squelch keeps the receivers from NACKing in futility
// (TBD) send NORM_CMD(EOT) instead? - no // (TBD) send NORM_CMD(EOT) instead? - no
// Perhaps I should send a flush anyway w/ (next_tx_object_id - 1) and squelch accordingly?
// This condition shouldn't occur if we have state on the most recent object ... we should
// unless the app does bad things like "cancel" all of its tx objects ...
// Maybe we shouldn't send anything if we have no pending tx objects? No need to flush, etc
// if all tx object state is gone ...
if (ServerQueueSquelch(next_tx_object_id)) if (ServerQueueSquelch(next_tx_object_id))
{ {
if (flush_count < NORM_ROBUST_FACTOR) flush_count++; if (flush_count < NORM_ROBUST_FACTOR) flush_count++;
flush_timer.SetInterval(4*grtt_advertised); DMSG(4, "NormSession::ServerQueueFlush() node>%lu squelch queued (flush_count:%u)...\n",
ActivateTimer(flush_timer);
DMSG(8, "NormSession::ServerQueueFlush() node>%lu squelch queued (flush_count:%u)...\n",
LocalNodeId(), flush_count); LocalNodeId(), flush_count);
} }
else else
@ -999,30 +1068,8 @@ void NormSession::ServerQueueFlush()
DMSG(0, "NormSession::ServerQueueFlush() warning: node>%lu unable to queue squelch\n", DMSG(0, "NormSession::ServerQueueFlush() warning: node>%lu unable to queue squelch\n",
LocalNodeId()); LocalNodeId());
} }
return;
} }
NormCmdFlushMsg* flush = (NormCmdFlushMsg*)GetMessageFromPool(); flush_timer.SetInterval(2*grtt_advertised);
if (flush)
{
flush->Init();
flush->SetDestination(address);
flush->SetGrtt(grtt_quantized);
flush->SetBackoffFactor((unsigned char)backoff_factor);
flush->SetGroupSize(gsize_quantized);
flush->SetObjectId(objectId);
flush->SetFecBlockId(blockId);
flush->SetFecSymbolId(segmentId);
QueueMessage(flush);
if (flush_count < NORM_ROBUST_FACTOR) flush_count++;
DMSG(4, "NormSession::ServerQueueFlush() node>%lu, flush queued (flush_count:%u)...\n",
LocalNodeId(), flush_count);
}
else
{
DMSG(0, "NormSession::ServerQueueFlush() node>%lu message_pool exhausted! (couldn't flush)\n",
LocalNodeId());
}
flush_timer.SetInterval(4*grtt_advertised);
ActivateTimer(flush_timer); ActivateTimer(flush_timer);
} // end NormSession::ServerQueueFlush() } // end NormSession::ServerQueueFlush()
@ -2098,6 +2145,8 @@ void NormSession::ServerHandleAckMessage(const struct timeval& currentTime, cons
} }
else else
{ {
// This can happen when new watermarks are set when an old watermark is still
// pending.
DMSG(0, "NormSession::ServerHandleAckMessage() received wrong watermark ACK?!\n"); DMSG(0, "NormSession::ServerHandleAckMessage() received wrong watermark ACK?!\n");
} }
} }
@ -2168,6 +2217,7 @@ void NormSession::ServerHandleNackMessage(const struct timeval& currentTime, Nor
bool squelchQueued = false; bool squelchQueued = false;
// Get the index of our next pending NORM_DATA transmission
NormObjectId txObjectIndex; NormObjectId txObjectIndex;
NormBlockId txBlockIndex; NormBlockId txBlockIndex;
if (ServerGetFirstPending(txObjectIndex)) if (ServerGetFirstPending(txObjectIndex))
@ -2193,7 +2243,7 @@ void NormSession::ServerHandleNackMessage(const struct timeval& currentTime, Nor
txObjectIndex = next_tx_object_id; txObjectIndex = next_tx_object_id;
txBlockIndex = 0; txBlockIndex = 0;
} }
bool holdoff = (repair_timer.IsActive() && !repair_timer.GetRepeatCount()); bool holdoff = (repair_timer.IsActive() && !repair_timer.GetRepeatCount());
enum NormRequestLevel {SEGMENT, BLOCK, INFO, OBJECT}; enum NormRequestLevel {SEGMENT, BLOCK, INFO, OBJECT};
while ((requestLength = nack.UnpackRepairRequest(req, requestOffset))) while ((requestLength = nack.UnpackRepairRequest(req, requestOffset)))
@ -2202,15 +2252,26 @@ void NormSession::ServerHandleNackMessage(const struct timeval& currentTime, Nor
requestOffset += requestLength; requestOffset += requestLength;
NormRequestLevel requestLevel; NormRequestLevel requestLevel;
if (req.FlagIsSet(NormRepairRequest::SEGMENT)) if (req.FlagIsSet(NormRepairRequest::SEGMENT))
{
requestLevel = SEGMENT; requestLevel = SEGMENT;
}
else if (req.FlagIsSet(NormRepairRequest::BLOCK)) else if (req.FlagIsSet(NormRepairRequest::BLOCK))
{
requestLevel = BLOCK; requestLevel = BLOCK;
}
else if (req.FlagIsSet(NormRepairRequest::OBJECT)) else if (req.FlagIsSet(NormRepairRequest::OBJECT))
{
requestLevel = OBJECT; requestLevel = OBJECT;
else }
else if (req.FlagIsSet(NormRepairRequest::INFO))
{ {
requestLevel = INFO; requestLevel = INFO;
ASSERT(req.FlagIsSet(NormRepairRequest::INFO)); }
else
{
DMSG(0, "NormSession::ServerHandleNackMessage() node>%lu recvd repair request w/ invalid repair level\n",
LocalNodeId());
continue;
} }
NormRepairRequest::Iterator iterator(req); NormRepairRequest::Iterator iterator(req);
@ -2280,6 +2341,23 @@ void NormSession::ServerHandleNackMessage(const struct timeval& currentTime, Nor
} }
else else
{ {
// Update our minimum tx repair index as needed
if (tx_repair_pending)
{
if (nextObjectId <= tx_repair_object_min)
{
tx_repair_object_min = nextObjectId;
tx_repair_block_min = 0;
tx_repair_segment_min = 0;
}
}
else
{
tx_repair_pending = true;
tx_repair_object_min = nextObjectId;
tx_repair_block_min = 0;
tx_repair_segment_min = 0;
}
object->HandleInfoRequest(); object->HandleInfoRequest();
startTimer = true; startTimer = true;
} }
@ -2307,6 +2385,23 @@ void NormSession::ServerHandleNackMessage(const struct timeval& currentTime, Nor
} }
else else
{ {
// Update our minimum tx repair index as needed
if (tx_repair_pending)
{
if (nextObjectId <= tx_repair_object_min)
{
tx_repair_object_min = nextObjectId;
tx_repair_block_min = 0;
tx_repair_segment_min = 0;
}
}
else
{
tx_repair_pending = true;
tx_repair_object_min = nextObjectId;
tx_repair_block_min = 0;
tx_repair_segment_min = 0;
}
tx_repair_mask.Set(nextObjectId); tx_repair_mask.Set(nextObjectId);
startTimer = true; startTimer = true;
} }
@ -2382,6 +2477,31 @@ void NormSession::ServerHandleNackMessage(const struct timeval& currentTime, Nor
} }
else else
{ {
// Update our minimum tx repair index as needed
if (tx_repair_pending)
{
if (nextObjectId < tx_repair_object_min)
{
tx_repair_object_min = nextObjectId;
tx_repair_block_min = nextBlockId;
tx_repair_segment_min = 0;
}
else if (nextObjectId == tx_repair_object_min)
{
if (nextBlockId <= tx_repair_block_min)
{
tx_repair_block_min = nextBlockId;
tx_repair_segment_min = 0;
}
}
}
else
{
tx_repair_pending = true;
tx_repair_object_min = nextObjectId;
tx_repair_block_min = nextBlockId;
tx_repair_segment_min = 0;
}
object->HandleBlockRequest(nextBlockId, lastBlockId); object->HandleBlockRequest(nextBlockId, lastBlockId);
startTimer = true; startTimer = true;
} }
@ -2394,17 +2514,23 @@ void NormSession::ServerHandleNackMessage(const struct timeval& currentTime, Nor
if (nextBlockId != prevBlockId) freshBlock = true; if (nextBlockId != prevBlockId) freshBlock = true;
if (freshBlock) if (freshBlock)
{ {
freshBlock = false;
// Is this entire block already repair pending? // Is this entire block already repair pending?
if (object->IsRepairSet(nextBlockId)) if (object->IsRepairSet(nextBlockId))
break; continue;
if (!(block = object->FindBlock(nextBlockId))) if (NULL == (block = object->FindBlock(nextBlockId)))
{ {
// Is this entire block already tx pending? // Is this entire block already tx pending?
if (!object->IsPendingSet(nextBlockId)) if (object->IsPendingSet(nextBlockId))
{
// Entire block already tx pending, don't worry about individual segments
DMSG(4, "NormSession::ServerHandleNackMessage() node>%lu "
"recvd SEGMENT repair request for pending block.\n");
continue;
}
else
{ {
// Try to recover block including parity calculation // Try to recover block including parity calculation
if (!(block = object->ServerRecoverBlock(nextBlockId))) if (NULL == (block = object->ServerRecoverBlock(nextBlockId)))
{ {
if (NormObject::STREAM == object->GetType()) if (NormObject::STREAM == object->GetType())
{ {
@ -2419,26 +2545,20 @@ void NormSession::ServerHandleNackMessage(const struct timeval& currentTime, Nor
} }
else else
{ {
// Resource constrained, move on. // Resource constrained, move on to next repair request
DMSG(2, "NormSession::ServerHandleNackMessage() node>%lu " DMSG(2, "NormSession::ServerHandleNackMessage() node>%lu "
"Warning - server is resource contrained ...\n"); "Warning - server is resource contrained ...\n");
} }
break; continue;
} }
} }
else
{
// Entire block already tx pending, don't recover
DMSG(4, "NormSession::ServerHandleNackMessage() node>%lu "
"recvd SEGMENT repair request for pending block.\n");
break;
}
} }
freshBlock = false;
numErasures = extra_parity; numErasures = extra_parity;
prevBlockId = nextBlockId; prevBlockId = nextBlockId;
} } // end if (freshBlock)
ASSERT(NULL != block);
// If stream && explicit data repair, lock the data for retransmission // If stream && explicit data repair, lock the data for retransmission
if (object->IsStream() && (nextSegmentId < ndata)) if (object->IsStream() && (nextSegmentId < ndata))
{ {
bool attemptLock = true; bool attemptLock = true;
@ -2541,16 +2661,50 @@ void NormSession::ServerHandleNackMessage(const struct timeval& currentTime, Nor
} }
else else
{ {
// Update our minimum tx repair index as needed
ASSERT(nextBlockId == block->GetId());
UINT16 nextBlockSize = object->GetBlockSize(nextBlockId);
if (tx_repair_pending)
{
if (nextObjectId < tx_repair_object_min)
{
tx_repair_block_min = nextBlockId;
tx_repair_segment_min = (nextSegmentId < nextBlockSize) ?
nextSegmentId : (nextBlockSize - 1);
}
else if (nextObjectId == tx_repair_object_min)
{
if (nextBlockId < tx_repair_block_min)
{
tx_repair_block_min = nextBlockId;
tx_repair_segment_min = (nextSegmentId < nextBlockSize) ?
nextSegmentId : (nextBlockSize - 1);
}
else if (nextBlockId == tx_repair_block_min)
{
if (nextSegmentId < tx_repair_segment_min)
tx_repair_segment_min = nextSegmentId;
}
}
}
else
{
tx_repair_pending = true;
tx_repair_object_min = nextObjectId;
tx_repair_block_min = nextBlockId;
tx_repair_segment_min = (nextSegmentId < nextBlockSize) ?
nextSegmentId : (nextBlockSize - 1);
}
block->HandleSegmentRequest(nextSegmentId, lastSegmentId, block->HandleSegmentRequest(nextSegmentId, lastSegmentId,
object->GetBlockSize(block->GetId()), nextBlockSize, nparity,
nparity, numErasures); numErasures);
startTimer = true; startTimer = true;
} // end if/else (holdoff) } // end if/else (holdoff)
break; break;
case INFO: case INFO:
// We already dealt with INFO request above with respect to initiating repair
nextObjectId++; nextObjectId++;
if (nextObjectId > lastObjectId) if (nextObjectId > lastObjectId) inRange = false;
inRange = false;
break; break;
} // end switch(requestLevel) } // end switch(requestLevel)
} // end while(inRange) } // end while(inRange)
@ -2649,7 +2803,9 @@ bool NormSession::ServerQueueSquelch(NormObjectId objectId)
{ {
ASSERT(NormObject::STREAM == obj->GetType()); ASSERT(NormObject::STREAM == obj->GetType());
squelch->SetObjectId(objectId); squelch->SetObjectId(objectId);
squelch->SetFecBlockId(((NormStreamObject*)obj)->StreamBufferLo()); NormBlockId blockId = static_cast<NormStreamObject*>(obj)->StreamBufferLo();
squelch->SetFecBlockId(blockId);
squelch->SetFecBlockLen(obj->GetBlockSize(blockId));
squelch->SetFecSymbolId(0); squelch->SetFecSymbolId(0);
squelch->ResetInvalidObjectList(); squelch->ResetInvalidObjectList();
while ((obj = iterator.GetNextObject())) while ((obj = iterator.GetNextObject()))
@ -2662,10 +2818,13 @@ bool NormSession::ServerQueueSquelch(NormObjectId objectId)
if (obj) if (obj)
{ {
squelch->SetObjectId(obj->GetId()); squelch->SetObjectId(obj->GetId());
NormBlockId blockId;
if (obj->IsStream()) if (obj->IsStream())
squelch->SetFecBlockId(((NormStreamObject*)obj)->StreamBufferLo()); blockId =static_cast<NormStreamObject*>(obj)->StreamBufferLo();
else else
squelch->SetFecBlockId(0); blockId = NormBlockId(0);
squelch->SetFecBlockId(blockId);
squelch->SetFecBlockLen(obj->GetBlockSize(blockId));
squelch->SetFecSymbolId(0); squelch->SetFecSymbolId(0);
nextId = obj->GetId() + 1; nextId = obj->GetId() + 1;
} }
@ -2674,6 +2833,7 @@ bool NormSession::ServerQueueSquelch(NormObjectId objectId)
// Squelch to point to future object // Squelch to point to future object
squelch->SetObjectId(next_tx_object_id); squelch->SetObjectId(next_tx_object_id);
squelch->SetFecBlockId(0); squelch->SetFecBlockId(0);
squelch->SetFecBlockLen(0); // (TBD) should this be "ndata" instead? but we can't be sure
squelch->SetFecSymbolId(0); squelch->SetFecSymbolId(0);
nextId = next_tx_object_id; nextId = next_tx_object_id;
} }
@ -2821,6 +2981,7 @@ bool NormSession::ServerBuildRepairAdv(NormCmdRepairAdvMsg& cmd)
bool NormSession::OnRepairTimeout(ProtoTimer& /*theTimer*/) bool NormSession::OnRepairTimeout(ProtoTimer& /*theTimer*/)
{ {
tx_repair_pending = false;
if (repair_timer.GetRepeatCount()) if (repair_timer.GetRepeatCount())
{ {
// NACK aggregation period has ended. (incorporate accumulated repair requests) // NACK aggregation period has ended. (incorporate accumulated repair requests)
@ -2859,9 +3020,7 @@ bool NormSession::OnRepairTimeout(ProtoTimer& /*theTimer*/)
} }
} }
} // end while (iterator.GetNextObject()) } // end while (iterator.GetNextObject())
// (TBD) should this be just PromptServer() instead??? PromptServer();
// (note we do set posted_tx_queue_empty when repair state blocks stream progress)
TouchServer();
// BACKOFF related code // BACKOFF related code
// Holdoff initiation of new repair cycle for one GRTT // Holdoff initiation of new repair cycle for one GRTT
// (TBD) for unicast sessions, use CLR RTT ??? // (TBD) for unicast sessions, use CLR RTT ???
@ -3053,8 +3212,7 @@ bool NormSession::SendMessage(NormMsg& msg)
} }
else else
{ {
if (tx_socket->SendTo(msg.GetBuffer(), msgSize, if (tx_socket->SendTo(msg.GetBuffer(), msgSize, msg.GetDestination()))
msg.GetDestination()))
{ {
// Separate send/recv tracing // Separate send/recv tracing
if (trace) if (trace)

6
common/normSession.h
View File

@ -162,7 +162,7 @@ class NormSession
void SetTxRate(double txRate) void SetTxRate(double txRate)
{ {
txRate /= 8.0; // convert to bytes/sec txRate /= 8.0; // convert to bytes/sec
SetTxRateInternal(txRate); SetTxRateInternal(txRate);
} }
double BackoffFactor() {return backoff_factor;} double BackoffFactor() {return backoff_factor;}
void SetBackoffFactor(double value) {backoff_factor = value;} void SetBackoffFactor(double value) {backoff_factor = value;}
@ -457,6 +457,10 @@ class NormSession
NormObjectId watermark_object_id; NormObjectId watermark_object_id;
NormBlockId watermark_block_id; NormBlockId watermark_block_id;
NormSegmentId watermark_segment_id; NormSegmentId watermark_segment_id;
bool tx_repair_pending;
NormObjectId tx_repair_object_min;
NormBlockId tx_repair_block_min;
NormSegmentId tx_repair_segment_min;
// for unicast nack/cc feedback suppression // for unicast nack/cc feedback suppression
bool advertise_repairs; bool advertise_repairs;

37
common/normTest.cpp
View File

@ -68,6 +68,9 @@ int main(int argc, char* argv[])
//NormSetTOS(session, 0x20); //NormSetTOS(session, 0x20);
// NOTE: These are debugging routines available (not for normal app use) // NOTE: These are debugging routines available (not for normal app use)
// (IMPORTANT NOTE: On Win32 builds with Norm.dll, this "SetDebugLevel()" has no
// effect on the Protolib instance in the NORM DLL !!!
// (TBD - provide a "NormSetDebugLevel()" function for this purpose!)
SetDebugLevel(2); SetDebugLevel(2);
// Uncomment to turn on debug NORM message tracing // Uncomment to turn on debug NORM message tracing
//NormSetMessageTrace(session, true); //NormSetMessageTrace(session, true);
@ -82,10 +85,10 @@ int main(int argc, char* argv[])
NormSetGrttEstimate(session, 0.001); // 1 msec initial grtt NormSetGrttEstimate(session, 0.001); // 1 msec initial grtt
NormSetTransmitRate(session, 1.0e+06); // in bits/second NormSetTransmitRate(session, 1.0e+05); // in bits/second
// Uncomment to enable TCP-friendly congestion control (overrides NormSetTransmitRate()) // Uncomment to enable TCP-friendly congestion control (overrides NormSetTransmitRate())
NormSetCongestionControl(session, true); //NormSetCongestionControl(session, true);
//NormSetTransmitRateBounds(session, 1000.0, -1.0); //NormSetTransmitRateBounds(session, 1000.0, -1.0);
@ -96,20 +99,22 @@ int main(int argc, char* argv[])
// is _not_ recommended when unicast feedback may be // is _not_ recommended when unicast feedback may be
// possible!) // possible!)
//NormSetTxPort(session, 6001); //NormSetTxPort(session, 6001);
//NormSetDefaultUnicastNack(session, true);
// Uncomment to allow reuse of rx port // Uncomment to allow reuse of rx port
// (This allows multiple "normTest" instances on the same machine // (This allows multiple "normTest" instances on the same machine
// for the same NormSession - note those instances must use // for the same NormSession - note those instances must use
// unique local NormNodeIds (see NormCreateSession() above). // unique local NormNodeIds (see NormCreateSession() above).
//NormSetRxPortReuse(session, true); NormSetRxPortReuse(session, true);
// Uncomment to receive your own traffic // Uncomment to receive your own traffic
//NormSetLoopback(session, true); NormSetLoopback(session, true);
//NormSetSilentReceiver(session, true); //NormSetSilentReceiver(session, true);
// Uncomment this line to participate as a receiver // Uncomment this line to participate as a receiver
//NormStartReceiver(session, 1024*1024); NormStartReceiver(session, 1024*1024);
// Uncomment to set large rx socket buffer size // Uncomment to set large rx socket buffer size
// (might be needed for high rate sessions) // (might be needed for high rate sessions)
@ -119,25 +124,29 @@ int main(int argc, char* argv[])
NormSessionId sessionId = (NormSessionId)rand(); NormSessionId sessionId = (NormSessionId)rand();
// Uncomment the following line to start sender // Uncomment the following line to start sender
NormStartSender(session, sessionId, 1024*1024, 1400, 64, 8); //NormStartSender(session, sessionId, 1024*1024, 1400, 64, 8);
//NormSetAutoParity(session, 6); //NormSetAutoParity(session, 6);
// Uncomment to set large tx socket buffer size // Uncomment to set large tx socket buffer size
// (might be needed for high rate sessions) // (might be needed for high rate sessions)
NormSetTxSocketBuffer(session, 512000); //NormSetTxSocketBuffer(session, 512000);
NormAddAckingNode(session, NORM_NODE_NONE); //15); //NormGetLocalNodeId(session)); NormAddAckingNode(session, NORM_NODE_NONE); //15); //NormGetLocalNodeId(session));
NormObjectHandle stream = NORM_OBJECT_INVALID; NormObjectHandle stream = NORM_OBJECT_INVALID;
#ifdef WIN32
const char* filePath = "C:\\Adamson\\Images\\Art\\giger205.jpg";
#else
const char* filePath = "/home/adamson/images/art/giger/giger205.jpg"; const char* filePath = "/home/adamson/images/art/giger/giger205.jpg";
#endif
//const char* filePath = "/home/adamson/pkgs/rh73.tgz"; //const char* filePath = "/home/adamson/pkgs/rh73.tgz";
const char* fileName = "file1.jpg"; const char* fileName = "file1.jpg";
const char* fileName2 = "file2.jpg"; const char* fileName2 = "file2.jpg";
// Uncomment this line to send a stream instead of the file // Uncomment this line to send a stream instead of the file
stream = NormStreamOpen(session, 4*1024*1024); //stream = NormStreamOpen(session, 4*1024*1024);
// NORM_FLUSH_PASSIVE automatically flushes full writes to // NORM_FLUSH_PASSIVE automatically flushes full writes to
// the stream. // the stream.
@ -261,14 +270,6 @@ int main(int argc, char* argv[])
case NORM_TX_OBJECT_PURGED: case NORM_TX_OBJECT_PURGED:
DMSG(2, "normTest: NORM_TX_OBJECT_PURGED event ...\n"); DMSG(2, "normTest: NORM_TX_OBJECT_PURGED event ...\n");
break; break;
case NORM_CC_ACTIVE:
DMSG(2, "normTest: NORM_CC_ACTIVE event ...\n");
break;
case NORM_CC_INACTIVE:
DMSG(2, "normTest: NORM_CC_INACTIVE event ...\n");
break;
case NORM_RX_OBJECT_NEW: case NORM_RX_OBJECT_NEW:
DMSG(3, "normTest: NORM_RX_OBJECT_NEW event ...\n"); DMSG(3, "normTest: NORM_RX_OBJECT_NEW event ...\n");
@ -344,7 +345,7 @@ int main(int argc, char* argv[])
//else //else
// TRACE("validated recv msg len:%d\n", len); // TRACE("validated recv msg len:%d\n", len);
recvCount = value+1; recvCount = value+1;
if (0 == msgCount % 100) if (0 == msgCount % 1000)
{ {
TRACE("normTest: status> msgCount:%d of total:%d (%lf)\n", TRACE("normTest: status> msgCount:%d of total:%d (%lf)\n",
msgCount, recvCount, 100.0*((double)msgCount)/((double)recvCount)); msgCount, recvCount, 100.0*((double)msgCount)/((double)recvCount));

2
common/normVersion.h
View File

@ -32,6 +32,6 @@
#ifndef _NORM_VERSION #ifndef _NORM_VERSION
#define _NORM_VERSION #define _NORM_VERSION
#define VERSION "1.3b8" #define VERSION "1.3b9"
#endif // _NORM_VERSION #endif // _NORM_VERSION

12
examples/normFileRecv.cpp
View File

@ -144,16 +144,16 @@ int main(int argc, char* argv[])
fprintf(stderr, "normFileRecv: NORM_RX_OBJECT_ABORTED event ...\n"); fprintf(stderr, "normFileRecv: NORM_RX_OBJECT_ABORTED event ...\n");
break; break;
case NORM_REMOTE_SERVER_NEW: case NORM_REMOTE_SENDER_NEW:
fprintf(stderr, "normFileRecv: NORM_REMOTE_SERVER_NEW event ...\n"); fprintf(stderr, "normFileRecv: NORM_REMOTE_SENDER_NEW event ...\n");
break; break;
case NORM_REMOTE_SERVER_ACTIVE: case NORM_REMOTE_SENDER_ACTIVE:
fprintf(stderr, "normFileRecv: NORM_REMOTE_SERVER_ACTIVE event ...\n"); fprintf(stderr, "normFileRecv: NORM_REMOTE_SENDER_ACTIVE event ...\n");
break; break;
case NORM_REMOTE_SERVER_INACTIVE: case NORM_REMOTE_SENDER_INACTIVE:
fprintf(stderr, "normFileRecv: NORM_REMOTE_SERVER_INACTIVE event ...\n"); fprintf(stderr, "normFileRecv: NORM_REMOTE_SENDER_INACTIVE event ...\n");
break; break;
default: default:

2
examples/normFileSend.cpp
View File

@ -73,7 +73,7 @@ int main(int argc, char* argv[])
// Uncomment to turn on debug NORM message tracing // Uncomment to turn on debug NORM message tracing
NormSetMessageTrace(session, true); NormSetMessageTrace(session, true);
// Uncomment to turn on some random packet loss // Uncomment to turn on some random packet loss
// NormSetTxLoss(session, 10.0); // 10% packet loss //NormSetTxLoss(session, 10.0); // 10% packet loss
struct timeval currentTime; struct timeval currentTime;
ProtoSystemTime(currentTime); ProtoSystemTime(currentTime);
// Uncomment to get different packet loss patterns from run to run // Uncomment to get different packet loss patterns from run to run

BIN
win32/Norm.suo
View File

Binary file not shown.