From 0ff199fc3c1579ff47532f1cb62933f771eeea84 Mon Sep 17 00:00:00 2001 From: Brian Adamson Date: Sun, 14 Jun 2020 11:21:19 -0400 Subject: [PATCH] normStreamer example feature additions, code cleanup --- examples/normStreamer.cpp | 30 +++++++++++++++--- src/common/normNode.cpp | 63 +++++++++++++++++++++----------------- src/common/normSession.cpp | 15 ++------- 3 files changed, 63 insertions(+), 45 deletions(-) diff --git a/examples/normStreamer.cpp b/examples/normStreamer.cpp index 76248b0..62a8324 100644 --- a/examples/normStreamer.cpp +++ b/examples/normStreamer.cpp @@ -122,6 +122,7 @@ class NormStreamer { assert(NORM_SESSION_INVALID != norm_session); NormAddAckingNode(norm_session, ackId); + acking_node_count++; norm_acking = true; // invoke ack-based flow control } void SetAutoAck(bool enable) @@ -293,7 +294,7 @@ class NormStreamer private: NormSessionHandle norm_session; bool is_multicast; - bool probe_tos; + UINT8 probe_tos; bool loopback; bool is_running; @@ -316,6 +317,7 @@ class NormStreamer bool tx_watermark_pending; bool norm_acking; bool auto_ack; + unsigned int acking_node_count; bool tx_ack_pending; NormFlushMode flush_mode; // TBD - allow for "none", "passive", "active" options bool fti_info; @@ -367,8 +369,8 @@ NormStreamer::NormStreamer() input_socket(ProtoSocket::UDP), input_file(stdin), input_fd(fileno(stdin)), input_ready(true), input_needed(false), input_msg_length(0), input_index(0), tx_stream (NORM_OBJECT_INVALID), tx_ready(true), - tx_stream_buffer_max(0), tx_stream_buffer_count(0), tx_stream_bytes_remain(0), - tx_watermark_pending(false), norm_acking(false), auto_ack(false), tx_ack_pending(false), flush_mode(NORM_FLUSH_ACTIVE), + tx_stream_buffer_max(0), tx_stream_buffer_count(0), tx_stream_bytes_remain(0), tx_watermark_pending(false), + norm_acking(false), auto_ack(false), acking_node_count(0), tx_ack_pending(false), flush_mode(NORM_FLUSH_ACTIVE), fti_info(false), ack_ex(false), rx_stream(NORM_OBJECT_INVALID), rx_ready(false), rx_needed(false), msg_sync(false), output_bucket_rate(0.0), output_bucket_interval(0.0), output_bucket_depth(0), output_bucket_count(0), output_socket(ProtoSocket::UDP), output_file(stdout), output_fd(fileno(stdout)), output_ready(true), @@ -645,6 +647,13 @@ bool NormStreamer::Start(bool sender, bool receiver) NormSetFlowControl(norm_session, 0.0); NormTrackingStatus trackingMode = auto_ack? NORM_TRACK_RECEIVERS : NORM_TRACK_NONE; NormSetAutoAckingNodes(norm_session, trackingMode); + if (auto_ack && (0 == acking_node_count)) + { + // This allows for the receivrer(s) to start after the sender + // as the sender will persistently send ack requests until + // a receiver responds. + NormAddAckingNode(norm_session, NORM_NODE_NONE); + } } // Pick a random instance id for now struct timeval currentTime; @@ -1160,12 +1169,23 @@ void NormStreamer::HandleNormEvent(const NormEvent& event) //fprintf(stderr, "new GRTT = %lf\n", NormGetGrttEstimate(norm_session)); break; + case NORM_ACKING_NODE_NEW: + if (0 == acking_node_count) + NormRemoveAckingNode(event.session, NORM_NODE_NONE); + acking_node_count++; + break; + case NORM_TX_WATERMARK_COMPLETED: TRACE("NORM_TX_WATERMARK_COMPLETED ...\n"); if (NORM_ACK_SUCCESS == NormGetAckingStatus(norm_session)) { //fprintf(stderr, "WATERMARK COMPLETED\n"); - if (tx_watermark_pending) + if (0 == acking_node_count) + { + // Keep probing until some receiver shows up + NormResetWatermark(norm_session); + } + else if (tx_watermark_pending) { // Flow control ack request was pending. tx_watermark_pending = false; @@ -1473,7 +1493,7 @@ int main(int argc, char* argv[]) return -1; } i++; - normStreamer.SetProbeTOS(tos); + normStreamer.SetProbeTOS((UINT8)tos); } else if (0 == strncmp(cmd, "info", len)) { diff --git a/src/common/normNode.cpp b/src/common/normNode.cpp index b83b85c..719cbb6 100644 --- a/src/common/normNode.cpp +++ b/src/common/normNode.cpp @@ -762,7 +762,7 @@ void NormSenderNode::HandleCommand(const struct timeval& currentTime, backoffTime = 0.25 * r * maxBackoff + 0.75 * backoffTime; cc_timer.SetInterval(backoffTime); - PLOG(PL_TRACE, "NormSenderNode::HandleCommand() node>%lu begin CC back-off: %lf sec)...\n", + PLOG(PL_DEBUG, "NormSenderNode::HandleCommand() node>%lu begin CC back-off: %lf sec)...\n", (unsigned long)LocalNodeId(), backoffTime); session.ActivateTimer(cc_timer); break; @@ -2393,13 +2393,16 @@ bool NormSenderNode::OnRepairTimeout(ProtoTimer& /*theTimer*/) (unsigned long)LocalNodeId(), 8.0e-03*NormUnquantizeRate(ext.GetCCRate()), rtt_estimate, ccLoss, (UINT16)nominal_packet_size, slow_start); ext.SetCCSequence(cc_sequence); - // Cancel potential pending NORM_ACK(CC) since we are NACKing - if (cc_timer.IsActive()) + if (0 == session.GetProbeTOS()) // always send NormAck(CC) for special TOS case { - // Set holdoff timeout to refrain from sending too much cc feedback - cc_timer.SetInterval(grtt_estimate*backoff_factor); - cc_timer.Reschedule(); - cc_timer.DecrementRepeatCount(); + // Cancel potential pending NORM_ACK(CC) since we are NACKing + if (cc_timer.IsActive()) + { + // Set holdoff timeout to refrain from sending too much cc feedback + cc_timer.SetInterval(grtt_estimate*backoff_factor); + cc_timer.Reschedule(); + cc_timer.DecrementRepeatCount(); + } } } // end if (cc_enable) @@ -2991,11 +2994,14 @@ bool NormSenderNode::OnCCTimeout(ProtoTimer& /*theTimer*/) // Build and send NORM_ACK(CC) if (ack_pending && !ack_ex_pending && (1 == cc_timer.GetRepeatCount())) { - // Send ACK flush right away (CC feedback is included) - if (ack_timer.IsActive()) ack_timer.Deactivate(); - if (cc_timer.IsActive()) cc_timer.Deactivate(); // will be reactivated if needed - OnAckTimeout(ack_timer); - return false; + if (0 == session.GetProbeTOS()) // always send NormAck(CC) in special TOS case + { + // Send ACK flush right away (CC feedback is included) + if (ack_timer.IsActive()) ack_timer.Deactivate(); + if (cc_timer.IsActive()) cc_timer.Deactivate(); // will be reactivated if needed + OnAckTimeout(ack_timer); + return false; + } } switch (cc_timer.GetRepeatCount()) { @@ -3005,7 +3011,7 @@ bool NormSenderNode::OnCCTimeout(ProtoTimer& /*theTimer*/) case 1: { - // We weren't suppressed, so build an ACK(RTT) and send + // We weren't suppressed, so build an ACK(CC) and send NormAckMsg* ack = (NormAckMsg*)session.GetMessageFromPool(); if (!ack) { @@ -3105,20 +3111,23 @@ bool NormSenderNode::OnAckTimeout(ProtoTimer& /*theTimer*/) if (session.SendMessage(*ack)) { ack_pending = false; - cc_feedback_needed = false; - if (cc_enable && !is_clr && !is_plr && session.Address().IsMulticast()) + if (0 == session.GetProbeTOS()) // Always send NormAck(CC) for special TOS case { - // Install cc feedback holdoff - cc_timer.SetInterval(grtt_estimate*backoff_factor); - if (cc_timer.IsActive()) - cc_timer.Reschedule(); - else - session.ActivateTimer(cc_timer); - cc_timer.DecrementRepeatCount(); // put timer into "holdoff" phase - } - else if (cc_timer.IsActive()) - { - cc_timer.Deactivate(); + cc_feedback_needed = false; + if (cc_enable && !is_clr && !is_plr && session.Address().IsMulticast()) + { + // Install cc feedback holdoff + cc_timer.SetInterval(grtt_estimate*backoff_factor); + if (cc_timer.IsActive()) + cc_timer.Reschedule(); + else + session.ActivateTimer(cc_timer); + cc_timer.DecrementRepeatCount(); // put timer into "holdoff" phase + } + else if (cc_timer.IsActive()) + { + cc_timer.Deactivate(); + } } } else @@ -3726,8 +3735,6 @@ bool NormLossEstimator2::Update(const struct timeval& currentTime, event_time = event_time_orig = currentTime; newLossEvent = true; - //TRACE("time>%lf new loss event fraction = %lf\n", ProtoTime(currentTime).GetValue(), LossFraction()); - } } else diff --git a/src/common/normSession.cpp b/src/common/normSession.cpp index e715f1a..7524786 100644 --- a/src/common/normSession.cpp +++ b/src/common/normSession.cpp @@ -426,8 +426,6 @@ bool NormSession::OpenProtoCap() PLOG(PL_WARN, "NormSession::OpenProtoCap() warning: unable to get interface source address\n"); } - TRACE("got source address: %s\n", src_addr.GetHostString()); - if (dst_addr_list.IsEmpty()) { PLOG(PL_FATAL, "NormSession::OpenProtoCap() error: unable to add any addresses to dst_addr_list!!\n"); @@ -2059,7 +2057,7 @@ bool NormSession::QueueTxObject(NormObject *obj) NormObject *oldest = tx_table.Find(tx_table.RangeLo()); if (oldest->IsRepairPending() || oldest->IsPending()) { - PLOG(PL_ALWAYS, "NormSession::QueueTxObject() all held objects repair pending:%d (repair active:%d) pending:%d\n", + PLOG(PL_DEBUG, "NormSession::QueueTxObject() all held objects repair pending:%d (repair active:%d) pending:%d\n", oldest->IsRepairPending(), repair_timer.IsActive(), oldest->IsPending()); posted_tx_queue_empty = false; return false; @@ -2075,7 +2073,7 @@ bool NormSession::QueueTxObject(NormObject *obj) } else { - PLOG(PL_ALWAYS, "NormSession::QueueTxObject() asserting flow control for object delay:%lf sec\n", delay); + PLOG(PL_DEBUG, "NormSession::QueueTxObject() asserting flow control for object delay:%lf sec\n", delay); // TBD - flow control as should allow for TX_QUEUE_VACANCY posting for session ActivateFlowControl(delay, oldest->GetId(), NormController::TX_QUEUE_EMPTY); posted_tx_queue_empty = false; @@ -2390,7 +2388,6 @@ void NormSession::RxSocketRecvHandler(ProtoSocket &theSocket, bool ecnStatus = false; #ifdef SIMULATE ecnStatus = theSocket.GetEcnStatus(); - //if (ecnStatus) TRACE("NORM RECEIVED PACKET W/ ECN BIT SET!!!!!\n"); #endif // SIMULATE bool wasUnicast; if (destAddr.IsValid()) @@ -3042,7 +3039,6 @@ void NormSession::ReceiverHandleObjectMessage(const struct timeval ¤tTime, Notify(NormController::REMOTE_SENDER_ADDRESS, theSender, NULL); } theSender->UpdateRecvRate(currentTime, msg.GetLength()); - //if (ecnStatus) TRACE("UPDATING LOSS EST w/ ECN pkt\n"); theSender->UpdateLossEstimate(currentTime, msg.GetSequence(), ecnStatus); theSender->IncrementRecvTotal(msg.GetLength()); // for statistics only (TBD) #ifdef NORM_DEBUG theSender->HandleObjectMessage(msg); @@ -4830,14 +4826,12 @@ bool NormSession::OnTxTimeout(ProtoTimer & /*theTimer*/) message_queue.Prepend(msg); if (tx_timer.IsActive()) tx_timer.Deactivate(); - //TRACE("starting output notification ...\n"); tx_socket->StartOutputNotification(); return false; // since timer was deactivated case MSG_SEND_FAILED: // Message was not sent due to socket error (no route, etc), so so just timeout and try again // (TBD - is there something smarter we should do) - //TRACE("MSG_SEND_FAILED\n"); if (!advertise_repairs) message_queue.Prepend(msg); if (tx_rate > 0.0) @@ -4874,7 +4868,6 @@ bool NormSession::OnTxTimeout(ProtoTimer & /*theTimer*/) NormSession::MessageStatus NormSession::SendMessage(NormMsg &msg) { - //TRACE("sending message length %hu\n", msg.GetLength()); bool isReceiverMsg = false; bool isProbe = false; bool sendRaw = false; @@ -4967,13 +4960,13 @@ NormSession::MessageStatus NormSession::SendMessage(NormMsg &msg) struct timeval currentTime; ProtoSystemTime(currentTime); theSender->CalculateGrttResponse(currentTime, grttResponse); - ack.SetGrttResponse(grttResponse); if (0 != probe_tos) sendRaw = true; } else { grttResponse.tv_sec = grttResponse.tv_usec = 0; } + ack.SetGrttResponse(grttResponse); break; } default: @@ -5118,7 +5111,6 @@ bool NormSession::RawSendTo(const char* buffer, unsigned int& numBytes, const Pr trafficClass &= ~((UINT8)ProtoSocket::ECN_ECT1); // clear ECT1 bit } - //TRACE("sending RAW with tos 0x%02x (probe_tos 0x%02x\n", trafficClass, probe_tos); switch (dstAddr.GetType()) { case ProtoAddress::IPv4: @@ -5570,7 +5562,6 @@ void NormSession::AdjustRate(bool onResponse) { // reduce rate if no active data transmission // (TBD) Perhaps we want to be less aggressive here someday - //TRACE("DATA INACTIVITY!\n"); txRate *= 0.5; } else if (clr && clr->IsActive())