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

Merge branch 'master' of https://github.com/USNavalResearchLaboratory/norm

pull/9/head
bebopagogo 2020-02-29 15:11:24 -05:00
parent 2cbb93d53c 91a1301097
commit 711024bf07
1 changed files with 35 additions and 20 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

55
examples/normStreamer.cpp
View File

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