From 276f7c4e20ee6556bd0d427a33c59fd146b4059a Mon Sep 17 00:00:00 2001 From: Jeff Weston Date: Wed, 11 Sep 2019 11:55:26 -0400 Subject: [PATCH] v1.3b9 --- TODO.TXT | 13 ++ VERSION.TXT | 30 +++ common/normApi.cpp | 6 +- common/normApi.h | 17 ++ common/normApp.cpp | 143 +++++++++++++- common/normEncoder.cpp | 32 +-- common/normMessage.h | 42 ++-- common/normNode.cpp | 209 +++++++++----------- common/normNode.h | 7 +- common/normObject.cpp | 80 ++++++-- common/normObject.h | 5 + common/normSegment.cpp | 1 + common/normSession.cpp | 402 ++++++++++++++++++++++++++------------ common/normSession.h | 6 +- common/normTest.cpp | 37 ++-- common/normVersion.h | 2 +- examples/normFileRecv.cpp | 12 +- examples/normFileSend.cpp | 2 +- win32/Norm.suo | Bin 20992 -> 20992 bytes 19 files changed, 713 insertions(+), 333 deletions(-) diff --git a/TODO.TXT b/TODO.TXT index 42bf50f..1a2cd31 100644 --- a/TODO.TXT +++ b/TODO.TXT @@ -44,7 +44,20 @@ 14) IMPORANT: double check ACKing in reponse to watermark polling and relation to cc feedback suppression, etc ... watermark acks should 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) + diff --git a/VERSION.TXT b/VERSION.TXT index 0200b3a..8be7276 100644 --- a/VERSION.TXT +++ b/VERSION.TXT @@ -8,6 +8,36 @@ Version 1.3b9 message erroneously dereferenced a NormObject pointer as NormFileObject (led to run-time issue in optimized code build) (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 ============= diff --git a/common/normApi.cpp b/common/normApi.cpp index 1963396..713f265 100644 --- a/common/normApi.cpp +++ b/common/normApi.cpp @@ -459,7 +459,7 @@ bool NormInstance::GetNextEvent(NormEvent* theEvent) // "Release" any previously-retained object or node handle if (NORM_OBJECT_INVALID != previous_notification->event.object) ((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(); notify_pool.Append(previous_notification); previous_notification = NULL; @@ -600,7 +600,7 @@ void NormInstance::Shutdown() // Release any previously-retained object or node handles if (NORM_OBJECT_INVALID != previous_notification->event.object) ((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(); notify_pool.Append(previous_notification); previous_notification = NULL; @@ -629,7 +629,7 @@ void NormInstance::Shutdown() } if (NORM_OBJECT_INVALID != n->event.object) ((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(); delete n; } diff --git a/common/normApi.h b/common/normApi.h index 2dbf9a7..7e7a307 100644 --- a/common/normApi.h +++ b/common/normApi.h @@ -501,6 +501,23 @@ bool NormNodeGetAddress(NormNodeHandle nodeHandle, NORM_API_LINKAGE 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 void NormNodeRetain(NormNodeHandle nodeHandle); diff --git a/common/normApp.cpp b/common/normApp.cpp index c3463c3..a6fc6e9 100644 --- a/common/normApp.cpp +++ b/common/normApp.cpp @@ -40,6 +40,7 @@ class NormApp : public NormController, public ProtoApp private: void ShowHelp(); void OnInputReady(); + bool AddAckingNodes(const char* ackingNodeList); enum CmdType {CMD_INVALID, CMD_NOARG, CMD_ARG}; CmdType CommandType(const char* cmd); @@ -111,6 +112,8 @@ class NormApp : public NormController, public ProtoApp double tx_repeat_interval; bool tx_repeat_clear; ProtoTimer interval_timer; + char* acking_node_list; // comma-delimited string + bool watermark_pending; // NormSession client-only parameters unsigned long rx_buffer_size; // bytes @@ -144,6 +147,7 @@ NormApp::NormApp() group_size(NormSession::DEFAULT_GSIZE_ESTIMATE), 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), + acking_node_list(NULL), watermark_pending(false), rx_buffer_size(1024*1024), rx_sock_buffer_size(0), rx_cache_path(NULL), post_processor(NULL), unicast_nacks(false), silent_client(false), tracing(false), tx_loss(0.0), rx_loss(0.0) @@ -170,6 +174,7 @@ NormApp::~NormApp() if (interface_name) delete[] interface_name; if (rx_cache_path) delete[] rx_cache_path; if (post_processor) delete post_processor; + if (acking_node_list) delete[] acking_node_list; } // NOTE on message flushing mode: @@ -216,6 +221,7 @@ const char* const NormApp::cmd_list[] = "+rinterval", // Interval (sec) between file/directory list repeats "-oneshot", // Transmit file(s), exiting upon TX_FLUSH_COMPLETED "-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 "+segment", // payload segment size (bytes) "+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" " -oneshot, // Exit upon sender TX_FLUSH_COMPLETED event (sender exits after 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" " +segment, // payload segment size (bytes)\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)) { 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)) { @@ -909,6 +941,70 @@ bool NormApp::ProcessCommands(int argc, const char*const* argv) return true; } // 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*/, ProtoDispatcher::Event /*theEvent*/, const void* userData) @@ -1100,8 +1196,8 @@ void NormApp::Notify(NormController::Event event, } break; case TX_QUEUE_EMPTY: - // Write to stream as needed - //DMSG(0, "NormApp::Notify(TX_QUEUE_EMPTY) ...\n"); + // Write to stream as needed + DMSG(3, "NormApp::Notify(TX_QUEUE_EMPTY) ...\n"); if (NULL != object) { if (input && (object == tx_stream)) @@ -1118,17 +1214,33 @@ void NormApp::Notify(NormController::Event event, } break; + case TX_OBJECT_SENT: + DMSG(3, "NormApp::Notify(TX_OBJECT_SENT) ...\n"); + break; + case TX_FLUSH_COMPLETED: + DMSG(3, "NormApp::Notify(TX_FLUSH_COMPLETED) ...\n"); if (tx_one_shot) { DMSG(0, "norm: transmit flushing completed, exiting.\n"); Stop(); } 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: { - //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 switch (object->GetType()) { @@ -1230,7 +1342,7 @@ void NormApp::Notify(NormController::Event event, } case RX_OBJECT_INFO: - //DMSG(0, "NormApp::Notify(RX_OBJECT_INFO) ...\n"); + DMSG(3, "NormApp::Notify(RX_OBJECT_INFO) ...\n"); switch(object->GetType()) { case NormObject::FILE: @@ -1271,7 +1383,7 @@ void NormApp::Notify(NormController::Event event, break; case RX_OBJECT_UPDATED: - //DMSG(0, "NormApp::Notify(RX_OBJECT_UPDATED) ...\n"); + DMSG(3, "NormApp::Notify(RX_OBJECT_UPDATED) ...\n"); switch (object->GetType()) { case NormObject::FILE: @@ -1414,7 +1526,7 @@ void NormApp::Notify(NormController::Event event, // (TBD) if we're not archiving files we should // manage our cache, deleting the cache // on shutdown ... - //DMSG(0, "NormApp::Notify(RX_OBJECT_COMPLETE) ...\n"); + DMSG(3, "NormApp::Notify(RX_OBJECT_COMPLETED) ...\n"); switch(object->GetType()) { case NormObject::FILE: @@ -1477,7 +1589,8 @@ bool NormApp::OnIntervalTimeout(ProtoTimer& /*theTimer*/) char temp[PATH_MAX]; strncpy(temp, fileNameInfo, len); 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", fileName); @@ -1487,6 +1600,15 @@ bool NormApp::OnIntervalTimeout(ProtoTimer& /*theTimer*/) ActivateTimer(interval_timer); 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); interval_timer.SetInterval(tx_object_interval); } @@ -1587,6 +1709,13 @@ bool NormApp::OnStartup(int argc, const char*const* argv) session->SetBackoffFactor(backoff_factor); session->ServerSetGrtt(grtt_estimate); 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 UINT16 instanceId = baseId; if (!session->StartServer(instanceId, tx_buffer_size, segment_size, ndata, nparity, interface_name)) diff --git a/common/normEncoder.cpp b/common/normEncoder.cpp index 16a9402..298afbc 100644 --- a/common/normEncoder.cpp +++ b/common/normEncoder.cpp @@ -103,7 +103,7 @@ bool NormEncoder::CreateGeneratorPolynomial() { unsigned char *tp, *tp1, *tp2; int degree = 2*npar; - if(gen_poly) delete gen_poly; + if(gen_poly) delete[] gen_poly; 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", GetErrorString()); - delete gen_poly; + delete[] gen_poly; return false; } if(!(tp1 = new unsigned char[2*degree])) { - delete tp; - delete gen_poly; + delete[] tp; + delete[] gen_poly; DMSG(0, "NormEncoder: Error allocating memory while computing gen_poly: %s\n", GetErrorString()); return false; } if(!(tp2 = new unsigned char[2*degree])) { - delete tp1; - delete tp; - delete gen_poly; + delete[] tp1; + delete[] tp; + delete[] gen_poly; DMSG(0, "NormEncoder: Error allocating memory while computing gen_poly: %s\n", GetErrorString()); return false; @@ -161,9 +161,9 @@ bool NormEncoder::CreateGeneratorPolynomial() memcpy(tp1, gen_poly, (npar+1)*sizeof(unsigned char)); memset(&tp1[npar+1], 0, (2*degree)-(npar+1)); } - delete tp2; - delete tp1; - delete tp; + delete[] tp2; + delete[] tp1; + delete[] tp; return true; } // end NormEncoder::CreateGeneratorPolynomial() @@ -303,27 +303,27 @@ void NormDecoder::Destroy() { if (scratch) { - delete scratch; + delete[] scratch; scratch = NULL; } if(o_vec) { for(int i=0; i> 8)) * (10.0/256.0); - double exponent = (double)(rate & 0xff); + double mantissa = ((double)(rate >> 4)) * (10.0/4096.0); + double exponent = (double)(rate & 0x000f); return mantissa * pow(10.0, exponent); } @@ -663,41 +663,39 @@ class NormDataMsg : public NormObjectMsg { ((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) { UINT16 temp16 = htons(len); 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) { UINT32 temp32 = htonl(offset); 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) { UINT16 temp16; memcpy(&temp16, payload+PAYLOAD_LENGTH_OFFSET, 2); 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) { UINT32 temp32; memcpy(&temp32, payload+PAYLOAD_OFFSET_OFFSET, 4); return (ntohl(temp32)); } - - private: enum @@ -710,9 +708,9 @@ class NormDataMsg : public NormObjectMsg enum { //PAYLOAD_FLAGS_OFFSET = 0, // deprecated - PAYLOAD_MSG_START_OFFSET = 0, - PAYLOAD_LENGTH_OFFSET = PAYLOAD_MSG_START_OFFSET+2, - PAYLOAD_OFFSET_OFFSET = PAYLOAD_LENGTH_OFFSET+2, + PAYLOAD_LENGTH_OFFSET = 0, + PAYLOAD_MSG_START_OFFSET = PAYLOAD_LENGTH_OFFSET+2, + PAYLOAD_OFFSET_OFFSET = PAYLOAD_MSG_START_OFFSET+2, PAYLOAD_DATA_OFFSET = PAYLOAD_OFFSET_OFFSET+4 }; }; // end class NormDataMsg @@ -1578,7 +1576,7 @@ class NormAckFlushMsg : public NormAckMsg } private: - // These are the payload offsets for "fec_id" = 129 + // Note - These are the payload offsets for "fec_id" = 129 // "fec_payload_id" field enum { diff --git a/common/normNode.cpp b/common/normNode.cpp index b95fcca..bc484eb 100644 --- a/common/normNode.cpp +++ b/common/normNode.cpp @@ -52,7 +52,7 @@ NormServerNode::NormServerNode(class NormSession& theSession, NormNodeId nodeId) cc_sequence(0), cc_enable(false), cc_rate(0.0), rtt_confirmed(false), is_clr(false), is_plr(false), 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) { @@ -67,7 +67,7 @@ NormServerNode::NormServerNode(class NormSession& theSession, NormNodeId nodeId) repair_timer.SetRepeat(1); 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; activity_timer.SetInterval(activityInterval); 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_usec = 0; - recv_rate_quantized = NormQuantizeRate(recv_rate); - } NormServerNode::~NormServerNode() @@ -299,29 +297,27 @@ void NormServerNode::FreeBuffers() segment_size = ndata = nparity = 0; } // 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, const NormCmdMsg& cmd) { UINT8 grttQuantized = cmd.GetGrtt(); - if (grttQuantized != grtt_quantized) - { - 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); - } + if (grttQuantized != grtt_quantized) UpdateGrttEstimate(grttQuantized); UINT8 gsizeQuantized = cmd.GetGroupSize(); if (gsizeQuantized != gsize_quantized) { @@ -421,16 +417,16 @@ void NormServerNode::HandleCommand(const struct timeval& currentTime, ExponentialRand(maxBackoff, gsize_estimate) : 0.0; // Bias backoff timeout based on our rate double r; - if (slow_start) + double ccLoss = slow_start ? 0.0 : LossEstimate(); + if (0.0 == ccLoss) { r = recv_rate / send_rate; cc_rate = 2.0 * recv_rate; } else { - cc_rate = NormSession::CalculateRate(nominal_packet_size, - rtt_estimate, - LossEstimate()); + double nominalSize = nominal_packet_size ? nominal_packet_size : segment_size; + cc_rate = NormSession::CalculateRate(nominalSize, rtt_estimate, ccLoss); r = cc_rate / send_rate; r = MIN(r, 0.9); 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 // and/or lower rate - double localRate = slow_start ? - (2.0*recv_rate) : - NormSession::CalculateRate(nominal_packet_size, - rtt_estimate, - LossEstimate()); + double nominalSize = nominal_packet_size ? nominal_packet_size : segment_size; + double ccLoss = slow_start ? 0.0 : LossEstimate(); + double localRate = (0.0 == ccLoss) ? + (2.0*recv_rate) : + NormSession::CalculateRate(nominalSize, + rtt_estimate, + ccLoss); // This increases our chance of being suppressed // (but is it a good idea?) localRate = MAX(localRate, cc_rate); @@ -890,27 +888,7 @@ char* NormServerNode::GetFreeSegment(NormObjectId objectId, NormBlockId blockId) void NormServerNode::HandleObjectMessage(const NormObjectMsg& msg) { UINT8 grttQuantized = msg.GetGrtt(); - if (grttQuantized != grtt_quantized) - { - 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); - } + if (grttQuantized != grtt_quantized) UpdateGrttEstimate(grttQuantized); UINT8 gsizeQuantized = msg.GetGroupSize(); if (gsizeQuantized != gsize_quantized) { @@ -1123,9 +1101,11 @@ void NormServerNode::HandleObjectMessage(const NormObjectMsg& msg) #endif // !SIMULATE 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); DeleteObject(obj); + obj = NULL; completion_count++; } } @@ -1141,7 +1121,7 @@ void NormServerNode::HandleObjectMessage(const NormObjectMsg& msg) // Optional "object boundary repair check (non-streams only!) // (checks for repair needs for objects _prior_ to 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); else RepairCheck(NormObject::THRU_INFO, objectId, blockId, segmentId); @@ -1368,7 +1348,8 @@ void NormServerNode::RepairCheck(NormObject::CheckLevel checkLevel, NormSegmentId segmentId) { ASSERT(synchronized); - if (objectId > max_pending_object) max_pending_object = objectId; + if (objectId > max_pending_object) + max_pending_object = objectId; if (!repair_timer.IsActive()) { // repair timer inactive @@ -1507,17 +1488,19 @@ bool NormServerNode::OnRepairTimeout(ProtoTimer& /*theTimer*/) if (rtt_confirmed) ext.SetCCFlag(NormCC::RTT); ext.SetCCRtt(rtt_quantized); - double ccLoss = LossEstimate(); + double ccLoss = slow_start ? 0.0 : LossEstimate(); UINT16 lossQuantized = NormQuantizeLoss(ccLoss); 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)); } 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, ccLoss); 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 // 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) - if (measurementInterval < NORM_TICK_MIN) - measurementInterval = NORM_TICK_MIN; + if (measurementInterval < NORM_TICK_MIN) measurementInterval = NORM_TICK_MIN; recv_accumulator += msgSize; - unsigned short oldRateQuantized = recv_rate_quantized; if (interval > 0.0) { double currentRecvRate = ((double)(recv_accumulator)) / interval; @@ -1759,36 +1740,17 @@ void NormServerNode::UpdateRecvRate(const struct timeval& currentTime, unsigned { recv_rate = currentRecvRate; } - recv_rate_quantized = NormQuantizeRate(recv_rate); } else if (0.0 == recv_rate) { // Approximate initial recv_rate when initial packets arrive in a burst 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 { recv_rate = 0.0; - recv_rate_quantized = NormQuantizeRate(0.0); prev_update_time = currentTime; recv_accumulator = 0; nominal_packet_size = msgSize; @@ -1800,13 +1762,8 @@ void NormServerNode::Activate(bool isObjectMsg) { if (!activity_timer.IsActive()) { - 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; + double activityInterval = 2*NORM_ROBUST_FACTOR*grtt_estimate; + if (activityInterval < ACTIVITY_INTERVAL_MIN) activityInterval = ACTIVITY_INTERVAL_MIN; activity_timer.SetInterval(activityInterval); session.ActivateTimer(activity_timer); server_active = false; @@ -1834,7 +1791,7 @@ bool NormServerNode::OnActivityTimeout(ProtoTimer& /*theTimer*/) } else { - DMSG(4, "NormServerNode::OnActivityTimeout() node>%lu for server>%lu\n", + DMSG(2, "NormServerNode::OnActivityTimeout() node>%lu for server>%lu\n", LocalNodeId(), GetId()); struct timeval currentTime; ::ProtoSystemTime(currentTime); @@ -1858,6 +1815,19 @@ bool NormServerNode::OnActivityTimeout(ProtoTimer& /*theTimer*/) // Let's try this instead RepairCheck(NormObject::THRU_OBJECT, // (TBD) thru object??? 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 { @@ -1876,24 +1846,30 @@ bool NormServerNode::UpdateLossEstimate(const struct timeval& currentTime, unsigned short seq, bool ecn) { - bool result = loss_estimator.Update(currentTime, seq, ecn); - if (result && slow_start) + if (loss_estimator.Update(currentTime, seq, ecn)) { - // Calculate loss initialization based on current receive rate - // and rtt estimation - double lossInit = (recv_rate * rtt_estimate) / - (nominal_packet_size*sqrt(3.0/2.0)); - lossInit = (lossInit*lossInit); - // At this point, "LastLossInterval()" will always be non-zero - double altLoss = (double)loss_estimator.LastLossInterval(); - double altInit = (altLoss > 0.0) ? (1.0 / altLoss) : 0.5; - if (altInit < lossInit) - loss_estimator.SetInitialLoss(altInit); - else - loss_estimator.SetInitialLoss(lossInit); - slow_start = false; + if (slow_start) + { + // Calculate loss initialization based on current receive rate + // and rtt estimation + double nominalSize = (nominal_packet_size > segment_size) ? nominal_packet_size : segment_size; + double lossInit = (recv_rate * rtt_estimate) / (nominalSize*sqrt(3.0/2.0)); + lossInit = (lossInit*lossInit); + // At this point, "LastLossInterval()" will always be non-zero! + double altLoss = (double)loss_estimator.LastLossInterval(); + double altInit = (altLoss > 0.0) ? (1.0 / altLoss) : 0.5; + if (altInit < lossInit) + loss_estimator.SetInitialLoss(altInit); + else + loss_estimator.SetInitialLoss(lossInit); + slow_start = false; + } + return true; + } + else + { + return false; } - return result; } // end NormServerNode::UpdateLossEstimate() void NormServerNode::AttachCCFeedback(NormAckMsg& ack) @@ -1910,18 +1886,18 @@ void NormServerNode::AttachCCFeedback(NormAckMsg& ack) if (rtt_confirmed) ext.SetCCFlag(NormCC::RTT); ext.SetCCRtt(rtt_quantized); - double ccLoss = LossEstimate(); + double ccLoss = slow_start ? 0.0 : LossEstimate(); UINT16 lossQuantized = NormQuantizeLoss(ccLoss); ext.SetCCLoss(lossQuantized); - if (slow_start) + if (0.0 == ccLoss) { ext.SetCCFlag(NormCC::START); ext.SetCCRate(NormQuantizeRate(2.0 * recv_rate)); } else { - double ccRate = NormSession::CalculateRate(nominal_packet_size, - rtt_estimate, ccLoss); + double nominalSize = (nominal_packet_size > segment_size) ? nominal_packet_size : segment_size; + double ccRate = NormSession::CalculateRate(nominalSize, rtt_estimate, ccLoss); 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", @@ -2006,7 +1982,16 @@ bool NormServerNode::OnAckTimeout(ProtoTimer& /*theTimer*/) AttachCCFeedback(*ack); ack->SetObjectId(watermark_object_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); if (unicast_nacks) ack->SetDestination(GetAddress()); @@ -2424,7 +2409,7 @@ double NormLossEstimator::LossFraction() 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_time(0.0), event_index_time(0.0), seeking_loss_event(true), diff --git a/common/normNode.h b/common/normNode.h index 8fe778f..22a81d3 100644 --- a/common/normNode.h +++ b/common/normNode.h @@ -148,7 +148,7 @@ class NormAckingNode : public NormNode NormAckingNode(class NormSession& theSession, NormNodeId nodeId); ~NormAckingNode(); bool IsPending() const - {return (!ack_received && ( req_count > 0));} + {return (!ack_received && (req_count > 0));} void Reset(unsigned int maxAttempts = NORM_ROBUST_FACTOR) { ack_received = false; @@ -238,7 +238,9 @@ class NormServerNode : public NormNode bool UnicastNacks() {return unicast_nacks;} 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, unsigned short theSequence, @@ -445,7 +447,6 @@ class NormServerNode : public NormNode bool slow_start; double send_rate; // sender advertised rate double recv_rate; // measured recv rate - unsigned short recv_rate_quantized; struct timeval prev_update_time; // for recv_rate measurement unsigned long recv_accumulator; // for recv_rate measurement double nominal_packet_size; diff --git a/common/normObject.cpp b/common/normObject.cpp index 663bf75..5d9e305 100644 --- a/common/normObject.cpp +++ b/common/normObject.cpp @@ -15,7 +15,7 @@ NormObject::NormObject(NormObject::Type theType, transport_id(transportId), segment_size(0), pending_info(false), repair_info(false), current_block_id(0), next_segment_id(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) { @@ -48,7 +48,9 @@ void NormObject::Release() { if (server) server->Release(); if (reference_count) + { reference_count--; + } else { 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) first_pass = true; return increasedRepair; } // end NormObject::TxReset() @@ -544,6 +547,7 @@ bool NormObject::AppendRepairAdv(NormCmdRepairAdvMsg& cmd) // This is used by sender for watermark check bool NormObject::FindRepairIndex(NormBlockId& blockId, NormSegmentId& segmentId) const { + ASSERT(!server); if (repair_info) { blockId = 0; @@ -1504,7 +1508,10 @@ bool NormObject::NextServerMsg(NormObjectMsg* msg) NormBlockId blockId; if (!GetFirstPending(blockId)) { - if (!IsStream()) + // Attempt to advance stream (probably was repair-delayed) + if (IsStream()) + static_cast(this)->StreamAdvance(); + else DMSG(0, "NormObject::NextServerMsg() pending object w/ no pending blocks?\n"); return false; } @@ -1544,17 +1551,27 @@ bool NormObject::NextServerMsg(NormObjectMsg* msg) } block->TxInit(blockId, numData, session.ServerAutoParity()); - if (!block_buffer.Insert(block)) + while (!block_buffer.Insert(block)) { ASSERT(STREAM == type); if (blockId > block_buffer.RangeLo()) { NormBlock* b = block_buffer.Find(block_buffer.RangeLo()); - ASSERT(b); - block_buffer.Remove(b); - session.ServerPutFreeBlock(b); - bool success = block_buffer.Insert(block); - ASSERT(success); + ASSERT(NULL != b); + bool push = static_cast(this)->GetPushMode(); + if (!push && (b->IsRepairPending() || IsRepairSet(b->GetId()))) + { + // 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()) { @@ -1574,6 +1591,7 @@ bool NormObject::NextServerMsg(NormObjectMsg* msg) else { ASSERT(0); + DMSG(0, "NormObject::NextServerMsg() invalid non-stream state!\n"); return false; } } @@ -1629,9 +1647,9 @@ bool NormObject::NextServerMsg(NormObjectMsg* msg) ASSERT(block->ParityReady(numData)); char* segment = block->GetSegment(segmentId); ASSERT(segment); - UINT16 payloadLength = block->GetSegSizeMax(); // segment_size; - if (IsStream()) payloadLength += NormDataMsg::GetStreamPayloadHeaderLength(); - data->SetPayload(segment, payloadLength); + // We only need to send FEC content to cover the biggest segment + // sent for the block. + data->SetPayload(segment, block->GetSegSizeMax()); } block->UnsetPending(segmentId); if (block->InRepair()) @@ -1645,16 +1663,29 @@ bool NormObject::NextServerMsg(NormObjectMsg* msg) pending_mask.Unset(blockId); } - // This lets NORM_STREAM objects continue indefinitely - if (IsStream() && !pending_mask.IsSet()) - static_cast(this)->StreamAdvance(); + if (!pending_mask.IsSet()) + { + if (IsStream()) + { + // This lets NORM_STREAM objects continue indefinitely + static_cast(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; } // end NormObject::NextServerMsg() void NormStreamObject::StreamAdvance() { 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 (block_buffer.CanInsert(nextBlockId)) @@ -1671,6 +1702,7 @@ void NormStreamObject::StreamAdvance() if (!block->IsTransmitPending()) { // ??? (TBD) Should this block be returned to the pool right now??? + // especially for "push" enabled streams if (pending_mask.Set(nextBlockId)) stream_next_id++; else @@ -1747,7 +1779,7 @@ NormBlock* NormObject::ServerRecoverBlock(NormBlockId blockId) return (NormBlock*)NULL; } } - // Attempt to re-generate parity + // Attempt to re-generate parity for the block if (CalculateBlockParity(block)) { if (!block_buffer.Insert(block)) @@ -2582,7 +2614,7 @@ UINT16 NormStreamObject::ReadSegment(NormBlockId blockId, //if ((UINT32)offsetDelta < object_size.LSB()) ASSERT(tx_index.block <= write_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()); if (b && !b->IsPending()) @@ -2666,21 +2698,24 @@ bool NormStreamObject::WriteSegment(NormBlockId blockId, block->GetFirstPending(read_index.segment); NormBlock* tempBlock = block; UINT32 tempOffset = read_offset; + NormStreamObject::Index tempIndex = read_index; // (TBD) uncomment the code so that only a single // UPDATED notification is posted??? if (notify_on_update) { notify_on_update = false; session.Notify(NormController::RX_OBJECT_UPDATED, server, this); - } + } block = stream_buffer.Find(stream_buffer.RangeLo()); 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 //ASSERT(0); - dataLost = true; + dataLost = true; block->UnsetPending(read_index.segment++); if (read_index.segment >= ndata) { @@ -3185,12 +3220,15 @@ UINT32 NormStreamObject::Write(const char* buffer, UINT32 len, bool eom) } 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 //ASSERT(deltaOffset >= 0); //if (deltaOffset >= (INT32)object_size.LSB()) ASSERT(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; DMSG(8, "NormStreamObject::Write() stream buffer full (1)\n", len, eom); diff --git a/common/normObject.h b/common/normObject.h index 31f11e1..c304a15 100644 --- a/common/normObject.h +++ b/common/normObject.h @@ -167,6 +167,9 @@ class NormObject NormBlock* ServerRecoverBlock(NormBlockId blockId); bool CalculateBlockParity(NormBlock* block); + /*bool IsFirstPass() {return first_pass;} + void ClearFirstPass() {first_pass = false};*/ + bool TxReset(NormBlockId firstBlock = NormBlockId(0), bool requeue = false); bool TxResetBlocks(NormBlockId nextId, NormBlockId lastId); bool TxUpdateBlock(NormBlock* theBlock, @@ -265,6 +268,7 @@ class NormObject // Here are some members used to let us know // our status with respect to the rest of the world + bool first_pass; // for sender objects bool accepted; bool notify_on_update; @@ -390,6 +394,7 @@ class NormStreamObject : public NormObject SetFlushMode(oldFlushMode); } void SetPushMode(bool state) {push_mode = state;} + bool GetPushMode() const {return push_mode;} bool IsClosing() {return stream_closing;} bool HasVacancy() diff --git a/common/normSegment.cpp b/common/normSegment.cpp index 7e0be60..548f2ec 100644 --- a/common/normSegment.cpp +++ b/common/normSegment.cpp @@ -243,6 +243,7 @@ bool NormBlock::TxReset(UINT16 numData, ptr++; } erasure_count = 0; + seg_size_max = 0; } } return increasedRepair; diff --git a/common/normSession.cpp b/common/normSession.cpp index 4f1de8c..300e29f 100644 --- a/common/normSession.cpp +++ b/common/normSession.cpp @@ -26,7 +26,7 @@ NormSession::NormSession(NormSessionMgr& sessionMgr, NormNodeId localNodeId) tx_cache_size_max((UINT32)20*1024*1024), flush_count(NORM_ROBUST_FACTOR+1), 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), probe_proactive(true), probe_pending(false), probe_reset(true), probe_data_check(false), grtt_interval(0.5), @@ -68,7 +68,7 @@ NormSession::NormSession(NormSessionMgr& sessionMgr, NormNodeId localNodeId) grtt_quantized = NormQuantizeRtt(DEFAULT_GRTT_ESTIMATE); grtt_measured = grtt_advertised = NormUnquantizeRtt(grtt_quantized); - + gsize_measured = DEFAULT_GSIZE_ESTIMATE; gsize_quantized = NormQuantizeGroupSize(DEFAULT_GSIZE_ESTIMATE); gsize_advertised = NormUnquantizeGroupSize(gsize_quantized); @@ -249,6 +249,11 @@ bool NormSession::SetMulticastInterface(const char* interfaceName) void NormSession::SetTxRateInternal(double txRate) { + if (!is_server) + { + tx_rate = txRate; + return; + } if (txRate < 0.0) { DMSG(0, "NormSession::SetTxRateInternal() invalid transmit rate!\n"); @@ -287,6 +292,7 @@ void NormSession::SetTxRateInternal(double txRate) grtt_quantized = NormQuantizeRtt(grtt_measured); grtt_advertised = NormUnquantizeRtt(grtt_quantized); + // What do we do when "pktInterval" > "grtt_max"? // We will take our lumps with some extra activity timeout NACKs when they happen? if (grtt_advertised > grtt_max) @@ -294,7 +300,6 @@ void NormSession::SetTxRateInternal(double txRate) grtt_quantized = NormQuantizeRtt(grtt_max); grtt_advertised = NormUnquantizeRtt(grtt_quantized); } - if (grttQuantizedOld != grtt_quantized) { 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 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_active = false; @@ -451,6 +460,7 @@ bool NormSession::StartServer(UINT16 instanceId, OnProbeTimeout(probe_timer); ActivateTimer(probe_timer); } + return true; } // end NormSession::StartServer() @@ -462,6 +472,11 @@ void NormSession::StopServer() probe_timer.Deactivate(); probe_reset = true; } + if (repair_timer.IsActive()) + { + repair_timer.Deactivate(); + tx_repair_pending = false; + } encoder.Destroy(); acking_node_tree.Destroy(); cc_node_list.Destroy(); @@ -525,8 +540,9 @@ void NormSession::Serve() obj = tx_table.Find(objectId); 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 NormObject* nextObj; @@ -535,7 +551,7 @@ void NormSession::Serve() NormSegmentId nextSegmentId = 0; if (obj) { - // Use current transmit pending object + // Get index (objectId::blockId::segmentId) of next transmit pending segment nextObj = obj; nextObjectId = objectId; if (nextObj->IsPending()) @@ -551,9 +567,8 @@ void NormSession::Serve() block->GetFirstPending(nextSegmentId); #endif // if/else PROTO_DEBUG // Adjust so watermark segmentId < block length - if (nextSegmentId >= nextObj->GetBlockSize(nextBlockId)) - nextSegmentId = nextObj->GetBlockSize(nextBlockId) - 1; - + UINT16 nextBlockSize = nextObj->GetBlockSize(nextBlockId); + if (nextSegmentId >= nextBlockSize) nextSegmentId = nextBlockSize - 1; } } else @@ -564,56 +579,77 @@ void NormSession::Serve() else { // Must be an active, but non-pending stream object + ASSERT(nextObj->IsStream()); nextBlockId = static_cast(nextObj)->GetNextBlockId(); nextSegmentId = static_cast(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); - if (ServerGetFirstRepairPending(nextObjectId)) + while ((nextObj = iterator.GetNextObject())) { - while ((nextObj = iterator.GetNextObject())) + if (repairObjectId < nextObj->GetId()) { - if (nextObjectId < nextObj->GetId()) - { - nextObj = tx_table.Find(nextObjectId); - break; - } - else if (nextObj->IsRepairPending()) - { - nextObjectId = nextObj->GetId(); - break; - } - } - } - else - { - nextObjectId = next_tx_object_id; - while ((nextObj = iterator.GetNextObject())) + nextObj = tx_table.Find(repairObjectId); + break; + } + else if (nextObj->IsRepairPending()) { - if (nextObj->IsRepairPending()) - { - nextObjectId = nextObj->GetId(); - break; - } - } - } - if (nextObj) + repairObjectId = nextObj->GetId(); + break; + } + } + if (nextObj) nextObj->FindRepairIndex(repairBlockId, repairSegmentId); + + // Get min of next transmit pending or repair pending segment + if ((repairObjectId < nextObjectId) || + ((repairObjectId == nextObjectId) && + ((repairBlockId < nextBlockId) || + ((repairBlockId == nextBlockId) && + (repairSegmentId < nextSegmentId))))) { -#ifdef PROTO_DEBUG - ASSERT(nextObj->FindRepairIndex(nextBlockId, nextSegmentId)); -#else - nextObj->FindRepairIndex(nextBlockId, nextSegmentId); -#endif + nextObjectId = repairObjectId; + nextBlockId = repairBlockId; + nextSegmentId = repairSegmentId; } - } + */ + } // end if (tx_repair_pending) + if ((nextObjectId > watermark_object_id) || ((nextObjectId == watermark_object_id) && ((nextBlockId > watermark_block_id) || - (((nextBlockId == watermark_block_id) && - (nextSegmentId > watermark_segment_id)))))) + ((nextBlockId == watermark_block_id) && + (nextSegmentId > watermark_segment_id))))) { if (ServerQueueWatermarkFlush()) { @@ -637,8 +673,7 @@ void NormSession::Serve() NormAckingNode* next; while ((next = static_cast(iterator.GetNextNode()))) { - //if (next->IsPending()) - next->ResetReqCount(); + next->ResetReqCount(); } } } @@ -674,21 +709,15 @@ void NormSession::Serve() msg->SetGroupSize(gsize_quantized); QueueMessage(msg); flush_count = 0; - //if (flush_timer.IsActive()) flush_timer.Deactivate(); - if (!obj->IsPending()) - { - if (obj->IsStream()) - posted_tx_queue_empty = true; // repair-delayed stream advance - else - tx_pending_mask.Unset(obj->GetId()); - // (TBD) post TX_OBJECT_SENT notification - - } - else if (obj->IsStream()) - { - // Is there room write to the stream - // should we post TX_QUEUE_VACANCY here? - } + // (TBD) ??? should streams every allowed to be non-pending? + // we _could_ re-architect streams a little bit and allow + // for this by having NormStreamObject::Write() control + // stream advancement ... I think it would be cleaner. + // (mod NormStreamObject::StreamAdvance() to depend upon + // what has been written and conversely set some pending + // state as calls to NormStreamObject::Write() are made. + if (!obj->IsPending() && !obj->IsStream()) + tx_pending_mask.Unset(obj->GetId()); } 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; posted_tx_queue_empty = true; @@ -758,7 +787,11 @@ void NormSession::Serve() if (flush_count < NORM_ROBUST_FACTOR) { // 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) { @@ -878,6 +911,16 @@ bool NormSession::ServerQueueWatermarkFlush() flush->SetGroupSize(gsize_quantized); flush->SetObjectId(watermark_object_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); NormNodeTreeIterator iterator(acking_node_tree); NormAckingNode* next; @@ -960,7 +1003,7 @@ bool NormSession::ServerQueueWatermarkFlush() 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; NormObject* obj = tx_table.Find(tx_table.RangeHi()); NormObjectId objectId; @@ -981,17 +1024,43 @@ void NormSession::ServerQueueFlush() blockId = obj->GetFinalBlockId(); 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 { // Why did I do this? - Brian // Because a squelch keeps the receivers from NACKing in futility // (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 (flush_count < NORM_ROBUST_FACTOR) flush_count++; - flush_timer.SetInterval(4*grtt_advertised); - ActivateTimer(flush_timer); - DMSG(8, "NormSession::ServerQueueFlush() node>%lu squelch queued (flush_count:%u)...\n", + DMSG(4, "NormSession::ServerQueueFlush() node>%lu squelch queued (flush_count:%u)...\n", LocalNodeId(), flush_count); } else @@ -999,30 +1068,8 @@ void NormSession::ServerQueueFlush() DMSG(0, "NormSession::ServerQueueFlush() warning: node>%lu unable to queue squelch\n", LocalNodeId()); } - return; } - 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->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); + flush_timer.SetInterval(2*grtt_advertised); ActivateTimer(flush_timer); } // end NormSession::ServerQueueFlush() @@ -2098,6 +2145,8 @@ void NormSession::ServerHandleAckMessage(const struct timeval& currentTime, cons } 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"); } } @@ -2168,6 +2217,7 @@ void NormSession::ServerHandleNackMessage(const struct timeval& currentTime, Nor bool squelchQueued = false; + // Get the index of our next pending NORM_DATA transmission NormObjectId txObjectIndex; NormBlockId txBlockIndex; if (ServerGetFirstPending(txObjectIndex)) @@ -2193,7 +2243,7 @@ void NormSession::ServerHandleNackMessage(const struct timeval& currentTime, Nor txObjectIndex = next_tx_object_id; txBlockIndex = 0; } - + bool holdoff = (repair_timer.IsActive() && !repair_timer.GetRepeatCount()); enum NormRequestLevel {SEGMENT, BLOCK, INFO, OBJECT}; while ((requestLength = nack.UnpackRepairRequest(req, requestOffset))) @@ -2202,15 +2252,26 @@ void NormSession::ServerHandleNackMessage(const struct timeval& currentTime, Nor requestOffset += requestLength; NormRequestLevel requestLevel; if (req.FlagIsSet(NormRepairRequest::SEGMENT)) + { requestLevel = SEGMENT; + } else if (req.FlagIsSet(NormRepairRequest::BLOCK)) + { requestLevel = BLOCK; + } else if (req.FlagIsSet(NormRepairRequest::OBJECT)) + { requestLevel = OBJECT; - else + } + else if (req.FlagIsSet(NormRepairRequest::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); @@ -2280,6 +2341,23 @@ void NormSession::ServerHandleNackMessage(const struct timeval& currentTime, Nor } 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(); startTimer = true; } @@ -2307,6 +2385,23 @@ void NormSession::ServerHandleNackMessage(const struct timeval& currentTime, Nor } 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); startTimer = true; } @@ -2382,6 +2477,31 @@ void NormSession::ServerHandleNackMessage(const struct timeval& currentTime, Nor } 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); startTimer = true; } @@ -2394,17 +2514,23 @@ void NormSession::ServerHandleNackMessage(const struct timeval& currentTime, Nor if (nextBlockId != prevBlockId) freshBlock = true; if (freshBlock) { - freshBlock = false; // Is this entire block already repair pending? if (object->IsRepairSet(nextBlockId)) - break; - if (!(block = object->FindBlock(nextBlockId))) + continue; + if (NULL == (block = object->FindBlock(nextBlockId))) { // 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 - if (!(block = object->ServerRecoverBlock(nextBlockId))) + if (NULL == (block = object->ServerRecoverBlock(nextBlockId))) { if (NormObject::STREAM == object->GetType()) { @@ -2419,26 +2545,20 @@ void NormSession::ServerHandleNackMessage(const struct timeval& currentTime, Nor } else { - // Resource constrained, move on. + // Resource constrained, move on to next repair request DMSG(2, "NormSession::ServerHandleNackMessage() node>%lu " "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; prevBlockId = nextBlockId; - } + } // end if (freshBlock) + ASSERT(NULL != block); // If stream && explicit data repair, lock the data for retransmission - if (object->IsStream() && (nextSegmentId < ndata)) { bool attemptLock = true; @@ -2541,16 +2661,50 @@ void NormSession::ServerHandleNackMessage(const struct timeval& currentTime, Nor } 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, - object->GetBlockSize(block->GetId()), - nparity, numErasures); + nextBlockSize, nparity, + numErasures); startTimer = true; } // end if/else (holdoff) break; case INFO: + // We already dealt with INFO request above with respect to initiating repair nextObjectId++; - if (nextObjectId > lastObjectId) - inRange = false; + if (nextObjectId > lastObjectId) inRange = false; break; } // end switch(requestLevel) } // end while(inRange) @@ -2649,7 +2803,9 @@ bool NormSession::ServerQueueSquelch(NormObjectId objectId) { ASSERT(NormObject::STREAM == obj->GetType()); squelch->SetObjectId(objectId); - squelch->SetFecBlockId(((NormStreamObject*)obj)->StreamBufferLo()); + NormBlockId blockId = static_cast(obj)->StreamBufferLo(); + squelch->SetFecBlockId(blockId); + squelch->SetFecBlockLen(obj->GetBlockSize(blockId)); squelch->SetFecSymbolId(0); squelch->ResetInvalidObjectList(); while ((obj = iterator.GetNextObject())) @@ -2662,10 +2818,13 @@ bool NormSession::ServerQueueSquelch(NormObjectId objectId) if (obj) { squelch->SetObjectId(obj->GetId()); + NormBlockId blockId; if (obj->IsStream()) - squelch->SetFecBlockId(((NormStreamObject*)obj)->StreamBufferLo()); + blockId =static_cast(obj)->StreamBufferLo(); else - squelch->SetFecBlockId(0); + blockId = NormBlockId(0); + squelch->SetFecBlockId(blockId); + squelch->SetFecBlockLen(obj->GetBlockSize(blockId)); squelch->SetFecSymbolId(0); nextId = obj->GetId() + 1; } @@ -2674,6 +2833,7 @@ bool NormSession::ServerQueueSquelch(NormObjectId objectId) // Squelch to point to future object squelch->SetObjectId(next_tx_object_id); squelch->SetFecBlockId(0); + squelch->SetFecBlockLen(0); // (TBD) should this be "ndata" instead? but we can't be sure squelch->SetFecSymbolId(0); nextId = next_tx_object_id; } @@ -2821,6 +2981,7 @@ bool NormSession::ServerBuildRepairAdv(NormCmdRepairAdvMsg& cmd) bool NormSession::OnRepairTimeout(ProtoTimer& /*theTimer*/) { + tx_repair_pending = false; if (repair_timer.GetRepeatCount()) { // NACK aggregation period has ended. (incorporate accumulated repair requests) @@ -2859,9 +3020,7 @@ bool NormSession::OnRepairTimeout(ProtoTimer& /*theTimer*/) } } } // end while (iterator.GetNextObject()) - // (TBD) should this be just PromptServer() instead??? - // (note we do set posted_tx_queue_empty when repair state blocks stream progress) - TouchServer(); + PromptServer(); // BACKOFF related code // Holdoff initiation of new repair cycle for one GRTT // (TBD) for unicast sessions, use CLR RTT ??? @@ -3053,8 +3212,7 @@ bool NormSession::SendMessage(NormMsg& msg) } else { - if (tx_socket->SendTo(msg.GetBuffer(), msgSize, - msg.GetDestination())) + if (tx_socket->SendTo(msg.GetBuffer(), msgSize, msg.GetDestination())) { // Separate send/recv tracing if (trace) diff --git a/common/normSession.h b/common/normSession.h index a2c4c42..ff47f1a 100644 --- a/common/normSession.h +++ b/common/normSession.h @@ -162,7 +162,7 @@ class NormSession void SetTxRate(double txRate) { txRate /= 8.0; // convert to bytes/sec - SetTxRateInternal(txRate); + SetTxRateInternal(txRate); } double BackoffFactor() {return backoff_factor;} void SetBackoffFactor(double value) {backoff_factor = value;} @@ -457,6 +457,10 @@ class NormSession NormObjectId watermark_object_id; NormBlockId watermark_block_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 bool advertise_repairs; diff --git a/common/normTest.cpp b/common/normTest.cpp index 20b8440..b738509 100644 --- a/common/normTest.cpp +++ b/common/normTest.cpp @@ -68,6 +68,9 @@ int main(int argc, char* argv[]) //NormSetTOS(session, 0x20); // 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); // Uncomment to turn on debug NORM message tracing //NormSetMessageTrace(session, true); @@ -82,10 +85,10 @@ int main(int argc, char* argv[]) 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()) - NormSetCongestionControl(session, true); + //NormSetCongestionControl(session, true); //NormSetTransmitRateBounds(session, 1000.0, -1.0); @@ -96,20 +99,22 @@ int main(int argc, char* argv[]) // is _not_ recommended when unicast feedback may be // possible!) //NormSetTxPort(session, 6001); - + + //NormSetDefaultUnicastNack(session, true); + // Uncomment to allow reuse of rx port // (This allows multiple "normTest" instances on the same machine // for the same NormSession - note those instances must use // unique local NormNodeIds (see NormCreateSession() above). - //NormSetRxPortReuse(session, true); + NormSetRxPortReuse(session, true); // Uncomment to receive your own traffic - //NormSetLoopback(session, true); + NormSetLoopback(session, true); //NormSetSilentReceiver(session, true); // Uncomment this line to participate as a receiver - //NormStartReceiver(session, 1024*1024); + NormStartReceiver(session, 1024*1024); // Uncomment to set large rx socket buffer size // (might be needed for high rate sessions) @@ -119,25 +124,29 @@ int main(int argc, char* argv[]) NormSessionId sessionId = (NormSessionId)rand(); // 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); // Uncomment to set large tx socket buffer size // (might be needed for high rate sessions) - NormSetTxSocketBuffer(session, 512000); + //NormSetTxSocketBuffer(session, 512000); NormAddAckingNode(session, NORM_NODE_NONE); //15); //NormGetLocalNodeId(session)); 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"; +#endif //const char* filePath = "/home/adamson/pkgs/rh73.tgz"; const char* fileName = "file1.jpg"; const char* fileName2 = "file2.jpg"; // 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 // the stream. @@ -261,14 +270,6 @@ int main(int argc, char* argv[]) case NORM_TX_OBJECT_PURGED: DMSG(2, "normTest: NORM_TX_OBJECT_PURGED event ...\n"); 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: DMSG(3, "normTest: NORM_RX_OBJECT_NEW event ...\n"); @@ -344,7 +345,7 @@ int main(int argc, char* argv[]) //else // TRACE("validated recv msg len:%d\n", len); recvCount = value+1; - if (0 == msgCount % 100) + if (0 == msgCount % 1000) { TRACE("normTest: status> msgCount:%d of total:%d (%lf)\n", msgCount, recvCount, 100.0*((double)msgCount)/((double)recvCount)); diff --git a/common/normVersion.h b/common/normVersion.h index 1d8572e..bcd957c 100644 --- a/common/normVersion.h +++ b/common/normVersion.h @@ -32,6 +32,6 @@ #ifndef _NORM_VERSION #define _NORM_VERSION -#define VERSION "1.3b8" +#define VERSION "1.3b9" #endif // _NORM_VERSION diff --git a/examples/normFileRecv.cpp b/examples/normFileRecv.cpp index c030a39..618f51c 100644 --- a/examples/normFileRecv.cpp +++ b/examples/normFileRecv.cpp @@ -144,16 +144,16 @@ int main(int argc, char* argv[]) fprintf(stderr, "normFileRecv: NORM_RX_OBJECT_ABORTED event ...\n"); break; - case NORM_REMOTE_SERVER_NEW: - fprintf(stderr, "normFileRecv: NORM_REMOTE_SERVER_NEW event ...\n"); + case NORM_REMOTE_SENDER_NEW: + fprintf(stderr, "normFileRecv: NORM_REMOTE_SENDER_NEW event ...\n"); break; - case NORM_REMOTE_SERVER_ACTIVE: - fprintf(stderr, "normFileRecv: NORM_REMOTE_SERVER_ACTIVE event ...\n"); + case NORM_REMOTE_SENDER_ACTIVE: + fprintf(stderr, "normFileRecv: NORM_REMOTE_SENDER_ACTIVE event ...\n"); break; - case NORM_REMOTE_SERVER_INACTIVE: - fprintf(stderr, "normFileRecv: NORM_REMOTE_SERVER_INACTIVE event ...\n"); + case NORM_REMOTE_SENDER_INACTIVE: + fprintf(stderr, "normFileRecv: NORM_REMOTE_SENDER_INACTIVE event ...\n"); break; default: diff --git a/examples/normFileSend.cpp b/examples/normFileSend.cpp index b110778..101960b 100644 --- a/examples/normFileSend.cpp +++ b/examples/normFileSend.cpp @@ -73,7 +73,7 @@ int main(int argc, char* argv[]) // Uncomment to turn on debug NORM message tracing NormSetMessageTrace(session, true); // 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; ProtoSystemTime(currentTime); // Uncomment to get different packet loss patterns from run to run diff --git a/win32/Norm.suo b/win32/Norm.suo index ce5e9bda2338685a3f11694d38e33e4b9f6ac2c4..53ccb88a0e7c6360500d091d77e8cf57d9673f78 100644 GIT binary patch delta 799 zcmaKqO-xfk5P)Z2U&||4XkT9okxHT|kXKubf+Q6S)ke}Z5G(#1Xbqy0C`vIcCk|MQ zCK~@*NAVvKdjn(Riv};I-i*eZ#v2+B9tep9E^@KXR*D)Eck^XuXJ=<;b|#r&$qdV$ z6bBMGX^7oFwU~^u46yC0 zr2lnX35&W>p%fsWuN8pn`N&2j;dN9Jup(%)=s_&p2+{$k8%INmWB{llR7_i}n6u0X zRy_{OBWwtl<!;8w-$ zKInr)!773-HH%K013xRJlP&=_xMhs066RDp1N^9}cvh{$MbDUK_sla~P!!zmc9;%8 z0JcFCc2KIlTHB_Gh^cA=F13Vku5oAIBYNydKA*or>s2B}q={<;64!|v#29gtxJBG1 z?htp0d>WoOnP?5Zqsp4Bw~s)N3@t4`iK6{`_aWg6fZ##;7cPrHOxpJFdI&F>HdP*y`Q?JX2E zL8Tvr7?%4F;^vUYI8AZph$`Mk&KD+tx5P>U@NTL}{`T_mEHuMb@WK|F0nD|!@p+?y RU7g)nV>2+&*0+64W`bTWk+R7H<0d<>T)C*#SU=L`aPHk^f%FkscQ26M(m0GE7zkH!Ix zQJk?5zVWSd7vQ|Qzy#LRMSkBc=lu&FX2@D_+uPToQr!@gNKTpJ7EI6$aD;@w6eUV9 zNnhI*%Z8UB53_Za7kk2Mn2dBi>*2_K-MFb~*cJ~`>M(Aea^hC>k!*`r8GZ>L#9u~- z%*KZW4jT#lmhQl>#|PNX(!~dIJvqqmK(3qW+GX}ch3}oEe)cKC9FFCB4^5JkBB>aG z>A!j}QD&S}=b??~S0`P=4#(cE-m<@)7fZ@0U zF^J)BL&v8^Cq5taTI=hy*4I=IR#Vrb|Fr9B-Y$rv=N4^y{Kxy#nXC9A)9$t%A}{D6 zhP~&bo>dxLoI