diff --git a/examples/normStreamer.cpp b/examples/normStreamer.cpp index 81c2604..3fe0cc1 100644 --- a/examples/normStreamer.cpp +++ b/examples/normStreamer.cpp @@ -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) {