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

enabled default 'id' and 'ack auto' for normStreamer example

pull/9/head
Brian Adamson 2020-02-21 21:58:12 -05:00
parent d0df65a011
commit 91a1301097
1 changed files with 35 additions and 20 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

33
examples/normStreamer.cpp
View File

@ -124,8 +124,11 @@ class NormStreamer
NormAddAckingNode(norm_session, ackId); NormAddAckingNode(norm_session, ackId);
norm_acking = true; // invoke ack-based flow control norm_acking = true; // invoke ack-based flow control
} }
void SetAutoAck(bool enable)
{
auto_ack = enable;
norm_acking = enable;
}
bool Start(bool sender, bool receiver); bool Start(bool sender, bool receiver);
void Stop() void Stop()
@ -236,7 +239,6 @@ class NormStreamer
} }
} }
void WriteOutput(); void WriteOutput();
void WriteOutputSocket(); void WriteOutputSocket();
@ -309,6 +311,7 @@ class NormStreamer
unsigned int tx_stream_bytes_remain; unsigned int tx_stream_bytes_remain;
bool tx_watermark_pending; bool tx_watermark_pending;
bool norm_acking; bool norm_acking;
bool auto_ack;
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;
@ -361,7 +364,7 @@ NormStreamer::NormStreamer()
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), norm_acking(false), tx_ack_pending(false), flush_mode(NORM_FLUSH_ACTIVE), tx_watermark_pending(false), norm_acking(false), auto_ack(false), 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),
@ -631,6 +634,8 @@ bool NormStreamer::Start(bool sender, bool receiver)
// ack-based flow control enabled on command-line, // ack-based flow control enabled on command-line,
// so disable timer-based flow control // so disable timer-based flow control
NormSetFlowControl(norm_session, 0.0); NormSetFlowControl(norm_session, 0.0);
NormTrackingStatus trackingMode = auto_ack? NORM_TRACK_RECEIVERS : NORM_TRACK_NONE;
NormSetAutoAckingNodes(norm_session, trackingMode);
} }
// Pick a random instance id for now // Pick a random instance id for now
struct timeval currentTime; struct timeval currentTime;
@ -1401,11 +1406,12 @@ int main(int argc, char* argv[])
bool boostPriority = false; bool boostPriority = false;
unsigned int checkSequence = 0; // can set to 64 or 32 unsigned int checkSequence = 0; // can set to 64 or 32
// TBD - set these defaults to reasonable values or just use NormStreamer constructor defaults // TBD - set these defaults to reasonable values or just use NormStreamer constructor defaults
unsigned long inputSocketBufferSize = 256*1024*1024; // A zero value for socket buffers means use the operating system default sizing
unsigned long outputSocketBufferSize = 256*1024*1024; unsigned long inputSocketBufferSize = 0; // 6*1024*1024;
unsigned long txSocketBufferSize = 6*1024*1024; unsigned long outputSocketBufferSize = 0; // 6*1024*1024;
unsigned long rxSocketBufferSize = 6*1024*1024; unsigned long txSocketBufferSize = 0; // 6*1024*1024;
unsigned long streamBufferSize = 256*1024*1024; unsigned long rxSocketBufferSize = 0; // 6*1024*1024;
unsigned long streamBufferSize = 1*1024*1024;
// Instantiate a NormStreamer and set default params // Instantiate a NormStreamer and set default params
NormStreamer normStreamer; NormStreamer normStreamer;
@ -1558,6 +1564,12 @@ int main(int argc, char* argv[])
return -1; return -1;
} }
const char* alist = argv[i++]; const char* alist = argv[i++];
if (0 == strcmp(alist, "auto"))
{
normStreamer.SetAutoAck(true);
}
else
{
while ((NULL != alist) && (*alist != '\0')) while ((NULL != alist) && (*alist != '\0'))
{ {
// TBD - Do we need to skip leading white space? // TBD - Do we need to skip leading white space?
@ -1574,6 +1586,7 @@ int main(int argc, char* argv[])
if (NULL != alist) alist++; // point past comma if (NULL != alist) alist++; // point past comma
} }
} }
}
else if (0 == strncmp(cmd, "flush", len)) else if (0 == strncmp(cmd, "flush", len))
{ {
// "none", "passive", or "active" // "none", "passive", or "active"
@ -1926,12 +1939,14 @@ int main(int argc, char* argv[])
Usage(); Usage();
return -1; return -1;
} }
/*
if (NORM_NODE_NONE == nodeId) if (NORM_NODE_NONE == nodeId)
{ {
fprintf(stderr, "normStreamer error: no local 'id' provided!\n"); fprintf(stderr, "normStreamer error: no local 'id' provided!\n");
Usage(); Usage();
return -1; return -1;
} }
*/
if (boostPriority) if (boostPriority)
{ {