#include "normApi.h" #include "protoSocket.h" #include // for printf(), etc #include // for srand() #include // for strrchr(), memset(), etc #include // for gettimeofday() #include // for htons() #include // for, well, fnctl() #include // obvious child #include // embarrassingly obvious #include // Memory Lock. #include // Adjust scheduler (linux) #include // for setpriority() stuff #ifdef LINUX #include #endif // LINUX const unsigned int LOOP_MAX = 100; // Setting SHOOT_FIRST to non-zero means that an ACK request // will be used to advance the acking "watermark" point // with each message fully written to the transmit stream. // The alternative "ack later" behavior waits to send a new // ACK request until any pending flow control ACK requeset // has completed. This latter approach favors throughput // over timeliness of message delivery. I.e., lower data // rate applications that are concerned with low-latency message // delivery can potentially benefit from the "shoot first" // behavior while very high throughput applications that want // to "keep the pipe full as possible" can benefit from the // "ack later" behavior. The difference between these behaviors, // since ACK requests are cued for all messages when flow // control is _not_ pending, is somewhat subtle and developers // may want to assess both behaviors for their application. // Additionally, limiting ACK request to flow control only is // another possible approach as well as dynamically updating // something like the "tx_stream_buffer_count" with each // message ACK request initiated could be possible. The caution // with the SHOOT_FIRST type strategies and high throughput is // the application may end up "chasing" the ACK request until // flow control buffer limits are reached and end up with // "dead air" time. There are always tradeoffs! //#define SHOOT_FIRST 0 class NormStreamer { public: NormStreamer(); ~NormStreamer(); // some day build these directly into NORM API enum CCMode {NORM_FIXED, NORM_CC, NORM_CCE, NORM_CCL}; enum { MSG_HEADER_SIZE = 2, // Big Endian message length header size MSG_SIZE_MAX = 65535 // (including length header) }; void SetOutputFile(FILE* filePtr) { output_file = filePtr; output_fd = fileno(filePtr); } void SetLoopback(bool state) { loopback = state; if (NORM_SESSION_INVALID != norm_session) { NormSetMulticastLoopback(norm_session, state); NormSetLoopback(norm_session, state); // XXX test code } } void SetFtiInfo(bool state) { fti_info = state; if (NORM_SESSION_INVALID != norm_session) NormLimitObjectInfo(norm_session, state); } void SetAckEx(bool state) {ack_ex = state;} bool EnableUdpRelay(const char* relayAddr, unsigned short relayPort); bool EnableUdpListener(unsigned short thePort, const char* groupAddr, const char * interfaceName); bool UdpListenerEnabled() const {return input_socket.IsOpen();} bool UdpRelayEnabled() const {return output_socket.IsOpen();} int GetInputDescriptor() const {return (input_socket.IsOpen() ? input_socket.GetHandle() : fileno(input_file));} int GetOutputDescriptor() const {return (output_socket.IsOpen() ? output_socket.GetHandle() : fileno(output_file));} bool OpenNormSession(NormInstanceHandle instance, const char* addr, unsigned short port, NormNodeId nodeId); void CloseNormSession(); void SetNormCongestionControl(CCMode ccMode); void SetFlushMode(NormFlushMode flushMode) {flush_mode = flushMode;} void SetNormTxRate(double bitsPerSecond) { assert(NORM_SESSION_INVALID != norm_session); NormSetTxRate(norm_session, bitsPerSecond); } void SetNormMulticastInterface(const char* ifaceName) { assert(NORM_SESSION_INVALID != norm_session); NormSetMulticastInterface(norm_session, ifaceName); } void SetNormMessageTrace(bool state) { assert(NORM_SESSION_INVALID != norm_session); NormSetMessageTrace(norm_session, state); } void AddAckingNode(NormNodeId ackId) { 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) { auto_ack = enable; norm_acking = enable; } bool Start(bool sender, bool receiver); void Stop() {is_running = false;} bool IsRunning() const {return is_running;} void HandleNormEvent(const NormEvent& event); // Sender methods int GetInputFile() const {return input_fd;} void SetInputReady() {input_ready = true;} bool InputReady() const {return input_ready;} bool InputNeeded() const {return input_needed;} void ReadInput(); void ReadInputSocket(); bool TxPending() const {return (!input_needed && (input_index < input_msg_length));} bool TxReady() const {return (tx_ready && (!norm_acking || (tx_stream_buffer_count < tx_stream_buffer_max)));} void SendData(); unsigned int WriteToStream(const char* buffer, unsigned int numBytes); void FlushStream(bool eom, NormFlushMode flushMode); // Receiver methods bool RxNeeded() const {return rx_needed;} bool RxReady() const {return rx_ready;} void RecvData(); int GetOutputFile() const {return output_fd;} void SetOutputReady() {output_ready = true;} bool OutputReady() const {return output_ready;} bool OutputPending() const {return (!rx_needed && (output_index < output_msg_length));} void SetOutputBucketRate(double bitsPerSecond) { output_bucket_rate = bitsPerSecond / 8.0; // convert to bytes per second output_bucket_interval = 1.0 / output_bucket_rate; } void SetOutputBucketDepth(unsigned int numBytes) {output_bucket_depth = numBytes;} unsigned int GetOutputBucketDepth() const {return output_bucket_depth;} double GetOutputBucketTimeout() const { if (0 != output_bucket_depth) { if (OutputPending()) { unsigned int pendingBytes = output_msg_length - output_index; if (pendingBytes > output_bucket_count) { return ((double)(pendingBytes - output_bucket_count)) * output_bucket_interval; } else { return 0.0; } } else { return -1.0; } } else { return 0.0; } } double GetOutputBucketFillTime() const { return (output_bucket_count < output_bucket_depth) ? ((double)(output_bucket_depth - output_bucket_count)) * output_bucket_interval : 0.0; } bool OutputBucketReady() const { if (0 != output_bucket_depth) { unsigned int pendingBytes = output_msg_length - output_index; return (output_bucket_count >= pendingBytes); } else { return true; } } void CreditOutputBucket(double interval) { if (0 != output_bucket_depth) { output_bucket_count += interval * output_bucket_rate; if (output_bucket_count > output_bucket_depth) output_bucket_count = output_bucket_depth; } } void WriteOutput(); void WriteOutputSocket(); void OmitHeader(bool state) {omit_header = state;} unsigned long GetInputByteCount() const {return input_byte_count;} unsigned long GetTxByteCount() const {return tx_byte_count;} // These can only be called post-OpenNormSession() void SetSilentReceiver(bool state) {NormSetSilentReceiver(norm_session, state);} void SetTxLoss(double txloss) {NormSetTxLoss(norm_session, txloss);} void SetRxLoss(double rxloss) {NormSetRxLoss(norm_session, rxloss);} // Set the scheduler for running the app and norm threads. static bool BoostPriority(); void SetSegmentSize(unsigned short segmentSize) {segment_size = segmentSize;} void SetBlockSize(unsigned short blockSize) {block_size = blockSize;} void SetNumParity(unsigned short numParity) {num_parity = numParity;} void SetAutoParity(unsigned short autoParity) {auto_parity = autoParity;} void SetStreamBufferSize(unsigned int value) {stream_buffer_size = value;} void SetTxSocketBufferSize(unsigned int value) {tx_socket_buffer_size = value;} void SetRxSocketBufferSize(unsigned int value) {rx_socket_buffer_size = value;} void SetInputSocketBufferSize(unsigned int value) {input_socket_buffer_size = value;} void SetOutputSocketBufferSize(unsigned int value) {output_socket_buffer_size = value;} void SetProbeTOS(UINT8 value) {probe_tos = value;} // Check that sequence numbers increase by one each time. // Assumes that sequence number is 8- or 4-byte network-order first 8 bytes of buffer. void CheckSequenceNumber(const char* buffer, const char* source); void CheckSequenceNumber64(const char* buffer, const char* source); void CheckSequenceNumber32(const char* buffer, const char* source); void SetCheckSequence(unsigned int value) // 64 or 32 {check_sequence = value;} private: NormSessionHandle norm_session; bool is_multicast; UINT8 probe_tos; bool loopback; bool is_running; // State variables for reading input messages for transmission ProtoSocket input_socket; // optional UDP socket to "listen" FILE* input_file; int input_fd; // stdin by default bool input_ready; bool input_needed; char input_buffer[MSG_SIZE_MAX]; unsigned int input_msg_length; unsigned int input_index; NormObjectHandle tx_stream; bool tx_ready; unsigned int tx_stream_buffer_max; unsigned int tx_stream_buffer_threshold; // flow control threshold unsigned int tx_stream_buffer_count; unsigned int tx_stream_bytes_remain; 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; bool ack_ex; // Receive stream and state variables for writing received messages to output NormObjectHandle rx_stream; bool rx_ready; bool rx_needed; bool msg_sync; double output_bucket_rate; // bytes per second double output_bucket_interval; // seconds per byte unsigned int output_bucket_depth; // bytes unsigned int output_bucket_count; // bytes ProtoSocket output_socket; // optional UDP socket for recv msg output ProtoAddress relay_addr; // dest addr for recv msg relay FILE* output_file; int output_fd; // stdout by default bool output_ready; char output_buffer[MSG_SIZE_MAX]; unsigned int output_msg_length; unsigned int output_index; // These are some options mainly for testing purposes bool omit_header; // if "true", receive message length header is _not_ written to output bool rx_silent; //double tx_loss; unsigned long input_byte_count; unsigned long tx_byte_count; unsigned short segment_size; unsigned short block_size; unsigned short num_parity; unsigned short auto_parity; unsigned long stream_buffer_size; unsigned int tx_socket_buffer_size; unsigned int rx_socket_buffer_size; unsigned int input_socket_buffer_size; unsigned int output_socket_buffer_size; unsigned int check_sequence; uint64_t sequence_prev; }; // end class NormStreamer NormStreamer::NormStreamer() : norm_session(NORM_SESSION_INVALID), is_multicast(false), probe_tos(0), loopback(false), is_running(false), 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), 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), output_msg_length(0), output_index(0), omit_header(false), rx_silent(false), input_byte_count(0), tx_byte_count(0), segment_size(1398), block_size(64), num_parity(0), auto_parity(0), stream_buffer_size(2*1024*1024), tx_socket_buffer_size(0), rx_socket_buffer_size(0), input_socket_buffer_size(0), output_socket_buffer_size(0), check_sequence(0), sequence_prev(0) { } NormStreamer::~NormStreamer() { } bool NormStreamer::BoostPriority() { #ifdef LINUX pid_t this_process = getpid() ; int policy = SCHED_FIFO ; int max_priority = sched_get_priority_max(policy) ; struct sched_param schedule_parameters ; memset((void*)&schedule_parameters, 0, sizeof(schedule_parameters)) ; schedule_parameters.sched_priority = max_priority ; int status = sched_setscheduler(this_process, policy, &schedule_parameters) ; if (0 != status) { fprintf(stderr, "%s:=>sched_setscheduler failed (%d), %s\n", __PRETTY_FUNCTION__, errno, strerror(errno) ) ; return false ; } else { fprintf(stderr, "%s:=>sched_setscheduler set priority to %d for process %u \n", __PRETTY_FUNCTION__, max_priority, this_process ) ; } #else // (TBD) Do something differently if "pthread sched param"? if (0 != setpriority(PRIO_PROCESS, getpid(), -20)) { PLOG(PL_ERROR, "NormStreamer::BoostPriority() error: setpriority() error: %s\n", GetErrorString()); return false; } #endif // if/else LINUX return true; } #ifndef ntohll //Convert net-order to host-order. uint64_t ntohll(uint64_t value) { static const int betest = 1 ; union MyUnion { uint64_t i64; uint32_t i32[2]; }; uint64_t rval = value; bool host_is_little_endian = ( 1 == (int)(*(char*)&betest) ) ; if ( host_is_little_endian ) { MyUnion u; u.i64 = value; uint32_t temp = u.i32[0]; u.i32[0] = ntohl(u.i32[1]); u.i32[1] = ntohl(temp); rval = u.i64; } return rval ; } #endif // !nothll void NormStreamer::CheckSequenceNumber64(const char* buffer, const char* source) { uint64_t temp; memcpy((void*)&temp, (void*)buffer, sizeof(temp)); uint64_t sequence = ntohll(temp); if (0 != sequence_prev) { int64_t delta = (int64_t)(sequence - sequence_prev); if (1 != delta) { fprintf(stderr, "normStreamer: %s dropped %lu packets seq:%lu seq_prev:%lu\n", source, (unsigned long)delta, (unsigned long)sequence, (unsigned long)sequence_prev); } } sequence_prev = sequence; } // end NormStreamer::CheckSequenceNumber64() void NormStreamer::CheckSequenceNumber32(const char* buffer, const char* source) { uint32_t temp; memcpy((void*)&temp, (void*)buffer, sizeof(temp)); uint32_t sequence = ntohll(temp); if (0 != sequence_prev) { int32_t delta = (int32_t)(sequence - sequence_prev); if (1 != delta) { fprintf(stderr, "normStreamer: %s dropped %lu packets seq:%lu seq_prev:%lu\n", source, (unsigned long)delta, (unsigned long)sequence, (unsigned long)sequence_prev); } } sequence_prev = sequence; } // end NormStreamer::CheckSequenceNumber32() void NormStreamer::CheckSequenceNumber(const char* buffer, const char* source) { switch (check_sequence) { case 32: CheckSequenceNumber32(buffer, source); break; case 64: CheckSequenceNumber64(buffer, source) ; break; default: break; } } // end NormStreamer::CheckSequenceNumber() bool NormStreamer::EnableUdpRelay(const char* relayAddr, unsigned short relayPort) { if (!output_socket.Open()) { fprintf(stderr, "normStreamer error: unable to open 'relay' socket\n"); return false ; } if (!output_socket.SetTxBufferSize(output_socket_buffer_size)) { fprintf(stderr, "normStreamer warning: unable to set desired 'relay' socket buffer size (retrieved value:%u)\n", output_socket.GetTxBufferSize()); } if (!relay_addr.ResolveFromString(relayAddr)) { fprintf(stderr, "normStreamer error: invalid relay address\n"); output_socket.Close(); return false; } relay_addr.SetPort(relayPort); // TBD - validate port number?? return true; } // end bool EnableUdpRelay() bool NormStreamer::EnableUdpListener(unsigned short thePort, const char* groupAddr, const char * interfaceName) { if (!input_socket.Open(thePort)) { fprintf(stderr, "normStreamer error: unable to open 'listen' socket on port %hu\n", thePort); return false; } if (!input_socket.SetRxBufferSize(input_socket_buffer_size)) { fprintf(stderr, "normStreamer error: unable to set desired 'listen' socket buffer size\n"); return false; } if (NULL != groupAddr) { ProtoAddress addr; if (!addr.ResolveFromString(groupAddr) || (!addr.IsMulticast())) { fprintf(stderr, "normStreamer error: invalid 'listen' group address\n"); input_socket.Close(); return false ; } if (!input_socket.JoinGroup(addr, interfaceName)) { fprintf(stderr, "normStreamer error: unable to join 'listen' group address\n"); input_socket.Close(); return false; } } return true; } // end NormStreamer::EnableUdpListener() bool NormStreamer::OpenNormSession(NormInstanceHandle instance, const char* addr, unsigned short port, NormNodeId nodeId) { if (NormIsUnicastAddress(addr)) is_multicast = false; else is_multicast = true; norm_session = NormCreateSession(instance, addr, port, nodeId); if (NORM_SESSION_INVALID == norm_session) { fprintf(stderr, "normStreamer error: unable to create NORM session\n"); return false; } if (is_multicast) { NormSetRxPortReuse(norm_session, true); if (loopback) { NormSetMulticastLoopback(norm_session, true); } } NormSetLoopback(norm_session, loopback); // Set some default parameters (maybe we should put parameter setting in Start()) NormSetDefaultSyncPolicy(norm_session, NORM_SYNC_STREAM); if (!is_multicast) NormSetDefaultUnicastNack(norm_session, true); NormSetTxRobustFactor(norm_session, 20); NormSetGrttProbingTOS(norm_session, probe_tos); NormSetFragmentation(norm_session, true); // so that IP ID gets set for SMF DPD return true; } // end NormStreamer::OpenNormSession() void NormStreamer::CloseNormSession() { if (NORM_SESSION_INVALID == norm_session) return; NormDestroySession(norm_session); norm_session = NORM_SESSION_INVALID; } // end NormStreamer::CloseNormSession() void NormStreamer::SetNormCongestionControl(CCMode ccMode) { assert(NORM_SESSION_INVALID != norm_session); switch (ccMode) { case NORM_CC: // default TCP-friendly congestion control NormSetEcnSupport(norm_session, false, false, false); break; case NORM_CCE: // "wireless-ready" ECN-only congestion control NormSetEcnSupport(norm_session, true, true); break; case NORM_CCL: // "loss tolerant", non-ECN congestion control NormSetEcnSupport(norm_session, false, false, true); break; case NORM_FIXED: // "fixed" constant data rate NormSetEcnSupport(norm_session, false, false, false); break; } if (NORM_FIXED != ccMode) NormSetCongestionControl(norm_session, true); else NormSetCongestionControl(norm_session, false); } // end NormStreamer::SetNormCongestionControl() bool NormStreamer::Start(bool sender, bool receiver) { // Note the session NORM buffer size is set the same s stream_buffer_size unsigned int bufferSize = stream_buffer_size; if (receiver) { if (!NormPreallocateRemoteSender(norm_session, bufferSize, segment_size, block_size, num_parity, stream_buffer_size)) fprintf(stderr, "normStreamer warning: unable to preallocate remote sender\n"); fprintf(stderr, "normStreamer: receiver ready.\n"); if (!NormStartReceiver(norm_session, bufferSize)) { fprintf(stderr, "normStreamer error: unable to start NORM receiver\n"); return false; } if (0 != mlockall(MCL_CURRENT | MCL_FUTURE)) fprintf(stderr, "normStreamer error: failed to lock memory for receiver.\n"); if (0 != rx_socket_buffer_size) NormSetRxSocketBuffer(norm_session, rx_socket_buffer_size); rx_needed = true; rx_ready = false; } if (sender) { NormSetGrttEstimate(norm_session, 0.001); //NormSetGrttMax(norm_session, 0.100); NormSetBackoffFactor(norm_session, 0); if (norm_acking) { // 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); 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; gettimeofday(¤tTime, NULL); srand(currentTime.tv_usec); // seed random number generator NormSessionId instanceId = (NormSessionId)rand(); if (fti_info) NormLimitObjectInfo(norm_session, true); if (!NormStartSender(norm_session, instanceId, bufferSize, segment_size, block_size, num_parity)) { fprintf(stderr, "normStreamer error: unable to start NORM sender\n"); if (receiver) NormStopReceiver(norm_session); return false; } if (auto_parity > 0) NormSetAutoParity(norm_session, auto_parity < num_parity ? auto_parity : num_parity); if (0 != tx_socket_buffer_size) NormSetTxSocketBuffer(norm_session, tx_socket_buffer_size); if (NORM_OBJECT_INVALID == (tx_stream = NormStreamOpen(norm_session, stream_buffer_size))) { fprintf(stderr, "normStreamer error: unable to open NORM tx stream\n"); NormStopSender(norm_session); if (receiver) NormStopReceiver(norm_session); return false; } else { if (0 != mlockall(MCL_CURRENT|MCL_FUTURE)) fprintf(stderr, "normStreamer warning: failed to lock memory for sender.\n"); } tx_stream_buffer_max = NormGetStreamBufferSegmentCount(bufferSize, segment_size, block_size); tx_stream_buffer_max -= block_size; // a little safety margin (perhaps not necessary) tx_stream_buffer_threshold = tx_stream_buffer_max / 8; tx_stream_buffer_count = 0; tx_stream_bytes_remain = 0; tx_watermark_pending = false; tx_ack_pending = false; tx_ready = true; input_index = input_msg_length = 0; input_needed = true; input_ready = true; } is_running = true; return true; } // end NormStreamer::Start(); void NormStreamer::ReadInputSocket() { unsigned int loopCount = 0; NormSuspendInstance(NormGetInstance(norm_session)); while (input_needed && input_ready && (loopCount < LOOP_MAX)) { loopCount++; unsigned int numBytes = MSG_SIZE_MAX - MSG_HEADER_SIZE; ProtoAddress srcAddr; if (input_socket.RecvFrom(input_buffer+MSG_HEADER_SIZE, numBytes, srcAddr)) { if (0 == numBytes) { input_ready = false; break; } input_index = 0; input_msg_length = numBytes + MSG_HEADER_SIZE; input_byte_count += input_msg_length; unsigned short msgSize = input_msg_length;; msgSize = htons(msgSize); memcpy(input_buffer, &msgSize, MSG_HEADER_SIZE); input_needed = false; if (TxReady()) SendData(); } else { // TBD - handle error? input_ready = false; } } NormResumeInstance(NormGetInstance(norm_session)); } // end NormStreamer::ReadInputSocket() void NormStreamer::ReadInput() { if (UdpListenerEnabled()) return ReadInputSocket(); // The loop count makes sure we don't spend too much time here // before going back to the main loop to handle NORM events, etc unsigned int loopCount = 0; NormSuspendInstance(NormGetInstance(norm_session)); while (input_needed && input_ready && (loopCount < LOOP_MAX)) { loopCount++; //if (100 == loopCount) // fprintf(stderr, "normStreamer ReadInput() loop count max reached\n"); unsigned int numBytes; if (input_index < MSG_HEADER_SIZE) { // Reading message length header for next message to send numBytes = MSG_HEADER_SIZE - input_index; } else { // Reading message body assert(input_index < input_msg_length); numBytes = input_msg_length - input_index; } ssize_t result = read(input_fd, input_buffer + input_index, numBytes); if (result > 0) { input_index += result; input_byte_count += result; if (MSG_HEADER_SIZE == input_index) { // We have now read the message size header // TBD - support other message header formats? // (for now, assume 2-byte message length header) uint16_t msgSize ; memcpy(&msgSize, input_buffer, MSG_HEADER_SIZE); msgSize = ntohs(msgSize); input_msg_length = msgSize; } else if (input_index == input_msg_length) { // Message input complete input_index = 0; // reset index for transmission phase input_needed = false; if (TxReady()) SendData(); } else { // Still need more input // (wait for next input notification to read more) input_ready = false; } } else if (0 == result) { // end-of-file reached, TBD - trigger final flushing and wrap-up fprintf(stderr, "normStreamer: input end-of-file detected ...\n"); NormStreamClose(tx_stream, true); if (norm_acking) { if (ack_ex) { const char* req = "Hello, acker"; NormSetWatermarkEx(norm_session, tx_stream, req, strlen(req) + 1, true); } else { NormSetWatermark(norm_session, tx_stream, true); } tx_ack_pending = false; } input_needed = false; } else { switch (errno) { case EINTR: continue; // interrupted, try again case EAGAIN: // input starved, wait for next notification input_ready = false; break; default: // TBD - handle this better perror("normStreamer error reading input"); break; } break; } } // end while (input_needed && input_ready) NormResumeInstance(NormGetInstance(norm_session)); } // end NormStreamer::ReadInput() void NormStreamer::SendData() { while (TxReady() && !input_needed) { // Note WriteToStream() or FlushStream() will set "tx_ready" to // false upon flow control thus negating TxReady() status assert(input_index < input_msg_length); assert(input_msg_length); if ((0 != check_sequence) && (0 == input_index)) CheckSequenceNumber(input_buffer+MSG_HEADER_SIZE, __func__); input_index += WriteToStream(input_buffer + input_index, input_msg_length - input_index); if (input_index == input_msg_length) { // Complete message was sent, so set eom and optionally flush if (NORM_FLUSH_NONE != flush_mode) FlushStream(true, flush_mode); else NormStreamMarkEom(tx_stream); input_index = input_msg_length = 0; input_needed = true; } else { //fprintf(stderr, "SendData() impeded by flow control\n"); } } // end while (TxReady() && !input_needed) } // end NormStreamer::SendData() unsigned int NormStreamer::WriteToStream(const char* buffer, unsigned int numBytes) { unsigned int bytesWritten; if (norm_acking) { // This method uses NormStreamWrite(), but limits writes by explicit ACK-based flow control status if (tx_stream_buffer_count < tx_stream_buffer_max) { // 1) How many buffer bytes are available? unsigned int bytesAvailable = segment_size * (tx_stream_buffer_max - tx_stream_buffer_count); bytesAvailable -= tx_stream_bytes_remain; // unflushed segment portion if (bytesAvailable < numBytes) numBytes = bytesAvailable; assert(numBytes); // 2) Write to the stream bytesWritten = NormStreamWrite(tx_stream, buffer, numBytes); tx_byte_count += bytesWritten; // 3) Update "tx_stream_buffer_count" accordingly unsigned int totalBytes = bytesWritten + tx_stream_bytes_remain; unsigned int numSegments = totalBytes / segment_size; tx_stream_bytes_remain = totalBytes % segment_size; tx_stream_buffer_count += numSegments; //assert(bytesWritten == numBytes); // this could fail if timer-based flow control is left enabled // 3) Check if we need to issue a watermark ACK request? if (!tx_watermark_pending && (tx_stream_buffer_count >= tx_stream_buffer_threshold)) { // Initiate flow control ACK request //fprintf(stderr, "write-initiated flow control ACK REQUEST\n"); if (ack_ex) { const char* req = "Hello, acker"; NormSetWatermarkEx(norm_session, tx_stream, req, strlen(req) + 1);//, true); } else { NormSetWatermark(norm_session, tx_stream);//, true); } tx_watermark_pending = true; tx_ack_pending = false; } } else { fprintf(stderr, "normStreamer: sender flow control limited\n"); return 0; } } else { bytesWritten = NormStreamWrite(tx_stream, buffer, numBytes); tx_byte_count += bytesWritten; } if (bytesWritten != numBytes) //NormStreamWrite() was (at least partially) blocked { //fprintf(stderr, "NormStreamWrite() blocked by flow control ...\n"); tx_ready = false; } return bytesWritten; } // end NormStreamer::WriteToStream() void NormStreamer::FlushStream(bool eom, NormFlushMode flushMode) { if (norm_acking) { bool setWatermark = false; if (0 != tx_stream_bytes_remain) { // The flush will force the runt segment out, so we increment our buffer usage count // (and initiate flow control watermark ack request if buffer mid-point threshold exceeded tx_stream_buffer_count++; tx_stream_bytes_remain = 0; if (!tx_watermark_pending && (tx_stream_buffer_count >= tx_stream_buffer_threshold)) { setWatermark = true; tx_watermark_pending = true; //fprintf(stderr, "flush-initiated flow control ACK REQUEST\n"); } } // The check for "tx_watermark_pending" here prevents a new watermark // ack request from being set until the pending flow control ack is // received. This favors avoiding dead air time over saving "chattiness" if (setWatermark) { // Flush passive since watermark will invoke active request // (TBD - do non-acking nodes NACK to watermark when not ack target?) NormStreamFlush(tx_stream, eom, NORM_FLUSH_PASSIVE); } else if (tx_watermark_pending) { // Pre-existing pending flow control watermark ack request #if SHOOT_FIRST // Go ahead and set a fresh watermark // TBD - not sure this mode works properly ... may need to // keep track of unacknowledged byte count and decrement accordingly // when ack arrives NormStreamFlush(tx_stream, eom, NORM_FLUSH_PASSIVE); setWatermark = true; #else // ACK_LATER // Wait until flow control ACK is received before issuing another ACK request NormStreamFlush(tx_stream, eom, flushMode); tx_ack_pending = true; // will call NormSetWatermark() upon flow control ack completion #endif } else { // Since we're acking, use active ack request in lieu of active flush NormStreamFlush(tx_stream, eom, NORM_FLUSH_PASSIVE); setWatermark = true; } if (setWatermark) { if (ack_ex) { const char* req = "Hello, acker"; NormSetWatermarkEx(norm_session, tx_stream, req, strlen(req) + 1, true); } else { NormSetWatermark(norm_session, tx_stream, true); } tx_ack_pending = false; } } else { NormStreamFlush(tx_stream, eom, flushMode); } } // end NormStreamer::FlushStream() void NormStreamer::RecvData() { // The loop count makes sure we don't spend too much time here // before going back to the main loop to handle NORM events, etc unsigned int loopCount = 0; // Reads data from rx_stream to available output_buffer NormSuspendInstance(NormGetInstance(norm_session)); while (rx_needed && rx_ready && (loopCount < LOOP_MAX)) { loopCount++; //if (100 == loopCount) // fprintf(stderr, "normStreamer RecvData() loop count max reached.\n"); // Make sure we have msg_sync (TBD - skip this for byte streaming) if (!msg_sync) { msg_sync = NormStreamSeekMsgStart(rx_stream); if (!msg_sync) { rx_ready = false; break; // wait for next NORM_RX_OBJECT_UPDATED to re-sync } } unsigned int bytesWanted; if (output_index < MSG_HEADER_SIZE) { // Receiving message header bytesWanted = MSG_HEADER_SIZE - output_index; } else { // Receiving message body assert(output_index < output_msg_length); bytesWanted = output_msg_length - output_index; } unsigned bytesRead = bytesWanted; if (!NormStreamRead(rx_stream, output_buffer + output_index, &bytesRead)) { // Stream broken (should _not_ happen if norm_acking flow control) //fprintf(stderr, "normStreamer error: BROKEN stream detected, re-syncing ...\n"); msg_sync = false; output_index = output_msg_length = 0; continue; } output_index += bytesRead; /*if (0 == bytesRead) { rx_ready = false; } else*/ if (bytesRead != bytesWanted) { //continue; rx_ready = false; // didn't get all we need } else if (MSG_HEADER_SIZE == output_index) { // We have now read the message size header // TBD - support other message header formats? // (for now, assume 2-byte message length header) uint16_t msgSize ; memcpy(&msgSize, output_buffer, MSG_HEADER_SIZE); output_msg_length = ntohs(msgSize); } else if (output_index == output_msg_length) { // Received full message rx_needed = false; output_index = 0; // reset for writing to output if (output_ready && OutputBucketReady()) WriteOutput(); } } NormResumeInstance(NormGetInstance(norm_session)); } // end NormStreamer::RecvData() void NormStreamer::WriteOutputSocket() { if (output_ready && !rx_needed) { assert(output_index < output_msg_length); unsigned int payloadSize = output_msg_length - MSG_HEADER_SIZE; unsigned int numBytes = payloadSize; if ((0 != check_sequence)) CheckSequenceNumber(output_buffer+MSG_HEADER_SIZE, __func__); if (output_socket.SendTo(output_buffer+MSG_HEADER_SIZE, numBytes, relay_addr)) { if (numBytes != payloadSize) { // sendto() was blocked output_ready = false; return; } if (0 != output_bucket_depth) { // Debit output token bucket since it's active ASSERT(output_bucket_count >= payloadSize); output_bucket_count -= payloadSize; } rx_needed = true; output_index = output_msg_length = 0; } else { output_ready = false; } } } // end NormStreamer::WriteOutputSocket() void NormStreamer::WriteOutput() { if (UdpRelayEnabled()) { WriteOutputSocket(); return; } while (output_ready && !rx_needed) { assert(output_index < output_msg_length); if ((0 != check_sequence) && (0 == output_index)) CheckSequenceNumber(output_buffer+MSG_HEADER_SIZE,__func__); ssize_t result = write(output_fd, output_buffer + output_index, output_msg_length - output_index); if (result >= 0) { if (0 != output_bucket_depth) { // Debit output token bucket since it's active if (result > output_bucket_count) TRACE("result:%d output_bucket_count:%u\n", (int)result, output_bucket_count); ASSERT(output_bucket_count >= result); output_bucket_count -= result; } output_index += result; if (output_index == output_msg_length) { // Complete message written rx_needed = true; output_index = output_msg_length = 0; if ((NORM_OBJECT_INVALID == tx_stream) && (NORM_OBJECT_INVALID == rx_stream)) Stop(); // receive stream was terminated by sender } else { output_ready = false; } } else { switch (errno) { case EINTR: perror("normStreamer output EINTR"); continue; // interupted, try again case EAGAIN: // output blocked, wait for next notification //perror("normStreamer output blocked"); output_ready = false; break; default: perror("normStreamer error writing output"); break; } break; } } } // end NormStreamer::WriteOutput() void NormStreamer::HandleNormEvent(const NormEvent& event) { switch (event.type) { case NORM_TX_QUEUE_EMPTY: //TRACE("normStreamer: flow control empty ...\n"); tx_ready = true; break; case NORM_TX_QUEUE_VACANCY: //TRACE("normStreamer: flow control relieved ...\n"); tx_ready = true; break; case NORM_GRTT_UPDATED: //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: if (NORM_ACK_SUCCESS == NormGetAckingStatus(norm_session)) { //fprintf(stderr, "WATERMARK COMPLETED\n"); 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; tx_stream_buffer_count -= tx_stream_buffer_threshold; //fprintf(stderr, "flow control ACK completed\n"); if (tx_ack_pending) { if (ack_ex) { const char* req = "Hello, acker"; NormSetWatermarkEx(norm_session, tx_stream, req, strlen(req) + 1, true); } else { NormSetWatermark(norm_session, tx_stream, true); } tx_ack_pending = false; } } } else { // TBD - we could see who didn't ACK and possibly remove them // from our acking list. For now, we are infinitely // persistent by always resetting the watermark ack request // For example, an application could make a decision at this // point, depending upon some count of ACK request failures // to choose to remove a previously included receiver. fprintf(stderr, "flow control watermark reset\n"); if (tx_ack_pending) { // May as well advance the ack request point if (ack_ex) { const char* req = "Hello, acker"; NormSetWatermarkEx(norm_session, tx_stream, req, strlen(req) + 1, true); } else { NormSetWatermark(norm_session, tx_stream, true); } tx_ack_pending = false; } else { NormResetWatermark(norm_session); } } if (ack_ex) { // This iterates through the acking nodes looking for responses // to our application-defined NormSetWatermarkEx() request NormAckingStatus ackingStatus; NormNodeId nodeId = NORM_NODE_NONE; // this inits NormGetNextAckingNode() iteration while (NormGetNextAckingNode(event.session, &nodeId, &ackingStatus)) { if (NORM_ACK_SUCCESS == ackingStatus) { // This node acked, so look for AckEx response // In our example/test case here, we use strings for the content char buffer[256]; buffer[0] = '\0'; unsigned int buflen = 256; if (NormGetAckEx(event.session, nodeId, buffer, &buflen)) fprintf(stderr, "Received APP_ACK from node>%u \"%s\"\n", nodeId, buffer); } } } break; case NORM_TX_OBJECT_PURGED: // tx_stream graceful close completed NormStopSender(norm_session); tx_stream = NORM_OBJECT_INVALID; if (NORM_OBJECT_INVALID == rx_stream) Stop(); break; case NORM_REMOTE_SENDER_INACTIVE: //fprintf(stderr, "REMOTE SENDER INACTIVE node: %u\n", NormNodeGetId(event.sender)); //NormNodeDelete(event.sender); break; case NORM_RX_OBJECT_NEW: if ((NORM_OBJECT_INVALID == rx_stream) && (NORM_OBJECT_STREAM == NormObjectGetType(event.object))) { rx_stream = event.object; rx_ready = true; // By setting initial "msg_sync" to true, we can detect when // stream beginning was missed (for NORM_SYNC_STREAM only) msg_sync = false; rx_needed = true; output_index = output_msg_length = 0; } else { fprintf(stderr, "normStreamer warning: NORM_RX_OBJECT_NEW while already receiving?!\n"); } case NORM_RX_OBJECT_UPDATED: rx_ready = true; break; case NORM_RX_OBJECT_ABORTED: //fprintf(stderr, "NORM_RX_OBJECT_ABORTED\n");// %hu\n", NormObjectGetTransportId(event.object)); rx_stream = NORM_OBJECT_INVALID; rx_needed = false; rx_ready = false; break; case NORM_RX_OBJECT_COMPLETED: // Rx stream has closed // TBD - set state variables so any pending output is // written out and things shutdown if not sender, too fprintf(stderr, "normStreamer: rx_stream completed.\n"); // if rx_needed is true, all output has been written if (rx_needed && (NORM_OBJECT_INVALID == tx_stream)) { NormNodeHandle sender = NormObjectGetSender(rx_stream); // Wait a couple of GRTT's to ACK sender double exitTime = 20.0 * NormNodeGetGrtt(sender); if (exitTime < 1.0) exitTime = 1.0; fprintf(stderr, "normStreamer reception completed, exiting in %f seconds ...\n", (float)exitTime); sleep(exitTime); // TBD - use our user-defined NormSession timeout instead? (retaining rx_stream) if (rx_needed && (NORM_OBJECT_INVALID == tx_stream)) Stop(); } rx_stream = NORM_OBJECT_INVALID; rx_ready = false; rx_needed = false; break; case NORM_RX_ACK_REQUEST: { char buffer[256]; buffer[0] = '\0'; unsigned int buflen = 256; NormNodeGetWatermarkEx(event.sender, buffer, &buflen); fprintf(stderr, "Received NORM_RX_ACK_REQUEST: \"%s\"\n", buffer); // Send a reply const char* ack = "Yes, master"; NormNodeSendAckEx(event.sender, ack, strlen(ack) + 1); break; } default: break; } //NormReleasePreviousEvent(NormGetInstance(norm_session)); } // end NormStreamer::HandleNormEvent() void Usage() { fprintf(stderr, "Usage: normStreamer id {send|recv} [addr [/]]\n" " [interface ] [loopback] [info] [ptos ] [ex]\n" " [cc|cce|ccl|rate ]\n" " [ack auto|[,,...]]\n" " [flush {none|passive|active}]\n" " [listen [/]] [linterface ]\n" " [relay /] [limit [/]]\n" " [output ] [boost] [debug ] [trace]\n" " [log ] [segment ] [block ]\n" " [parity ] [auto ]\n" " [insockbuffer ] [outsockbuffer ]\n" " [txsockbuffer ] [rxsockbuffer ]\n" " [streambuffer ]\n" " [check64 | check32] [omit] [silent]\n" " [txloss ] [rxloss ]\n"); } // end Usage() void PrintHelp() { fprintf(stderr, "\nHelp for normStreamer:\n\n") ; fprintf(stderr, "The 'normStreamer' application sends messages from STDIN (or a listening UDP socket) to one or more\n" "receiving nodes using the NORM protocol. Received messages are output to STDOUT (or relayed to\n" "to a UDP destination address/port). Key command line options are:\n\n" " id -- Specifies the node id for the local NORM instance (required)\n" " send | recv -- Specifies whether this node will be a sender and/or receiver (must choose at least one)\n" " addr [/] -- specifies the network address over which to send/receive NORM protocol\n" " interface -- Specifies the name of the network interface on which to conduct NORM protocol\n" " (e.g., 'eth0')\n" " loopback -- Enables 'loopback' sessions on the same host machine. Required for multicast loopback.\n" " ptos -- Set special IP traffic class (TOS) for GRTT probing and acknowledgments\n" " info -- Limits FTI header extension to NORM_INFO message only (reduced overhead)\n" " rate -- sets fixed sender rate (and receiver token bucket rate if 'limit' option is used)\n" " [cc|cce|ccl] -- Enables optional NORM congestion control mode (overrides 'rate')\n" " ack [] -- Instructs sender to request positive acknowledgement from listed receiver nodes\n" " flush [] -- Choose 'none', 'passive', or 'active' message stream flushing mode. If 'none',\n" " NORM_DATA packets will always be packed with message content up to the full\n" " segment size. If 'passive', short NORM_DATA packets will be sent to transmit\n" " any messages as soon as possible. If 'active', NORM stream will be flushed\n" " on a per-message basis as with 'passive' mode, but positive acknowledgment will\n" " _also_ be requested if a list of acking receiver node ids has beeen provided.\n" " listen [/] -- Specifies the port and optional multicast address which the sender uses to listen\n" " for UDP packets to transmit to the receiver(s) via the NORM protocol\n" " linterface -- Specifies the name of the network interface on which to listen for UDP packet\n" " payloads to send to the receiver(s) via NORM protocol\n" " relay / -- Specifies the address/port for which to relay (as UDP datagrams) received messages\n" " limit [/] -- Token bucket rate/depth for optional receiver output limiter (smooths bursty output\n" " upon NORM loss recovery). When UDP 'relay' is used, this option is useful to avoid\n" " overly bursty UDP output. The is in units of bits/second and the is\n" " in units of bytes. If not specified here, the value set by 'rate' command is used\n" " as the token bucket rate.\n" " check64 | check32 -- Enables checking that packet sequence numbers in the first 4/8 bytes of received\n" " packets increment properly (optional)\n" " insockbuffer -- Specifies the size of the 'listen' UDP socket buffer (optional).\n" " outsockbuffer -- Specifies the size of the 'relay' UDP socket buffer (optional).\n" " txsockbuffer -- Specifies the size of the NORM/UDP transmit socket buffer (optional).\n" " rxsockbuffer -- Specifies the size of the NORM/UDP receive socket buffer (optional).\n" " streambuffer -- Specifies the size of the NORM stream buffer (optional).\n\n"); Usage(); } // end PrintHelp() int main(int argc, char* argv[]) { // REQUIRED parameters initiailization NormNodeId nodeId = NORM_NODE_NONE; bool send = false; bool recv = false; char sessionAddr[64]; strcpy(sessionAddr, "224.1.2.3"); unsigned int sessionPort = 6003; char listenAddr[64]; // UDP :listen" multicast addr listenAddr[0] = '\0'; unsigned int listenPort = 0; // UDP "listen" port for UDP "listen" const char* listenIface = NULL; // UDP "listen" interface char relayAddr[64]; relayAddr[0] = '\0'; unsigned int relayPort = 0; double txRate = 0.0; // used for non-default NORM_FIXED ccMode NormStreamer::CCMode ccMode = NormStreamer::NORM_CC; const char* mcastIface = NULL; NormNodeId ackingNodeList[256]; unsigned int ackingNodeCount = 0; bool loopback = false; bool ftiInfo = false; bool ackEx = false; int debugLevel = 0; bool trace = false; const char* logFile = NULL; bool omitHeaderOnOutput = false; bool silentReceiver = false; double txloss = 0.0; double rxloss = 0.0; 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 // 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; normStreamer.SetFlushMode(NORM_FLUSH_NONE); // Parse command-line int i = 1; while (i < argc) { const char* cmd = argv[i++]; size_t len = strlen(cmd); if (0 == strncmp(cmd, "help", len)) { PrintHelp() ; exit(0); } else if (0 == strncmp(cmd, "send", len)) { send = true; } else if (0 == strncmp(cmd, "recv", len)) { recv = true; } else if (0 == strncmp(cmd, "loopback", len)) { loopback = true; } else if (0 == strncmp(cmd, "ptos", len)) { if (i >= argc) { fprintf(stderr, "normStreamer error: missing 'ptos' value!\n"); Usage(); return -1; } int tos = -1; int result = sscanf(argv[i], "%i", &tos); if (1 != result) { unsigned int utos; result = sscanf(argv[i], "%x", &utos); tos = utos; } if ((1 != result) || (tos < 0) || (tos > 255)) { fprintf(stderr, "normStreamer error: invalid 'ptos' value!\n"); Usage(); return -1; } i++; normStreamer.SetProbeTOS((UINT8)tos); } else if (0 == strncmp(cmd, "info", len)) { ftiInfo = true; } else if (0 == strncmp(cmd, "ex", len)) { // This enables testing/demonstrating NormSetWatermarkEx() operation ackEx = true; } else if (0 == strncmp(cmd, "addr", len)) { if (i >= argc) { fprintf(stderr, "normStreamer error: missing 'addr[/port]' value!\n"); Usage(); return -1; } const char* addrPtr = argv[i++]; const char* portPtr = strchr(addrPtr, '/'); if (NULL == portPtr) { strncpy(sessionAddr, addrPtr, 63); sessionAddr[63] = '\0'; } else { size_t addrLen = portPtr - addrPtr; if (addrLen > 63) addrLen = 63; // should issue error message strncpy(sessionAddr, addrPtr, addrLen); sessionAddr[addrLen] = '\0'; portPtr++; sessionPort = atoi(portPtr); } } else if (0 == strncmp(cmd, "listen", len)) { if (i >= argc) { fprintf(stderr, "normStreamer error: missing '[mcastAddr/]port]' value!\n"); Usage(); return -1; } const char* addrPtr = argv[i++]; const char* portPtr = strchr(addrPtr, '/'); if (NULL != portPtr) { size_t addrLen = portPtr - addrPtr; if (addrLen > 63) addrLen = 63; // should issue error message strncpy(listenAddr, addrPtr, addrLen); listenAddr[addrLen] = '\0'; portPtr++; listenPort = atoi(portPtr); } else { // no address, just port listenPort = atoi(addrPtr); addrPtr = NULL; } } else if (0 == strncmp(cmd, "relay", len)) { if (i >= argc) { fprintf(stderr, "normStreamer error: missing relay 'dstAddr/port' value!\n"); Usage(); return -1; } const char* addrPtr = argv[i++]; const char* portPtr = strchr(addrPtr, '/'); if (NULL == portPtr) { fprintf(stderr, "normStreamer error: missing relay 'port' value!\n"); Usage(); return -1; } if (NULL != portPtr) { size_t addrLen = portPtr - addrPtr; if (addrLen > 63) addrLen = 63; // should issue error message strncpy(relayAddr, addrPtr, addrLen); relayAddr[addrLen] = '\0'; portPtr++; relayPort = atoi(portPtr); } } else if (0 == strncmp(cmd, "output", len)) { if (i >= argc) { fprintf(stderr, "normStreamer error: missing output 'device' name!\n"); Usage(); return -1; } FILE* outfile = fopen(argv[i++], "w+"); if (NULL == outfile) { fprintf(stderr, "normStreamer output device fopen() error: %s\n", GetErrorString()); Usage(); return -1; } normStreamer.SetOutputFile(outfile); } else if (0 == strncmp(cmd, "id", len)) { if (i >= argc) { fprintf(stderr, "normStreamer error: missing 'id' value!\n"); Usage(); return -1; } nodeId = atoi(argv[i++]); } else if (0 == strncmp(cmd, "ack", len)) { // comma-delimited acking node id list if (i >= argc) { fprintf(stderr, "normStreamer error: missing 'id' value!\n"); Usage(); return -1; } const char* alist = argv[i++]; if (0 == strcmp(alist, "auto")) { normStreamer.SetAutoAck(true); } else { while ((NULL != alist) && (*alist != '\0')) { // 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 } } } else if (0 == strncmp(cmd, "flush", len)) { // "none", "passive", or "active" if (i >= argc) { fprintf(stderr, "nodeMsgr error: missing 'flush' !\n"); Usage(); return -1; } const char* mode = argv[i++]; if (0 == strcmp(mode, "none")) { normStreamer.SetFlushMode(NORM_FLUSH_NONE); } else if (0 == strcmp(mode, "passive")) { normStreamer.SetFlushMode(NORM_FLUSH_PASSIVE); } else if (0 == strcmp(mode, "active")) { normStreamer.SetFlushMode(NORM_FLUSH_ACTIVE); } else { fprintf(stderr, "normMsgr error: invalid 'flush' mode \"%s\"\n", mode); return -1; } } else if (0 == strncmp(cmd, "rate", len)) { if (i >= argc) { fprintf(stderr, "normStreamer error: missing 'rate' value!\n"); Usage(); return -1; } if (1 != sscanf(argv[i++], "%lf", &txRate)) { fprintf(stderr, "normStreamer error: invalid transmit rate!\n"); Usage(); return -1; } // set fixed-rate operation ccMode = NormStreamer::NORM_FIXED; normStreamer.SetOutputBucketRate(txRate); } else if (0 == strcmp(cmd, "cc")) { ccMode = NormStreamer::NORM_CC; } else if (0 == strcmp(cmd, "cce")) { ccMode = NormStreamer::NORM_CCE; } else if (0 == strcmp(cmd, "ccl")) { ccMode = NormStreamer::NORM_CCL; } else if (0 == strncmp(cmd, "interface", len)) { if (i >= argc) { fprintf(stderr, "normStreamer error: missing 'interface' !\n"); Usage(); return -1; } mcastIface = argv[i++]; } else if (0 == strncmp(cmd, "linterface", len)) { if (i >= argc) { fprintf(stderr, "normStreamer error: missing 'linterface' !\n"); Usage(); return -1; } listenIface = argv[i++]; } else if (0 == strncmp(cmd, "insockbuffer", len)) { unsigned long value = 0 ; if (i >= argc) { fprintf(stderr, "normStreamer error: missing 'insockbuffer' size!\n"); Usage(); return -1; } if (1 != sscanf(argv[i++], "%lu", &value)) { fprintf(stderr, "normStreamer error: invalid 'insockbuffer' size\n"); Usage(); return -1; } inputSocketBufferSize = value; } else if (0 == strncmp(cmd, "outsockbuffer", len)) { unsigned long value = 0 ; if (i >= argc) { fprintf(stderr, "normStreamer error: missing 'outsockbuffer' size!\n"); Usage(); return -1; } if (1 != sscanf(argv[i++], "%lu", &value)) { fprintf(stderr, "normStreamer error: invalid 'outsockbuffer' size!\n"); Usage(); return -1; } outputSocketBufferSize = value; } else if (0 == strncmp(cmd, "limit", len)) { // format: limit [/] with 'rate' in bps and 'size' in bytes if (i >= argc) { fprintf(stderr, "normStreamer error: missing 'limit' size!\n"); Usage(); return -1; } const char* ratePtr = argv[i++]; const char* sizePtr = strchr(ratePtr, '/'); unsigned int rateLen = 0; if (NULL != sizePtr) rateLen = sizePtr++ - ratePtr; else sizePtr = ratePtr; if (0 != rateLen) { if (rateLen > 63) { fprintf(stderr, "normStreamer error: out-of-bounds 'limit' rate\n"); Usage(); return -1; } char rateText[64]; strncpy(rateText, ratePtr, rateLen); rateText[rateLen] = '\0'; double value; if (1 != sscanf(rateText, "%lf", &value)) { fprintf(stderr, "normStreamer error: invalid 'limit' rate\n"); Usage(); return -1; } normStreamer.SetOutputBucketRate(value); } unsigned long value; if (1 != sscanf(sizePtr, "%lu", &value)) { fprintf(stderr, "normStreamer error: invalid 'limit' size\n"); Usage(); return -1; } normStreamer.SetOutputBucketDepth(value); } else if (0 == strncmp(cmd, "txsockbuffer", len)) { unsigned long value = 0 ; if (i >= argc) { fprintf(stderr, "normStreamer error: missing 'txsockbuffer' size!\n"); Usage(); return -1; } if (1 != sscanf(argv[i++], "%lu", &value)) { fprintf(stderr, "normStreamer error: invalid 'txsockbuffer' size!\n"); Usage(); return -1; } txSocketBufferSize = value; } else if (0 == strncmp(cmd, "rxsockbuffer", len)) { unsigned long value = 0 ; if (i >= argc) { fprintf(stderr, "normStreamer error: missing 'rxsockbuffer' size!\n"); Usage(); return -1; } if (1 != sscanf(argv[i++], "%lu", &value)) { fprintf(stderr, "normStreamer error: invalid 'rxsockbuffer' size!\n"); Usage(); return -1; } rxSocketBufferSize = value; } else if (0 == strncmp(cmd, "segment", len)) { if (i >= argc) { fprintf(stderr, "normStreamer error: missing 'segment' size!\n"); Usage(); return -1; } unsigned short value; if (1 != sscanf(argv[i++], "%hu", &value)) { fprintf(stderr, "normStreamer error: invalid 'segment' size!\n"); Usage(); return -1; } normStreamer.SetSegmentSize(value); } else if (0 == strncmp(cmd, "block", len)) { if (i >= argc) { fprintf(stderr, "normStreamer error: missing 'block' size!\n"); Usage(); return -1; } unsigned short value; if (1 != sscanf(argv[i++], "%hu", &value)) { fprintf(stderr, "normStreamer error: invalid 'block' size!\n"); Usage(); return -1; } normStreamer.SetBlockSize(value); } else if (0 == strncmp(cmd, "parity", len)) { if (i >= argc) { fprintf(stderr, "normStreamer error: missing 'parity' count!\n"); Usage(); return -1; } unsigned short value; if (1 != sscanf(argv[i++], "%hu", &value)) { fprintf(stderr, "normStreamer error: invalid 'parity' count!\n"); Usage(); return -1; } normStreamer.SetNumParity(value); } else if (0 == strncmp(cmd, "auto", len)) { if (i >= argc) { fprintf(stderr, "normStreamer error: missing 'auto' parity count!\n"); Usage(); return -1; } unsigned short value; if (1 != sscanf(argv[i++], "%hu", &value)) { fprintf(stderr, "normStreamer error: invalid 'auto' parity count!\n"); Usage(); return -1; } normStreamer.SetAutoParity(value); } else if (0 == strncmp(cmd, "streambuffer", len)) { unsigned long value = 0 ; if (i >= argc) { fprintf(stderr, "normStreamer error: missing 'streambuffer' size!\n"); Usage(); return -1; } if (1 != sscanf(argv[i++], "%lu", &value)) { fprintf(stderr, "normStreamer error: invalid 'streambuffer' size!\n"); Usage(); return -1; } streamBufferSize = value; } else if ( 0 == strncmp(cmd,"chkseq", len) ) { checkSequence = 64; // same as "check64" for "historical" reasons } else if ( 0 == strncmp(cmd,"check64", len) ) { checkSequence = 64; } else if ( 0 == strncmp(cmd,"check32", len) ) { checkSequence = 32; } else if (0 == strncmp(cmd, "omit", len)) { omitHeaderOnOutput = true; } else if (0 == strncmp(cmd, "silent", len)) { silentReceiver = true; } else if (0 == strncmp(cmd, "boost", len)) { boostPriority = true; } else if (0 == strncmp(cmd, "txloss", len)) { if (1 != sscanf(argv[i++], "%lf", &txloss)) { fprintf(stderr, "normStreamer error: invalid 'txloss' value!\n"); Usage(); return -1; } } else if (0 == strncmp(cmd, "rxloss", len)) { if (1 != sscanf(argv[i++], "%lf", &rxloss)) { fprintf(stderr, "normStreamer error: invalid 'rxloss' value!\n"); Usage(); return -1; } } else if (0 == strncmp(cmd, "debug", len)) { if (i >= argc) { fprintf(stderr, "normStreamer error: missing 'debug' !\n"); Usage(); return -1; } debugLevel = atoi(argv[i++]); } else if (0 == strncmp(cmd, "trace", len)) { trace = true; } else if (0 == strncmp(cmd, "log", len)) { if (i >= argc) { fprintf(stderr, "normStreamer error: missing 'log' !\n"); Usage(); return -1; } logFile = argv[i++]; } else if (0 == strncmp(cmd, "help", len)) { Usage(); return 0; } else { fprintf(stderr, "normStreamer error: invalid command \"%s\"!\n", cmd); Usage(); return -1; } } if (!send && !recv) { fprintf(stderr, "normStreamer error: not configured to send or recv!\n"); Usage(); return -1; } /* if (NORM_NODE_NONE == nodeId) { fprintf(stderr, "normStreamer error: no local 'id' provided!\n"); Usage(); return -1; } */ if (boostPriority) { if (!normStreamer.BoostPriority()) { fprintf(stderr, "normStreamer error: setting scheduler/ priority boost failed (requires 'sudo').\n"); return -1; } } if (0 != listenPort) { normStreamer.SetInputSocketBufferSize(inputSocketBufferSize); if (!normStreamer.EnableUdpListener(listenPort, listenAddr, listenIface)) { fprintf(stderr, "normStreamer error: Failed to enable UDP listener\n") ; return -1; } } if (0 != relayPort) { // TBD - check addr/port validity? normStreamer.SetOutputSocketBufferSize(outputSocketBufferSize); if (! normStreamer.EnableUdpRelay(relayAddr, relayPort)) { fprintf(stderr, "normStreamer error: Failed to open UDP relay socket\n") ; return -1; } } // TBD - should provide more error checking of NORM API calls NormInstanceHandle normInstance = NormCreateInstance(boostPriority); NormSetDebugLevel(debugLevel); if ((NULL != logFile) && !NormOpenDebugLog(normInstance, logFile)) { perror("normStreamer error: unable to open log file"); Usage(); return -1; } normStreamer.SetCheckSequence(checkSequence); normStreamer.SetTxSocketBufferSize(txSocketBufferSize); normStreamer.SetRxSocketBufferSize(rxSocketBufferSize); normStreamer.SetStreamBufferSize(streamBufferSize); normStreamer.SetLoopback(loopback); normStreamer.SetFtiInfo(ftiInfo); normStreamer.SetAckEx(ackEx); if (omitHeaderOnOutput) normStreamer.OmitHeader(true); if (!normStreamer.OpenNormSession(normInstance, sessionAddr, sessionPort, (NormNodeId)nodeId)) { fprintf(stderr, "normStreamer error: unable to open NORM session\n"); NormDestroyInstance(normInstance); return -1; } if (silentReceiver) normStreamer.SetSilentReceiver(true); if (txloss > 0.0) normStreamer.SetTxLoss(txloss); if (rxloss > 0.0) normStreamer.SetRxLoss(rxloss); for (unsigned int i = 0; i < ackingNodeCount; i++) normStreamer.AddAckingNode(ackingNodeList[i]); normStreamer.SetNormCongestionControl(ccMode); if (NormStreamer::NORM_FIXED == ccMode) normStreamer.SetNormTxRate(txRate); if (NULL != mcastIface) normStreamer.SetNormMulticastInterface(mcastIface); if (trace) normStreamer.SetNormMessageTrace(true); // TBD - set NORM session parameters normStreamer.Start(send, recv); // TBD - add WIN32 support using win32InputHandler code // and MsgWaitForMultipleObjectsEx() instead of select() int normfd = NormGetDescriptor(normInstance); // Get input/output descriptors and set to non-blocking i/o int inputfd = normStreamer.GetInputDescriptor(); int outputfd = normStreamer.GetOutputDescriptor(); if (-1 == fcntl(inputfd, F_SETFL, fcntl(inputfd, F_GETFL, 0) | O_NONBLOCK)) perror("normStreamer: fcntl(inputfd, O_NONBLOCK) error"); //if (!normStreamer.UdpRelayEnabled()) if (-1 == fcntl(outputfd, F_SETFL, fcntl(outputfd, F_GETFL, 0) | O_NONBLOCK)) perror("normStreamer: fcntl(outputfd, O_NONBLOCK) error"); fd_set fdsetInput, fdsetOutput; FD_ZERO(&fdsetInput); FD_ZERO(&fdsetOutput); #ifdef LINUX // We user timerfd on Linux for more precise timeouts int timerfd = timerfd_create(CLOCK_MONOTONIC, 0); if (timerfd < 0) { perror("normStreamer: timerfd_create() error"); return -1; } #endif // LINUX struct timeval lastTime; gettimeofday(&lastTime, NULL); struct timeval bucketTime = lastTime; while (normStreamer.IsRunning()) { int maxfd = -1; int fdMask = 0; bool waitOnNorm = false; double timeoutInterval = -1.0; if (send) { if (normStreamer.InputNeeded()) { if (normStreamer.InputReady()) { FD_CLR(inputfd, &fdsetInput); timeoutInterval = 0.0; } else { FD_SET(inputfd, &fdsetInput); if (inputfd > maxfd) maxfd = inputfd; fdMask |= 0x01; } } else { FD_CLR(inputfd, &fdsetInput); } if (normStreamer.TxPending()) { if (normStreamer.TxReady()) timeoutInterval = 0.0; else waitOnNorm = true; } } if (recv) { if (normStreamer.RxNeeded()) { if (normStreamer.RxReady()) timeoutInterval = 0.0; else waitOnNorm = true; } if (normStreamer.OutputPending()) { if (normStreamer.OutputReady()) { FD_CLR(outputfd, &fdsetOutput); if (timeoutInterval < 0.0) timeoutInterval = normStreamer.GetOutputBucketTimeout(); } else { FD_SET(outputfd, &fdsetOutput); if (outputfd > maxfd) maxfd = outputfd; fdMask |= 0x02; } } else { FD_CLR(outputfd, &fdsetOutput); } } if (waitOnNorm) { // we need to wait until NORM is tx_ready or rx_ready FD_SET(normfd, &fdsetInput); if (normfd > maxfd) maxfd = normfd; fdMask |= 0x04; } else { FD_CLR(normfd, &fdsetInput); } // Set timeout for select() ... TBD - it may be a slight // performance enhancement to skip the select() call when // the timeout needed is zero??? struct timeval timeout; struct timeval* timeoutPtr = &timeout; #ifdef LINUX if (timeoutInterval > 0.0) { // On Linux, we use the timerfd with our select() call to get // more precise timeouts than select() does alone on Linux struct timespec timeoutSpec; timeoutSpec.tv_sec = (unsigned int)timeoutInterval; timeoutSpec.tv_nsec = 1.0e+09*(timeoutInterval - (double)timeoutSpec.tv_sec); struct itimerspec timerSpec; timerSpec.it_interval.tv_sec = timerSpec.it_interval.tv_nsec = 0; timerSpec.it_value = timeoutSpec; if (0 == timerfd_settime(timerfd, 0, &timerSpec, 0)) { timeoutPtr = NULL; FD_SET(timerfd, &fdsetInput); if (outputfd > maxfd) maxfd = timerfd; fdMask |= 0x08; } else { FD_CLR(timerfd, &fdsetInput); timeout.tv_sec = (unsigned int)timeoutInterval; timeout.tv_usec = 1.0e+06*(timeoutInterval - (double)timeout.tv_sec); perror("normStreamer: timerfd_settime() error"); } } else { // No precision timing needed FD_CLR(timerfd, &fdsetInput); if (timeoutInterval < 0.0) { // We wait one second maximum for debugging purposes timeout.tv_sec = 1; timeout.tv_usec = 0; } else // if (0.0 == timeoutInterval) { timeout.tv_sec = timeout.tv_usec = 0; } } #else // non-LINUX if (timeoutInterval > 0.0) { timeout.tv_sec = (unsigned int)timeoutInterval; timeout.tv_usec = 1.0e+06*(timeoutInterval - (double)timeout.tv_sec); } else if (timeoutInterval < 0.0) { // We wait one second maximum for debugging purposes timeout.tv_sec = 1; timeout.tv_usec = 0; } else // if (0.0 == timeoutInterval) { timeout.tv_sec = timeout.tv_usec = 0; } #endif // if/else LINUX int result = select(maxfd+1, &fdsetInput, &fdsetOutput, NULL, timeoutPtr); switch (result) { case -1: switch (errno) { case EINTR: case EAGAIN: continue; default: perror("normStreamer select() error"); // TBD - stop NormStreamer break; } break; case 0: // timeout break; default: if (FD_ISSET(inputfd, &fdsetInput)) normStreamer.SetInputReady(); if (FD_ISSET(outputfd, &fdsetOutput)) normStreamer.SetOutputReady(); #ifdef LINUX if (FD_ISSET(timerfd, &fdsetInput)) { // clear the timerfd status by reading from it uint64_t expirations = 0; if (read(timerfd, &expirations, sizeof(expirations)) < 0) perror("normStreamer read(timerfd) error"); } #endif // LINUX break; } // We always clear out/handle pending NORM API events // (to keep event queue from building up) NormEvent event; while (NormGetNextEvent(normInstance, &event, false)) normStreamer.HandleNormEvent(event); struct timeval thisTime; gettimeofday(&thisTime, NULL); if (0 != normStreamer.GetOutputBucketDepth()) { // Credit output token bucket for time that has passed double interval = (double)(thisTime.tv_sec - bucketTime.tv_sec); if (thisTime.tv_usec > bucketTime.tv_usec) interval += 1.0e-06 * (thisTime.tv_usec - bucketTime.tv_usec); else interval -= 1.0e-06 * (bucketTime.tv_usec - thisTime.tv_usec); normStreamer.CreditOutputBucket(interval); bucketTime = thisTime; } // for debugging to see if anything gets "stuck" if ((thisTime.tv_sec - lastTime.tv_sec) >= 100) { if (send) fprintf(stderr, "normStreamer: inputNeeded:%d inputReady:%d txPending:%d txReady:%d inputCount:%lu txCount:%lu fdMask:%d\n", normStreamer.InputNeeded(), normStreamer.InputReady(), normStreamer.TxPending(), normStreamer.TxReady(), normStreamer.GetInputByteCount(), normStreamer.GetTxByteCount(), fdMask); if (recv) fprintf(stderr, "normStreamer: rxNeeded:%d rxReady:%d outputPending:%d outputReady:%d fdMask:%d\n", normStreamer.RxNeeded(), normStreamer.RxReady(), normStreamer.OutputPending(), normStreamer.OutputReady(), fdMask); lastTime = thisTime; } // As a result of input/output ready or NORM notification events: // 1) Recv from rx_stream if needed and ready if (normStreamer.RxNeeded() && normStreamer.RxReady()) normStreamer.RecvData(); // 2) Write any pending data to output if output is ready if (normStreamer.OutputPending() && normStreamer.OutputReady()) { if (normStreamer.OutputBucketReady()) normStreamer.WriteOutput(); } // 3) Read from input if needed and ready if (normStreamer.InputNeeded() && normStreamer.InputReady()) normStreamer.ReadInput(); // 4) Send any pending tx message if (normStreamer.TxPending() && normStreamer.TxReady()) normStreamer.SendData(); } // end while(normStreamer.IsRunning() #ifdef LINUX close(timerfd); #endif // LINUX fflush(stderr); close(normStreamer.GetOutputDescriptor()); // TBD - do this in the destructor? NormDestroyInstance(normInstance); fprintf(stderr, "normStreamer exiting ...\n"); return 0; } // end main()