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

normStreamer example feature additions, code cleanup

pull/18/head
Brian Adamson 2020-06-14 11:21:19 -04:00
parent ad31a1a32a
commit 0ff199fc3c
3 changed files with 63 additions and 45 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

30
examples/normStreamer.cpp
View File

@ -122,6 +122,7 @@ class NormStreamer
{ {
assert(NORM_SESSION_INVALID != norm_session); assert(NORM_SESSION_INVALID != norm_session);
NormAddAckingNode(norm_session, ackId); NormAddAckingNode(norm_session, ackId);
acking_node_count++;
norm_acking = true; // invoke ack-based flow control norm_acking = true; // invoke ack-based flow control
} }
void SetAutoAck(bool enable) void SetAutoAck(bool enable)
@ -293,7 +294,7 @@ class NormStreamer
private: private:
NormSessionHandle norm_session; NormSessionHandle norm_session;
bool is_multicast; bool is_multicast;
bool probe_tos; UINT8 probe_tos;
bool loopback; bool loopback;
bool is_running; bool is_running;
@ -316,6 +317,7 @@ class NormStreamer
bool tx_watermark_pending; bool tx_watermark_pending;
bool norm_acking; bool norm_acking;
bool auto_ack; bool auto_ack;
unsigned int acking_node_count;
bool tx_ack_pending; bool tx_ack_pending;
NormFlushMode flush_mode; // TBD - allow for "none", "passive", "active" options NormFlushMode flush_mode; // TBD - allow for "none", "passive", "active" options
bool fti_info; bool fti_info;
@ -367,8 +369,8 @@ NormStreamer::NormStreamer()
input_socket(ProtoSocket::UDP), input_file(stdin), input_fd(fileno(stdin)), input_ready(true), input_socket(ProtoSocket::UDP), input_file(stdin), input_fd(fileno(stdin)), input_ready(true),
input_needed(false), input_msg_length(0), input_index(0), input_needed(false), input_msg_length(0), input_index(0),
tx_stream (NORM_OBJECT_INVALID), tx_ready(true), tx_stream (NORM_OBJECT_INVALID), tx_ready(true),
tx_stream_buffer_max(0), tx_stream_buffer_count(0), tx_stream_bytes_remain(0), tx_stream_buffer_max(0), tx_stream_buffer_count(0), tx_stream_bytes_remain(0), tx_watermark_pending(false),
tx_watermark_pending(false), norm_acking(false), auto_ack(false), tx_ack_pending(false), flush_mode(NORM_FLUSH_ACTIVE), 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), 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_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), 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); NormSetFlowControl(norm_session, 0.0);
NormTrackingStatus trackingMode = auto_ack? NORM_TRACK_RECEIVERS : NORM_TRACK_NONE; NormTrackingStatus trackingMode = auto_ack? NORM_TRACK_RECEIVERS : NORM_TRACK_NONE;
NormSetAutoAckingNodes(norm_session, trackingMode); 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 // Pick a random instance id for now
struct timeval currentTime; struct timeval currentTime;
@ -1160,12 +1169,23 @@ void NormStreamer::HandleNormEvent(const NormEvent& event)
//fprintf(stderr, "new GRTT = %lf\n", NormGetGrttEstimate(norm_session)); //fprintf(stderr, "new GRTT = %lf\n", NormGetGrttEstimate(norm_session));
break; 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: case NORM_TX_WATERMARK_COMPLETED:
TRACE("NORM_TX_WATERMARK_COMPLETED ...\n"); TRACE("NORM_TX_WATERMARK_COMPLETED ...\n");
if (NORM_ACK_SUCCESS == NormGetAckingStatus(norm_session)) if (NORM_ACK_SUCCESS == NormGetAckingStatus(norm_session))
{ {
//fprintf(stderr, "WATERMARK COMPLETED\n"); //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. // Flow control ack request was pending.
tx_watermark_pending = false; tx_watermark_pending = false;
@ -1473,7 +1493,7 @@ int main(int argc, char* argv[])
return -1; return -1;
} }
i++; i++;
normStreamer.SetProbeTOS(tos); normStreamer.SetProbeTOS((UINT8)tos);
} }
else if (0 == strncmp(cmd, "info", len)) else if (0 == strncmp(cmd, "info", len))
{ {

63
src/common/normNode.cpp
View File

@ -762,7 +762,7 @@ void NormSenderNode::HandleCommand(const struct timeval& currentTime,
backoffTime = 0.25 * r * maxBackoff + 0.75 * backoffTime; backoffTime = 0.25 * r * maxBackoff + 0.75 * backoffTime;
cc_timer.SetInterval(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); (unsigned long)LocalNodeId(), backoffTime);
session.ActivateTimer(cc_timer); session.ActivateTimer(cc_timer);
break; break;
@ -2393,13 +2393,16 @@ bool NormSenderNode::OnRepairTimeout(ProtoTimer& /*theTimer*/)
(unsigned long)LocalNodeId(), 8.0e-03*NormUnquantizeRate(ext.GetCCRate()), (unsigned long)LocalNodeId(), 8.0e-03*NormUnquantizeRate(ext.GetCCRate()),
rtt_estimate, ccLoss, (UINT16)nominal_packet_size, slow_start); rtt_estimate, ccLoss, (UINT16)nominal_packet_size, slow_start);
ext.SetCCSequence(cc_sequence); ext.SetCCSequence(cc_sequence);
// Cancel potential pending NORM_ACK(CC) since we are NACKing if (0 == session.GetProbeTOS()) // always send NormAck(CC) for special TOS case
if (cc_timer.IsActive())
{ {
// Set holdoff timeout to refrain from sending too much cc feedback // Cancel potential pending NORM_ACK(CC) since we are NACKing
cc_timer.SetInterval(grtt_estimate*backoff_factor); if (cc_timer.IsActive())
cc_timer.Reschedule(); {
cc_timer.DecrementRepeatCount(); // 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) } // end if (cc_enable)
@ -2991,11 +2994,14 @@ bool NormSenderNode::OnCCTimeout(ProtoTimer& /*theTimer*/)
// Build and send NORM_ACK(CC) // Build and send NORM_ACK(CC)
if (ack_pending && !ack_ex_pending && (1 == cc_timer.GetRepeatCount())) if (ack_pending && !ack_ex_pending && (1 == cc_timer.GetRepeatCount()))
{ {
// Send ACK flush right away (CC feedback is included) if (0 == session.GetProbeTOS()) // always send NormAck(CC) in special TOS case
if (ack_timer.IsActive()) ack_timer.Deactivate(); {
if (cc_timer.IsActive()) cc_timer.Deactivate(); // will be reactivated if needed // Send ACK flush right away (CC feedback is included)
OnAckTimeout(ack_timer); if (ack_timer.IsActive()) ack_timer.Deactivate();
return false; if (cc_timer.IsActive()) cc_timer.Deactivate(); // will be reactivated if needed
OnAckTimeout(ack_timer);
return false;
}
} }
switch (cc_timer.GetRepeatCount()) switch (cc_timer.GetRepeatCount())
{ {
@ -3005,7 +3011,7 @@ bool NormSenderNode::OnCCTimeout(ProtoTimer& /*theTimer*/)
case 1: 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(); NormAckMsg* ack = (NormAckMsg*)session.GetMessageFromPool();
if (!ack) if (!ack)
{ {
@ -3105,20 +3111,23 @@ bool NormSenderNode::OnAckTimeout(ProtoTimer& /*theTimer*/)
if (session.SendMessage(*ack)) if (session.SendMessage(*ack))
{ {
ack_pending = false; ack_pending = false;
cc_feedback_needed = false; if (0 == session.GetProbeTOS()) // Always send NormAck(CC) for special TOS case
if (cc_enable && !is_clr && !is_plr && session.Address().IsMulticast())
{ {
// Install cc feedback holdoff cc_feedback_needed = false;
cc_timer.SetInterval(grtt_estimate*backoff_factor); if (cc_enable && !is_clr && !is_plr && session.Address().IsMulticast())
if (cc_timer.IsActive()) {
cc_timer.Reschedule(); // Install cc feedback holdoff
else cc_timer.SetInterval(grtt_estimate*backoff_factor);
session.ActivateTimer(cc_timer); if (cc_timer.IsActive())
cc_timer.DecrementRepeatCount(); // put timer into "holdoff" phase cc_timer.Reschedule();
} else
else if (cc_timer.IsActive()) session.ActivateTimer(cc_timer);
{ cc_timer.DecrementRepeatCount(); // put timer into "holdoff" phase
cc_timer.Deactivate(); }
else if (cc_timer.IsActive())
{
cc_timer.Deactivate();
}
} }
} }
else else
@ -3726,8 +3735,6 @@ bool NormLossEstimator2::Update(const struct timeval& currentTime,
event_time = event_time_orig = currentTime; event_time = event_time_orig = currentTime;
newLossEvent = true; newLossEvent = true;
//TRACE("time>%lf new loss event fraction = %lf\n", ProtoTime(currentTime).GetValue(), LossFraction());
} }
} }
else else

15
src/common/normSession.cpp
View File

@ -426,8 +426,6 @@ bool NormSession::OpenProtoCap()
PLOG(PL_WARN, "NormSession::OpenProtoCap() warning: unable to get interface source address\n"); 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()) if (dst_addr_list.IsEmpty())
{ {
PLOG(PL_FATAL, "NormSession::OpenProtoCap() error: unable to add any addresses to dst_addr_list!!\n"); 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()); NormObject *oldest = tx_table.Find(tx_table.RangeLo());
if (oldest->IsRepairPending() || oldest->IsPending()) 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()); oldest->IsRepairPending(), repair_timer.IsActive(), oldest->IsPending());
posted_tx_queue_empty = false; posted_tx_queue_empty = false;
return false; return false;
@ -2075,7 +2073,7 @@ bool NormSession::QueueTxObject(NormObject *obj)
} }
else 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 // TBD - flow control as should allow for TX_QUEUE_VACANCY posting for session
ActivateFlowControl(delay, oldest->GetId(), NormController::TX_QUEUE_EMPTY); ActivateFlowControl(delay, oldest->GetId(), NormController::TX_QUEUE_EMPTY);
posted_tx_queue_empty = false; posted_tx_queue_empty = false;
@ -2390,7 +2388,6 @@ void NormSession::RxSocketRecvHandler(ProtoSocket &theSocket,
bool ecnStatus = false; bool ecnStatus = false;
#ifdef SIMULATE #ifdef SIMULATE
ecnStatus = theSocket.GetEcnStatus(); ecnStatus = theSocket.GetEcnStatus();
//if (ecnStatus) TRACE("NORM RECEIVED PACKET W/ ECN BIT SET!!!!!\n");
#endif // SIMULATE #endif // SIMULATE
bool wasUnicast; bool wasUnicast;
if (destAddr.IsValid()) if (destAddr.IsValid())
@ -3042,7 +3039,6 @@ void NormSession::ReceiverHandleObjectMessage(const struct timeval &currentTime,
Notify(NormController::REMOTE_SENDER_ADDRESS, theSender, NULL); Notify(NormController::REMOTE_SENDER_ADDRESS, theSender, NULL);
} }
theSender->UpdateRecvRate(currentTime, msg.GetLength()); theSender->UpdateRecvRate(currentTime, msg.GetLength());
//if (ecnStatus) TRACE("UPDATING LOSS EST w/ ECN pkt\n");
theSender->UpdateLossEstimate(currentTime, msg.GetSequence(), ecnStatus); theSender->UpdateLossEstimate(currentTime, msg.GetSequence(), ecnStatus);
theSender->IncrementRecvTotal(msg.GetLength()); // for statistics only (TBD) #ifdef NORM_DEBUG theSender->IncrementRecvTotal(msg.GetLength()); // for statistics only (TBD) #ifdef NORM_DEBUG
theSender->HandleObjectMessage(msg); theSender->HandleObjectMessage(msg);
@ -4830,14 +4826,12 @@ bool NormSession::OnTxTimeout(ProtoTimer & /*theTimer*/)
message_queue.Prepend(msg); message_queue.Prepend(msg);
if (tx_timer.IsActive()) if (tx_timer.IsActive())
tx_timer.Deactivate(); tx_timer.Deactivate();
//TRACE("starting output notification ...\n");
tx_socket->StartOutputNotification(); tx_socket->StartOutputNotification();
return false; // since timer was deactivated return false; // since timer was deactivated
case MSG_SEND_FAILED: case MSG_SEND_FAILED:
// Message was not sent due to socket error (no route, etc), so so just timeout and try again // 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) // (TBD - is there something smarter we should do)
//TRACE("MSG_SEND_FAILED\n");
if (!advertise_repairs) if (!advertise_repairs)
message_queue.Prepend(msg); message_queue.Prepend(msg);
if (tx_rate > 0.0) if (tx_rate > 0.0)
@ -4874,7 +4868,6 @@ bool NormSession::OnTxTimeout(ProtoTimer & /*theTimer*/)
NormSession::MessageStatus NormSession::SendMessage(NormMsg &msg) NormSession::MessageStatus NormSession::SendMessage(NormMsg &msg)
{ {
//TRACE("sending message length %hu\n", msg.GetLength());
bool isReceiverMsg = false; bool isReceiverMsg = false;
bool isProbe = false; bool isProbe = false;
bool sendRaw = false; bool sendRaw = false;
@ -4967,13 +4960,13 @@ NormSession::MessageStatus NormSession::SendMessage(NormMsg &msg)
struct timeval currentTime; struct timeval currentTime;
ProtoSystemTime(currentTime); ProtoSystemTime(currentTime);
theSender->CalculateGrttResponse(currentTime, grttResponse); theSender->CalculateGrttResponse(currentTime, grttResponse);
ack.SetGrttResponse(grttResponse);
if (0 != probe_tos) sendRaw = true; if (0 != probe_tos) sendRaw = true;
} }
else else
{ {
grttResponse.tv_sec = grttResponse.tv_usec = 0; grttResponse.tv_sec = grttResponse.tv_usec = 0;
} }
ack.SetGrttResponse(grttResponse);
break; break;
} }
default: default:
@ -5118,7 +5111,6 @@ bool NormSession::RawSendTo(const char* buffer, unsigned int& numBytes, const Pr
trafficClass &= ~((UINT8)ProtoSocket::ECN_ECT1); // clear ECT1 bit 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()) switch (dstAddr.GetType())
{ {
case ProtoAddress::IPv4: case ProtoAddress::IPv4:
@ -5570,7 +5562,6 @@ void NormSession::AdjustRate(bool onResponse)
{ {
// reduce rate if no active data transmission // reduce rate if no active data transmission
// (TBD) Perhaps we want to be less aggressive here someday // (TBD) Perhaps we want to be less aggressive here someday
//TRACE("DATA INACTIVITY!\n");
txRate *= 0.5; txRate *= 0.5;
} }
else if (clr && clr->IsActive()) else if (clr && clr->IsActive())