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/normMsgr.cpp

1329 lines
43 KiB
C++
Raw Blame History

#include "normApi.h"
#include <unistd.h> // for read() and write()
#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
#define SHOOT_NOW 1
#include "protoCheck.h"
class NormMsgr
{
public:
NormMsgr();
~NormMsgr();
// some day build these directly into NORM API
enum CCMode {NORM_FIXED, NORM_CC, NORM_CCE, NORM_CCL};
enum
{
MSG_SIZE_MAX = 65535,
MSG_HEADER_SIZE = 2
};
// helper class and methods for message management
class MessageQueue;
class Message
{
friend class MessageQueue;
public:
Message();
Message(char* buffer, unsigned int size);
~Message();
bool Init(unsigned int size);
void Destroy();
unsigned int GetSize() const
{return msg_size;}
const char* GetHeader() const
{return msg_header;}
char* AccessHeader()
{return msg_header;}
const char* GetBuffer() const
{return msg_buffer;}
char* AccessBuffer()
{return msg_buffer;}
void ResetIndex(unsigned int index = 0)
{msg_index = index;}
void IncrementIndex(unsigned int count)
{msg_index += count;}
unsigned int GetIndex() const
{return msg_index;}
bool IsComplete() const
{return (msg_index == (MSG_HEADER_SIZE + msg_size));}
private:
unsigned int msg_size;
unsigned int msg_index;
char msg_header[MSG_HEADER_SIZE]; // this could be externalized to NormMsgr::input_msg_header / output_msg_header members
char* msg_buffer;
Message* prev;
Message* next;
}; // end class NormMsgr::Message
class MessageQueue
{
public:
MessageQueue();
~MessageQueue();
bool IsEmpty() const
{return (NULL == head);}
void Prepend(Message& msg);
void Append(Message& msg);
void Remove(Message& msg);
Message* RemoveHead();
Message* RemoveTail();
void Destroy();
unsigned int GetCount() const
{return msg_count;}
private:
Message* head;
Message* tail;
unsigned int msg_count;
}; // end class NormMsgr::MessageQueue
Message* NewMessage(unsigned int size);
void Destroy();
bool OpenNormSession(NormInstanceHandle instance,
const char* addr,
unsigned short port,
NormNodeId nodeId);
void CloseNormSession();
void SetNormCongestionControl(CCMode ccMode);
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 SetLoopback(bool state)
{
loopback = state;
if (NORM_SESSION_INVALID != norm_session)
NormSetMulticastLoopback(norm_session, state);
}
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);
norm_acking = true; // invoke ack-based flow control
}
void SetFlushing(bool state)
{norm_flushing = state;}
bool Start(bool sender, bool receiver);
void Stop()
{is_running = false;}
bool IsRunning() const
{return is_running;}
unsigned long GetSentCount()
{return sent_count;}
void HandleNormEvent(const NormEvent& event);
// Sender methods
FILE* GetInputFile() const
{return input_file;}
int GetInputDescriptor() const
{return input_fd;}
bool InputReady() const
{return input_ready;}
void SetInputReady()
{input_ready = true;}
bool InputNeeded() const
{return input_needed;}
bool InputMessageReady() const
{return ((NULL != input_msg) && !input_needed);}
bool ReadInput();
bool TxReady() const;
bool SendMessage();
bool EnqueueMessageObject();
// Receiver methods
void SetOutputFile(FILE* filePtr)
{
output_file = filePtr;
output_fd = fileno(filePtr);
}
FILE* GetOutputFile() const
{return output_file;}
int GetOutputDescriptor() const
{return output_fd;}
void SetOutputReady()
{output_ready = true;}
bool OutputReady() const
{return output_ready;}
bool OutputPending() const
{return (NULL != output_msg);}
bool RxNeeded() const
{return rx_needed;}
bool WriteOutput();
void OmitHeader(bool state)
{omit_header = state;}
// These can only be called post-OpenNormSession
void SetSilentReceiver(bool state)
{NormSetSilentReceiver(norm_session, true);}
void SetTxLoss(double txloss)
{NormSetTxLoss(norm_session, txloss);}
private:
bool is_running;
FILE* input_file; // stdin by default
int input_fd;
bool input_ready;
bool input_needed; //
bool input_finished;
Message* input_msg; // current input message being read/sent*
MessageQueue input_msg_list; // list of enqueued messages (in norm sender cache)
NormSessionHandle norm_session;
bool is_multicast;
bool loopback;
unsigned int norm_tx_queue_max; // max number of objects that can be enqueued at once
unsigned int norm_tx_queue_count; // count of unacknowledged enqueued objects (TBD - optionally track size too)
bool norm_flow_control_pending;
bool norm_tx_vacancy;
bool norm_acking;
bool norm_flushing;
NormObjectHandle norm_flush_object;
NormObjectHandle norm_last_object;
unsigned long sent_count;
bool rx_needed;
FILE* output_file;
int output_fd;
bool output_ready;
Message* output_msg;
MessageQueue output_msg_queue;
MessageQueue temp_msg_queue;
// 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;
}; // end class NormMsgr
NormMsgr::NormMsgr()
: input_file(stdin), input_fd(fileno(stdin)), input_ready(false), input_needed(false), input_finished(false),
input_msg(NULL), norm_session(NORM_SESSION_INVALID), is_multicast(false), loopback(false),
norm_tx_queue_max(8192), norm_tx_queue_count(0),
norm_flow_control_pending(false), norm_tx_vacancy(true), norm_acking(false),
norm_flushing(true), norm_flush_object(NORM_OBJECT_INVALID), norm_last_object(NORM_OBJECT_INVALID),
sent_count(0), rx_needed(false), output_file(stdout), output_fd(fileno(stdout)),
output_ready(true), output_msg(NULL),
omit_header(false), rx_silent(false), tx_loss(0.0)
{
}
NormMsgr::~NormMsgr()
{
Destroy();
}
void NormMsgr::Destroy()
{
if (NULL != input_msg)
{
delete input_msg;
input_msg = NULL;
}
input_msg_list.Destroy();
if (NULL != output_msg)
{
delete output_msg;
delete output_msg;
}
output_msg_queue.Destroy();
}
bool NormMsgr::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, "normMsgr 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())
fprintf(stderr, "setting rx cache limit to %u\n", norm_tx_queue_max);
if (norm_tx_queue_max > 65535/2) norm_tx_queue_max = 65535/2;
NormSetRxCacheLimit(norm_session, norm_tx_queue_max);
NormSetDefaultSyncPolicy(norm_session, NORM_SYNC_ALL);
if (!is_multicast)
NormSetDefaultUnicastNack(norm_session, true);
NormSetTxCacheBounds(norm_session, 10*1024*1024, norm_tx_queue_max, norm_tx_queue_max);
//NormSetMessageTrace(norm_session, true);
//NormSetTxRobustFactor(norm_session, 20);
return true;
} // end NormMsgr::OpenNormSession()
void NormMsgr::CloseNormSession()
{
if (NORM_SESSION_INVALID == norm_session) return;
NormDestroySession(norm_session);
norm_session = NORM_SESSION_INVALID;
} // end NormMsgr::CloseNormSession()
void NormMsgr::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:
NormSetEcnSupport(norm_session, false, false, false);
break;
}
if (NORM_FIXED != ccMode)
NormSetCongestionControl(norm_session, true);
else
NormSetCongestionControl(norm_session, false);
} // end NormMsgr::SetNormCongestionControl()
bool NormMsgr::Start(bool sender, bool receiver)
{
fprintf(stderr, "enter NormMsgr::Start() ...\n");
// TBD - make these command-line accessible
unsigned int bufferSize = 64*1024*1024;
unsigned int segmentSize = 1400;
unsigned int blockSize = 64;
unsigned int numParity = 0;
unsigned int txSockBufferSize = 4*1024*1024;
unsigned int rxSockBufferSize = 6*1024*1024;
if (receiver)
{
if (!NormStartReceiver(norm_session, bufferSize))
{
fprintf(stderr, "normMsgr error: unable to start NORM receiver\n");
return false;
}
// Note: NormPreallocateRemoteSender() MUST be called AFTER NormStartReceiver()
NormPreallocateRemoteSender(norm_session, bufferSize, segmentSize, blockSize, numParity, bufferSize);
NormSetRxSocketBuffer(norm_session, rxSockBufferSize);
rx_needed = true;
fprintf(stderr, "normMsgr: receiver ready.\n");
}
if (sender)
{
if (norm_acking)
{
// ack-based flow control enabled on command-line,
// so disable timer-based flow control
NormSetFlowControl(norm_session, 0.0);
}
NormSetGrttEstimate(norm_session, 0.001);
//NormSetGrttMax(norm_session, 0.100);
NormSetBackoffFactor(norm_session, 0);
// 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 (!NormStartSender(norm_session, instanceId, bufferSize, segmentSize, blockSize, numParity))
{
fprintf(stderr, "normMsgr error: unable to start NORM sender\n");
if (receiver) NormStopReceiver(norm_session);
return false;
}
//NormSetAutoParity(norm_session, 2);
NormSetTxSocketBuffer(norm_session, txSockBufferSize);
input_needed = true;
}
//ProtoCheckResetLogging();
is_running = true;
return true;
} // end NormMsgr::Start();
// Returns "true" when a complete message has been read
bool NormMsgr::ReadInput()
{
assert(input_needed);
while (input_needed)
{
size_t readLength;
char* bufferPtr;
if ((NULL == input_msg) || (input_msg->GetIndex() < MSG_HEADER_SIZE))
{
if (NULL == input_msg)
{
// Allocate a new message/buffer for input
if (NULL == (input_msg = new Message()))
{
perror("normMsgr new Message error");
Stop(); // fatal out of memory error
return false;
}
}
// need to read msg "length" header
readLength = MSG_HEADER_SIZE - input_msg->GetIndex();
bufferPtr = input_msg->AccessHeader() + input_msg->GetIndex();
}
else
{
unsigned int offset = input_msg->GetIndex() - MSG_HEADER_SIZE;
readLength = input_msg->GetSize() - offset;
bufferPtr = input_msg->AccessBuffer() + offset;
}
ssize_t result = read(input_fd, bufferPtr, readLength);
if (result > 0)
{
input_msg->IncrementIndex(result);
if ((size_t)result < readLength)
{
// Still need more input
// (wait for next input notification to read more)
return false;
}
else if (MSG_HEADER_SIZE == input_msg->GetIndex())
{
// 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_msg->GetHeader(), MSG_HEADER_SIZE);
msgSize = ntohs(msgSize) - MSG_HEADER_SIZE;
if (!input_msg->Init(msgSize))
{
perror("normMsgr: input message initialization error");
Stop(); // fatal out of memory error
return false;
}
}
if (input_msg->IsComplete()) // have read complete header and message
{
// We have now read in the complete message
input_needed = false;
}
}
else if (0 == result)
{
// end-of-file reached, TBD - trigger final flushing and wrap-up
fprintf(stderr, "normMsgr: input end-of-file detected (last:%p)...\n", norm_last_object);
delete input_msg;
input_msg = NULL;
input_ready = false;
input_needed = false;
input_finished = true;
if (norm_acking)
{
if (NORM_OBJECT_INVALID == norm_last_object)
is_running = false; // everything sent and acked
else if (!norm_flushing)
NormSetWatermark(norm_session, norm_last_object, true);
}
}
else // result < 0
{
switch (errno)
{
case EINTR:
continue; // interupted, try again
case EAGAIN:
// input starved, wait for next notification
break;
default:
perror("normMsgr error reading input");
break;
}
input_ready = false;
return false;
}
}
return true;
} // end NormMsgr::ReadInput()
bool NormMsgr::WriteOutput()
{
while (NULL != output_msg)
{
size_t writeLength;
const char* bufferPtr;
if (output_msg->GetIndex() < MSG_HEADER_SIZE)
{
writeLength = MSG_HEADER_SIZE - output_msg->GetIndex();
bufferPtr = output_msg->GetHeader() + output_msg->GetIndex();
}
else
{
unsigned int offset = output_msg->GetIndex() - MSG_HEADER_SIZE;
writeLength = output_msg->GetSize() - offset;
bufferPtr = output_msg->GetBuffer() + offset;
}
ssize_t result = write(output_fd, bufferPtr, writeLength);
if (result >= 0)
{
output_msg->IncrementIndex(result);
if ((size_t)result < writeLength)
output_ready = false; // blocked, wait for output notification
}
else
{
switch (errno)
{
case EINTR:
continue; // interupted, try again
case EAGAIN:
// input starved, wait for next notification
output_ready = false;
break;
default:
perror("normMsgr error writing output");
break;
}
return false;
}
if (output_msg->IsComplete())
{
fflush(output_file);
//delete output_msg;
temp_msg_queue.Append(*output_msg); // cache for debugging purposes
if (temp_msg_queue.GetCount() > 32)
{
//fprintf(stderr, "deleting cached recv'd message ...\n");
delete temp_msg_queue.RemoveHead();
}
output_msg = output_msg_queue.RemoveHead();
if (NULL != output_msg)
{
if (omit_header)
output_msg->ResetIndex(MSG_HEADER_SIZE);
else
output_msg->ResetIndex();
}
else
{
rx_needed = true;
}
break;
}
}
return true;
} // end NormMsgr::WriteOutput()
bool NormMsgr::TxReady() const
{
// This returns true if new tx data can be enqueued to NORM
// This is based on the state with respect to prior successful data
// data enqueuing (or stream writing) and NORM_TX_QUEUE_EMPTY or
// NORM_TX_QUEUE_VACANCY notifications (tracked by the "norm_tx_vacancy" variable,
// _and_ (if ack-based flow control is enabled) the norm_tx_queue_count or
// norm_stream_buffer_count status.
if (norm_tx_vacancy)
{
if (norm_tx_queue_count >= norm_tx_queue_max)
return false; // still waiting for ACK
return true;
}
else
{
return false;
}
} // end NormMsgr::NormTxReady()
bool NormMsgr::SendMessage()
{
if (EnqueueMessageObject())
{
// Our buffered message was sent, so reset input indices
// and request next message from input
input_msg_list.Append(*input_msg);
input_msg = NULL;
input_needed = true;
sent_count++;
return true;
}
// else will be prompted to retry by NORM event (queue vacancy, watermark completion)
return false;
} // end NormMsgr::SendMessage()
bool NormMsgr::EnqueueMessageObject()
{
if (norm_acking)
{
assert(norm_tx_queue_count < norm_tx_queue_max);
if (norm_tx_queue_count >= norm_tx_queue_max)
return false;
}
// Enqueue the message data for transmission
NormObjectHandle object = NormDataEnqueue(norm_session, input_msg->AccessBuffer(), input_msg->GetSize());
if (NORM_OBJECT_INVALID == object)
{
// This might happen if a non-acking receiver is present and
// has nacked for the oldest object in the queue even if all
// of our acking receivers have acknowledged it.
fprintf(stderr, "NO VACANCY count:%u max:%u\n", norm_tx_queue_count, norm_tx_queue_max);
norm_tx_vacancy = false;
return false;
}
//NormObjectRetain(object);
NormObjectSetUserData(object, input_msg); // so we can remove/delete upon purge
if (norm_acking)
{
// ack-based flow control has been enabled
norm_tx_queue_count++;
if (!norm_flow_control_pending && (norm_tx_queue_count >= (norm_tx_queue_max / 2)))
{
NormSetWatermark(norm_session, object, true); // overrideFlush == true
norm_last_object = object;
norm_flow_control_pending = true;
}
else if (norm_flushing) // per-message acking
{
#ifdef SHOOT_FIRST
NormSetWatermark(norm_session, object, true);
norm_last_object = object;
#else // ACK_LATER
if (norm_flow_control_pending)
{
norm_flush_object = object; // will be used as watermark upon flow control ack
}
else
{
NormSetWatermark(norm_session, object, true);
norm_last_object = object;
}
#endif // SHOOT_FIRST/ACK_LATER
}
else
{
//norm_last_object = object;
}
}
return true;
} // end NormMsgr::EnqueueMessageObject()
void NormMsgr::HandleNormEvent(const NormEvent& event)
{
switch (event.type)
{
case NORM_TX_QUEUE_EMPTY:
case NORM_TX_QUEUE_VACANCY:
norm_tx_vacancy = true;
break;
case NORM_GRTT_UPDATED:
//fprintf(stderr, "new GRTT = %lf\n", NormGetGrttEstimate(norm_session));
break;
case NORM_TX_WATERMARK_COMPLETED:
if (NORM_ACK_SUCCESS == NormGetAckingStatus(norm_session))
{
//fprintf(stderr, "WATERMARK COMPLETED\n");
norm_last_object = NORM_OBJECT_INVALID;
if (norm_flow_control_pending)
{
norm_tx_queue_count -= (norm_tx_queue_max / 2);
norm_flow_control_pending = false;
if (NORM_OBJECT_INVALID != norm_flush_object)
{
NormSetWatermark(norm_session, norm_flush_object, true);
norm_last_object = norm_flush_object;
norm_flush_object = NORM_OBJECT_INVALID;
}
}
if (input_finished && (NORM_OBJECT_INVALID == norm_last_object))
is_running = false;
}
else
{
// TBD - we could see who did and how didn't ACK and possibly remove them
// from our acking "membership". For now, we are infinitely
// persistent by resetting watermark ack request without clearing
// flow control
if (NORM_OBJECT_INVALID == norm_flush_object)
{
NormResetWatermark(norm_session);
}
else // might as request ack for most recent enqueued object
{
NormSetWatermark(norm_session, norm_flush_object, true);
norm_last_object = norm_flush_object;
norm_flush_object = NORM_OBJECT_INVALID;
}
}
break;
case NORM_TX_OBJECT_PURGED:
{
NormDataDetachData(event.object);
Message* msg = (Message*)NormObjectGetUserData(event.object);
if(event.object == norm_flush_object)
norm_flush_object = NORM_OBJECT_INVALID;
if (NULL != msg)
{
input_msg_list.Remove(*msg);
delete msg;
}
//NormObjectRelease(event.object);
//fprintf(stderr, "normMsgr LOGGING ALLOCATIONS\n");
//ProtoCheckLogAllocations(stderr);
break;
}
case NORM_REMOTE_SENDER_INACTIVE:
//fprintf(stderr, "REMOTE SENDER INACTIVE node: %u\n", NormNodeGetId(event.sender));
//NormNodeDelete(event.sender);
//logAllocs = true;
break;
case NORM_RX_OBJECT_ABORTED:
//fprintf(stderr, "NORM_RX_OBJECT_ABORTED\n");// %hu\n", NormObjectGetTransportId(event.object));
break;
case NORM_RX_OBJECT_COMPLETED:
{
sent_count++;
char* data = NormDataDetachData(event.object);
if (NULL != data)
{
Message* msg = new Message(data, NormObjectGetSize(event.object));
if (NULL == msg)
{
perror("normMsgr: new Message() error");
delete[] data; // TBD - may need to finally implement NormDelete() function!
// TBD Stop() as a fatal out of memory error?
break;
}
if (NULL == output_msg)
{
output_msg = msg;
if (omit_header)
output_msg->ResetIndex(MSG_HEADER_SIZE);
else
output_msg->ResetIndex();
}
else
{
output_msg_queue.Append(*msg);
}
rx_needed = false;
}
// else TBD - "termination" info-only object?
break;
}
default:
break;
}
//NormReleasePreviousEvent(NormGetInstance(norm_session));
} // end NormMsgr::HandleNormEvent()
void Usage()
{
fprintf(stderr, "Usage: normMsgr id <nodeId> {send &| recv} [addr <addr>[/<port>]][ack <node1>[,<node2>,...]\n"
" [cc|cce|ccl|rate <bitsPerSecond>][interface <name>][debug <level>][trace]\n"
" [flush {none|active}][omit][silent][txloss <lossFraction>]\n");
}
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;
NormNodeId ackingNodeList[256];
unsigned int ackingNodeCount = 0;
bool flushing = false;
double txRate = 0.0; // used for non-default NORM_FIXED ccMode
NormMsgr::CCMode ccMode = NormMsgr::NORM_CC;
const char* mcastIface = NULL;
int debugLevel = 0;
const char* debugLog = NULL; // stderr by default
bool trace = false;
bool omitHeaderOnOutput = false;
bool silentReceiver = false;
double txloss = 0.0;
bool loopback = false;
NormMsgr normMsgr;
// Parse command-line
int i = 1;
while (i < argc)
{
const char* cmd = argv[i++];
size_t len = strlen(cmd);
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, "addr", len))
{
if (i >= argc)
{
fprintf(stderr, "nodeMsgr 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, "output", len))
{
if (i >= argc)
{
fprintf(stderr, "normMsgr error: missing output 'device' name!\n");
Usage();
return -1;
}
FILE* outfile = fopen(argv[i++], "w+");
if (NULL == outfile)
{
perror("normMsgr output device fopen() error");
Usage();
return -1;
}
normMsgr.SetOutputFile(outfile);
}
else if (0 == strncmp(cmd, "id", len))
{
if (i >= argc)
{
fprintf(stderr, "nodeMsgr 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, "nodeMsgr error: missing 'id' <nodeId> value!\n");
Usage();
return -1;
}
const char* alist = argv[i++];
while ((NULL != alist) && (*alist != '\0'))
{
// TBD - Do we need to skip leading white space?
int id;
if (1 != sscanf(alist, "%d", &id))
{
fprintf(stderr, "nodeMsgr 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"))
{
flushing = false;
}
else if (0 == strcmp(mode, "passive"))
{
flushing = false;
}
else if (0 == strcmp(mode, "active"))
{
flushing = true;
}
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, "nodeMsgr error: missing 'rate' <bitsPerSecond> value!\n");
Usage();
return -1;
}
if (1 != sscanf(argv[i++], "%lf", &txRate))
{
fprintf(stderr, "nodeMsgr error: invalid transmit rate!\n");
Usage();
return -1;
}
// set fixed-rate operation
ccMode = NormMsgr::NORM_FIXED;
}
else if (0 == strcmp(cmd, "cc"))
{
ccMode = NormMsgr::NORM_CC;
}
else if (0 == strcmp(cmd, "cce"))
{
ccMode = NormMsgr::NORM_CCE;
}
else if (0 == strcmp(cmd, "ccl"))
{
ccMode = NormMsgr::NORM_CCL;
}
else if (0 == strncmp(cmd, "interface", len))
{
if (i >= argc)
{
fprintf(stderr, "nodeMsgr error: missing 'interface' <name>!\n");
Usage();
return -1;
}
mcastIface = argv[i++];
}
else if (0 == strncmp(cmd, "omit", len))
{
omitHeaderOnOutput = true;
}
else if (0 == strncmp(cmd, "silent", len))
{
silentReceiver = true;
}
else if (0 == strncmp(cmd, "txloss", len))
{
if (1 != sscanf(argv[i++], "%lf", &txloss))
{
fprintf(stderr, "nodeMsgr error: invalid 'txloss' value!\n");
Usage();
return -1;
}
}
else if (0 == strncmp(cmd, "debug", len))
{
if (i >= argc)
{
fprintf(stderr, "nodeMsgr error: missing 'debug' <level>!\n");
Usage();
return -1;
}
debugLevel = atoi(argv[i++]);
}
else if (0 == strncmp(cmd, "log", len))
{
if (i >= argc)
{
fprintf(stderr, "nodeMsgr error: missing 'log' <fileName>!\n");
Usage();
return -1;
}
debugLog = argv[i++];
}
else if (0 == strncmp(cmd, "trace", len))
{
trace = true;
}
else if (0 == strncmp(cmd, "help", len))
{
Usage();
return 0;
}
else
{
fprintf(stderr, "nodeMsgr error: invalid command \"%s\"!\n", cmd);
Usage();
return -1;
}
}
if (!send && !recv)
{
fprintf(stderr, "normMsgr error: not configured to send or recv!\n");
Usage();
return -1;
}
if (NORM_NODE_NONE == nodeId)
{
fprintf(stderr, "normMsgr error: no local 'id' provided!\n");
Usage();
return -1;
}
// TBD - should provide more error checking of calls
NormInstanceHandle normInstance = NormCreateInstance();
NormSetDebugLevel(debugLevel);
if (NULL != debugLog)
NormOpenDebugLog(normInstance, debugLog);
normMsgr.SetLoopback(loopback);
normMsgr.SetFlushing(flushing);
if (omitHeaderOnOutput) normMsgr.OmitHeader(true);
if (!normMsgr.OpenNormSession(normInstance, sessionAddr, sessionPort, (NormNodeId)nodeId))
{
fprintf(stderr, "normMsgr error: unable to open NORM session\n");
NormDestroyInstance(normInstance);
return -1;
}
if (silentReceiver) normMsgr.SetSilentReceiver(true);
if (txloss > 0.0) normMsgr.SetTxLoss(txloss);
for (unsigned int i = 0; i < ackingNodeCount; i++)
normMsgr.AddAckingNode(ackingNodeList[i]);
normMsgr.SetNormCongestionControl(ccMode);
if (NormMsgr::NORM_FIXED == ccMode)
normMsgr.SetNormTxRate(txRate);
if (NULL != mcastIface)
normMsgr.SetNormMulticastInterface(mcastIface);
if (trace) normMsgr.SetNormMessageTrace(true);
// TBD - set NORM session parameters
normMsgr.Start(send, recv);
int normfd = NormGetDescriptor(normInstance);
// Get input/output descriptors and set to non-blocking i/o
int inputfd = normMsgr.GetInputDescriptor();
if (-1 == fcntl(inputfd, F_SETFL, fcntl(inputfd, F_GETFL, 0) | O_NONBLOCK))
perror("normMsgr: fcntl(inputfd, O_NONBLOCK) error");
int outputfd = normMsgr.GetOutputDescriptor();
if (-1 == fcntl(outputfd, F_SETFL, fcntl(outputfd, F_GETFL, 0) | O_NONBLOCK))
perror("normMsgr: fcntl(outputfd, O_NONBLOCK) error");
fd_set fdsetInput, fdsetOutput;
FD_ZERO(&fdsetInput);
FD_ZERO(&fdsetOutput);
struct timeval lastTime;
gettimeofday(&lastTime, NULL);
while (normMsgr.IsRunning())
{
int maxfd = -1;
// Only wait on NORM if needed for tx readiness
bool waitOnNorm = true;
if (!(normMsgr.RxNeeded() || normMsgr.InputMessageReady()))
waitOnNorm = false; // no need to wait
else if (normMsgr.InputMessageReady() && normMsgr.TxReady())
waitOnNorm = false; // no need to wait if already tx ready
if (waitOnNorm)
{
maxfd = normfd;
FD_SET(normfd, &fdsetInput);
}
else
{
FD_CLR(normfd, &fdsetInput);
}
if (normMsgr.InputNeeded() && !normMsgr.InputReady())
{
FD_SET(inputfd, &fdsetInput);
if (inputfd > maxfd) maxfd = inputfd;
}
else
{
FD_CLR(inputfd, &fdsetInput);
}
if (normMsgr.OutputPending() && !normMsgr.OutputReady())
{
FD_SET(outputfd, &fdsetOutput);
if (outputfd > maxfd) maxfd = outputfd;
}
else
{
FD_CLR(outputfd, &fdsetOutput);
}
if (maxfd >= 0)
{
struct timeval timeout;
timeout.tv_sec = 1;
timeout.tv_usec = 0;
int result = select(maxfd+1, &fdsetInput, &fdsetOutput, NULL, &timeout);
switch (result)
{
case -1:
switch (errno)
{
case EINTR:
case EAGAIN:
continue;
default:
perror("normMsgr select() error");
// TBD - stop NormMsgr
break;
}
break;
case 0:
// shouldn't occur for now (no timeout)
//fprintf(stderr, "normMsgr timeout ...\n");
continue;
default:
if (FD_ISSET(inputfd, &fdsetInput))
{
normMsgr.SetInputReady();
}
/*if (FD_ISSET(normfd, &fdsetInput))
{
NormEvent event;
while (NormGetNextEvent(normInstance, &event, false))
normMsgr.HandleNormEvent(event);
}*/
if (FD_ISSET(outputfd, &fdsetOutput))
{
normMsgr.SetOutputReady();
}
break;
}
}
NormSuspendInstance(normInstance);
NormEvent event;
while (NormGetNextEvent(normInstance, &event, false))
normMsgr.HandleNormEvent(event);
// As a result of reading input or NORM notification events,
// we may be ready to read input and/or send a message if it's been read
const int LOOP_MAX = 100;
int loopCount = 0;
while (loopCount < LOOP_MAX)
{
loopCount++;
if (normMsgr.InputNeeded() && normMsgr.InputReady())
normMsgr.ReadInput();
if (normMsgr.InputMessageReady() && normMsgr.TxReady())
normMsgr.SendMessage();
// and/or output a received message if we need
if (normMsgr.OutputPending() && normMsgr.OutputReady())
normMsgr.WriteOutput();
}
NormResumeInstance(normInstance);
} // end while(normMsgr.IsRunning()
NormCloseDebugLog(normInstance);
fprintf(stderr, "destroying session ...\n");
normMsgr.CloseNormSession();
fprintf(stderr, "destroying instance ...\n");
NormDestroyInstance(normInstance);
normMsgr.Destroy();
fprintf(stderr, "normMsgr exiting ...\n");
} // end main()
NormMsgr::Message::Message()
: msg_size(0), msg_index(0), msg_buffer(NULL),
prev(NULL), next(NULL)
{
}
NormMsgr::Message::Message(char* buffer, unsigned int size)
: msg_size(size), msg_index(size+MSG_HEADER_SIZE), msg_buffer(buffer),
prev(NULL), next(NULL)
{
assert(2 == MSG_HEADER_SIZE);
uint16_t msgSize = size + MSG_HEADER_SIZE;
msgSize = htons(msgSize);
memcpy(msg_header, &msgSize, MSG_HEADER_SIZE);
}
NormMsgr::Message::~Message()
{
// in future we may need to use NormDelete()
// to delete msg_buffer if it came from NORM
if (NULL != msg_buffer)
{
//fprintf(stderr, "deleting msg_buffer ...\n");
delete[] msg_buffer;
msg_buffer = NULL;
}
}
bool NormMsgr::Message::Init(unsigned int size)
{
if ((NULL != msg_buffer) && (size != msg_size))
{
delete[] msg_buffer;
msg_buffer = NULL;
}
if (NULL == msg_buffer)
msg_buffer = new char[size];
if (NULL != msg_buffer)
{
assert(2 == MSG_HEADER_SIZE);
uint16_t msgSize = size + MSG_HEADER_SIZE;
msgSize = htons(msgSize);
memcpy(msg_header, &msgSize, MSG_HEADER_SIZE);
msg_index = MSG_HEADER_SIZE; // empty message
msg_size = size;
return true;
}
else
{
msg_index = msg_size = 0;
return false;
}
} // end NormMsgr::Message::Init()
NormMsgr::MessageQueue::MessageQueue()
: head(NULL), tail(NULL), msg_count(0)
{
}
NormMsgr::MessageQueue::~MessageQueue()
{
}
void NormMsgr::MessageQueue::Destroy()
{
while (NULL != head)
{
Message* msg = RemoveHead();
delete msg;
}
} // end NormMsgr::MessageQueue::Destroy()
void NormMsgr::MessageQueue::Prepend(Message& msg)
{
msg.prev = NULL;
if (NULL != head)
head->prev = &msg;
else
tail = &msg;
msg.next = head;
head = &msg;
msg_count++;
} // end NormMsgr::MessageQueue::Prepend()
void NormMsgr::MessageQueue::Append(Message& msg)
{
msg.next = NULL;
if (NULL != tail)
tail->next = &msg;
else
head = &msg;
msg.prev = tail;
tail = &msg;
msg_count++;
} // end NormMsgr::MessageQueue::Append()
void NormMsgr::MessageQueue::Remove(Message& msg)
{
if (NULL == msg.prev)
head = msg.next;
else
msg.prev->next = msg.next;
if (NULL == msg.next)
tail = msg.prev;
else
msg.next->prev = msg.prev;
msg.prev = msg.next = NULL;
msg_count--;
} // end NormMsgr::MessageQueue::Remove()
NormMsgr::Message* NormMsgr::MessageQueue::RemoveHead()
{
Message* msg = head;
if (NULL != head) Remove(*head);
return msg;
} // end NormMsgr::MessageQueue::RemoveHead()
NormMsgr::Message* NormMsgr::MessageQueue::RemoveTail()
{
Message* msg = tail;
if (NULL != tail) Remove(*tail);
return msg;
} // end NormMsgr::MessageQueue::RemoveTail()
Reference in New Issue View Git Blame Copy Permalink