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

2373 lines
88 KiB
C++
Executable File
Raw Blame History

#include "normApi.h"
#include "protoSocket.h"
#include <stdio.h> // for printf(), etc
#include <stdlib.h> // for srand()
#include <string.h> // for strrchr(), memset(), etc
#include <sys/time.h> // for gettimeofday()
#include <arpa/inet.h> // for htons()
#include <fcntl.h> // for, well, fnctl()
#include <errno.h> // obvious child
#include <assert.h> // embarrassingly obvious
#include <sys/mman.h> // Memory Lock.
#include <sched.h> // Adjust scheduler (linux)
#include <sys/resource.h> // for setpriority() stuff
#ifdef LINUX
#include <sys/timerfd.h>
#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);
}
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);
}
// 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(&currentTime, 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);
}
else
{
NormSetWatermark(norm_session, tx_stream);
}
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:
TRACE("NORM_TX_WATERMARK_COMPLETED ...\n");
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 <nodeIdInteger> {send|recv} [addr <addr>[/<port>]]\n"
" [interface <name>] [loopback] [info] [ptos <value>] [ex]\n"
" [cc|cce|ccl|rate <bitsPerSecond>]\n"
" [ack auto|<node1>[,<node2>,...]]\n"
" [flush {none|passive|active}]\n"
" [listen [<mcastAddr>/]<port>] [linterface <name>]\n"
" [relay <dstAddr>/<port>] [limit [<rate>/]<depth>]\n"
" [output <device>] [boost] [debug <level>] [trace]\n"
" [log <logfile>] [segment <bytes>] [block <count>]\n"
" [parity <count>] [auto <count>]\n"
" [insockbuffer <bytes>] [outsockbuffer <bytes>]\n"
" [txsockbuffer <bytes>] [rxsockbuffer <bytes>]\n"
" [streambuffer <bytes>]\n"
" [check64 | check32] [omit] [silent]\n"
" [txloss <lossFraction>] [rxloss <lossFraction>]\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 <nodeId> -- 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 <addr>[/<port>] -- specifies the network address over which to send/receive NORM protocol\n"
" interface <name> -- 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 <value> -- 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 <bitsPerSecond> -- 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 [<nodeId list>] -- Instructs sender to request positive acknowledgement from listed receiver nodes\n"
" flush [<flushMode>] -- 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 [<addr>/]<port> -- 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 <name> -- 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 <dstAddr>/<port> -- Specifies the address/port for which to relay (as UDP datagrams) received messages\n"
" limit [<rate>/]<depth> -- 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 <rate> is in units of bits/second and the <depth> 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 <bytes> -- Specifies the size of the 'listen' UDP socket buffer (optional).\n"
" outsockbuffer <bytes> -- Specifies the size of the 'relay' UDP socket buffer (optional).\n"
" txsockbuffer <bytes> -- Specifies the size of the NORM/UDP transmit socket buffer (optional).\n"
" rxsockbuffer <bytes> -- Specifies the size of the NORM/UDP receive socket buffer (optional).\n"
" streambuffer <bytes> -- 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' <nodeId> 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' <mode>!\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' <bitsPerSecond> 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' <name>!\n");
Usage();
return -1;
}
mcastIface = argv[i++];
}
else if (0 == strncmp(cmd, "linterface", len))
{
if (i >= argc)
{
fprintf(stderr, "normStreamer error: missing 'linterface' <name>!\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 [<rate>/<size>] 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' <level>!\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' <fileName>!\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
TRACE("creating instance ...\n");
NormInstanceHandle normInstance = NormCreateInstance(boostPriority);
TRACE("restarting instance ...\n");
//NormRestartInstance(normInstance); // xxx
TRACE("instance restarted.\n");
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()
Reference in New Issue View Git Blame Copy Permalink