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

v1.1b5

pull/2/head
Jeff Weston 2019-09-11 11:33:01 -04:00
parent 48a5d68c05
commit 25676f4b00
6 changed files with 241 additions and 93 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

206
common/normApp.cpp
View File

@ -9,6 +9,7 @@
#include <signal.h> // for SIGTERM/SIGINT handling #include <signal.h> // for SIGTERM/SIGINT handling
#include <errno.h> #include <errno.h>
#include <stdlib.h> #include <stdlib.h>
#include <ctype.h> // for "isspace()"
// Command-line application using Protolib EventDispatcher // Command-line application using Protolib EventDispatcher
class NormApp : public NormController, public ProtoApp class NormApp : public NormController, public ProtoApp
@ -22,7 +23,7 @@ class NormApp : public NormController, public ProtoApp
bool ProcessCommands(int argc, const char*const* argv); bool ProcessCommands(int argc, const char*const* argv);
void OnShutdown(); void OnShutdown();
bool ProcessCommand(const char* cmd, const char* val); bool OnCommand(const char* cmd, const char* val);
static void DoInputReady(ProtoDispatcher::Descriptor descriptor, static void DoInputReady(ProtoDispatcher::Descriptor descriptor,
ProtoDispatcher::Event theEvent, ProtoDispatcher::Event theEvent,
@ -41,6 +42,7 @@ class NormApp : public NormController, public ProtoApp
class NormObject* object); class NormObject* object);
bool OnIntervalTimeout(ProtoTimer& theTimer); bool OnIntervalTimeout(ProtoTimer& theTimer);
void OnControlEvent(ProtoSocket& theSocket, ProtoSocket::Event theEvent);
static const char* const cmd_list[]; static const char* const cmd_list[];
@ -48,6 +50,8 @@ class NormApp : public NormController, public ProtoApp
static void SignalHandler(int sigNum); static void SignalHandler(int sigNum);
#endif // UNIX #endif // UNIX
ProtoPipe control_pipe; // for remote control
bool control_remote;
NormSessionMgr session_mgr; NormSessionMgr session_mgr;
NormSession* session; NormSession* session;
NormStreamObject* tx_stream; NormStreamObject* tx_stream;
@ -90,6 +94,7 @@ class NormApp : public NormController, public ProtoApp
double tx_object_interval; double tx_object_interval;
int tx_repeat_count; int tx_repeat_count;
double tx_repeat_interval; double tx_repeat_interval;
bool tx_repeat_clear;
ProtoTimer interval_timer; ProtoTimer interval_timer;
// NormSession client-only parameters // NormSession client-only parameters
@ -108,7 +113,8 @@ class NormApp : public NormController, public ProtoApp
}; // end class NormApp }; // end class NormApp
NormApp::NormApp() NormApp::NormApp()
: session_mgr(GetTimerMgr(), GetSocketNotifier()), : control_pipe(ProtoPipe::MESSAGE), control_remote(false),
session_mgr(GetTimerMgr(), GetSocketNotifier()),
session(NULL), tx_stream(NULL), rx_stream(NULL), input(NULL), output(NULL), session(NULL), tx_stream(NULL), rx_stream(NULL), input(NULL), output(NULL),
input_index(0), input_length(0), input_active(false), input_index(0), input_length(0), input_active(false),
push_stream(false), msg_flush_mode(NormStreamObject::FLUSH_PASSIVE), push_stream(false), msg_flush_mode(NormStreamObject::FLUSH_PASSIVE),
@ -118,12 +124,16 @@ NormApp::NormApp()
segment_size(1024), ndata(32), nparity(16), auto_parity(0), extra_parity(0), segment_size(1024), ndata(32), nparity(16), auto_parity(0), extra_parity(0),
backoff_factor(NormSession::DEFAULT_BACKOFF_FACTOR), backoff_factor(NormSession::DEFAULT_BACKOFF_FACTOR),
tx_buffer_size(1024*1024), tx_buffer_size(1024*1024),
tx_object_interval(0.0), tx_repeat_count(0), tx_repeat_interval(0.0), tx_object_interval(0.0), tx_repeat_count(0), tx_repeat_interval(2.0), tx_repeat_clear(true),
rx_buffer_size(1024*1024), rx_cache_path(NULL), unicast_nacks(false), silent_client(false), rx_buffer_size(1024*1024), rx_cache_path(NULL), unicast_nacks(false), silent_client(false),
tracing(false), tx_loss(0.0), rx_loss(0.0) tracing(false), tx_loss(0.0), rx_loss(0.0)
{ {
control_pipe.SetListener(this, &NormApp::OnControlEvent);
control_pipe.SetNotifier(&GetSocketNotifier());
// Init tx_timer for 1.0 second interval, infinite repeats // Init tx_timer for 1.0 second interval, infinite repeats
session_mgr.SetController(this); session_mgr.SetController(this);
interval_timer.SetListener(this, &NormApp::OnIntervalTimeout); interval_timer.SetListener(this, &NormApp::OnIntervalTimeout);
interval_timer.SetInterval(0.0); interval_timer.SetInterval(0.0);
@ -161,8 +171,9 @@ const char* const NormApp::cmd_list[] =
"+moutput", // recv message stream output "+moutput", // recv message stream output
"+sendfile", // file/directory list to transmit "+sendfile", // file/directory list to transmit
"+interval", // delay time (sec) between files (0.0 sec default) "+interval", // delay time (sec) between files (0.0 sec default)
"+repeatcount", // How many times to repeat the file/directory list "+repeatcount", // How many times to repeat the file/directory list tx
"+rinterval", // Interval (sec) between file/directory list repeats "+rinterval", // Interval (sec) between file/directory list repeats
"-updatesOnly", // only send updated files on repeat transmission
"+rxcachedir", // recv file cache directory "+rxcachedir", // recv file cache directory
"+segment", // payload segment size (bytes) "+segment", // payload segment size (bytes)
"+block", // User data packets per FEC coding block (blockSize) "+block", // User data packets per FEC coding block (blockSize)
@ -175,27 +186,95 @@ const char* const NormApp::cmd_list[] =
"-unicastNacks", // unicast instead of multicast feedback messages "-unicastNacks", // unicast instead of multicast feedback messages
"-silentClient", // "silent" (non-nacking) client (EMCON mode) "-silentClient", // "silent" (non-nacking) client (EMCON mode)
"+processor", // receive file post processing command "+processor", // receive file post processing command
"+instance", // specify norm instance name for commands
NULL NULL
}; };
void NormApp::OnControlEvent(ProtoSocket& /*theSocket*/, ProtoSocket::Event theEvent)
{
TRACE("NormApp::OnControlEvent() ...\n");
switch (theEvent)
{
case ProtoSocket::RECV:
{
char buffer[8192];
unsigned int len = 8192;
if (control_pipe.Recv(buffer, len))
{
buffer[len] = '\0';
TRACE("norm: received command \"%s\"\n", buffer);
char* cmd = buffer;
char* val = NULL;
for (unsigned int i = 0; i < len; i++)
{
if (isspace(buffer[i]))
{
buffer[i++] = '\0';
val = buffer + i;
}
}
if (!OnCommand(cmd, val))
DMSG(0, "NormApp::OnControlEvent() bad command received\n");
}
else
{
DMSG(0, "NormApp::OnControlEvent() error receiving remote command\n");
}
break;
}
default:
DMSG(0, "NormApp::OnControlEvent() unhandled event type: %d\n", theEvent);
break;
}
} // end NormApp::OnControlEvent()
bool NormApp::ProcessCommand(const char* cmd, const char* val) bool NormApp::OnCommand(const char* cmd, const char* val)
{ {
CmdType type = CommandType(cmd); CmdType type = CommandType(cmd);
ASSERT(CMD_INVALID != type); ASSERT(CMD_INVALID != type);
unsigned int len = strlen(cmd); unsigned int len = strlen(cmd);
if ((CMD_ARG == type) && !val) if ((CMD_ARG == type) && !val)
{ {
DMSG(0, "NormApp::ProcessCommand(%s) missing argument\n", cmd); DMSG(0, "NormApp::OnCommand(%s) missing argument\n", cmd);
return false; return false;
} }
if (control_remote)
{
// Since the "instance" already exists, we send the command
// to the pre-existing "instance" instead of processing it ourself
char buffer[8192];
buffer[0] = '\0';
if (cmd) strncpy(buffer, cmd, 8191);
unsigned int len = strlen(buffer);
len = (len > 8191) ? 8191 : len;
if (val)
{
buffer[len++] = ' ';
strncat(buffer, val, 8192 - strlen(buffer));
}
len = strlen(buffer);
len = (len > 8191) ? 8191 : len;
buffer[len++] = '\0';
if (!control_pipe.Send(buffer, len))
{
DMSG(0, "NormApp::OnCommand() error sending command to remote instance\n");
return false;
}
else
{
DMSG(0, "NormApp::OnCommand() sent \"%s\" to remote\n", buffer);
}
}
if (!strncmp("debug", cmd, len)) if (!strncmp("debug", cmd, len))
{ {
int debugLevel = atoi(val); int debugLevel = atoi(val);
if ((debugLevel < 0) || (debugLevel > 12)) if ((debugLevel < 0) || (debugLevel > 12))
{ {
DMSG(0, "NormApp::ProcessCommand(segment) invalid debug level!\n"); DMSG(0, "NormApp::OnCommand(segment) invalid debug level!\n");
return false; return false;
} }
SetDebugLevel(debugLevel); SetDebugLevel(debugLevel);
@ -214,7 +293,7 @@ bool NormApp::ProcessCommand(const char* cmd, const char* val)
double txLoss = atof(val); double txLoss = atof(val);
if (txLoss < 0) if (txLoss < 0)
{ {
DMSG(0, "NormApp::ProcessCommand(txloss) invalid txRate!\n"); DMSG(0, "NormApp::OnCommand(txloss) invalid txRate!\n");
return false; return false;
} }
tx_loss = txLoss; tx_loss = txLoss;
@ -225,7 +304,7 @@ bool NormApp::ProcessCommand(const char* cmd, const char* val)
double rxLoss = atof(val); double rxLoss = atof(val);
if (rxLoss < 0) if (rxLoss < 0)
{ {
DMSG(0, "NormApp::ProcessCommand(rxloss) invalid txRate!\n"); DMSG(0, "NormApp::OnCommand(rxloss) invalid txRate!\n");
return false; return false;
} }
rx_loss = rxLoss; rx_loss = rxLoss;
@ -237,7 +316,7 @@ bool NormApp::ProcessCommand(const char* cmd, const char* val)
if (address) delete address; if (address) delete address;
if (!(address = new char[len+1])) if (!(address = new char[len+1]))
{ {
DMSG(0, "NormApp::ProcessCommand(address) allocation error:%s\n", DMSG(0, "NormApp::OnCommand(address) allocation error:%s\n",
strerror(errno)); strerror(errno));
return false; return false;
} }
@ -247,7 +326,7 @@ bool NormApp::ProcessCommand(const char* cmd, const char* val)
{ {
delete address; delete address;
address = NULL; address = NULL;
DMSG(0, "NormApp::ProcessCommand(address) missing port number!\n"); DMSG(0, "NormApp::OnCommand(address) missing port number!\n");
return false; return false;
} }
*ptr++ = '\0'; *ptr++ = '\0';
@ -256,7 +335,7 @@ bool NormApp::ProcessCommand(const char* cmd, const char* val)
{ {
delete address; delete address;
address = NULL; address = NULL;
DMSG(0, "NormApp::ProcessCommand(address) invalid port number!\n"); DMSG(0, "NormApp::OnCommand(address) invalid port number!\n");
return false; return false;
} }
port = portNum; port = portNum;
@ -266,7 +345,7 @@ bool NormApp::ProcessCommand(const char* cmd, const char* val)
int ttlTemp = atoi(val); int ttlTemp = atoi(val);
if ((ttlTemp < 1) || (ttlTemp > 255)) if ((ttlTemp < 1) || (ttlTemp > 255))
{ {
DMSG(0, "NormApp::ProcessCommand(ttl) invalid value!\n"); DMSG(0, "NormApp::OnCommand(ttl) invalid value!\n");
return false; return false;
} }
ttl = ttlTemp; ttl = ttlTemp;
@ -276,7 +355,7 @@ bool NormApp::ProcessCommand(const char* cmd, const char* val)
double txRate = atof(val); double txRate = atof(val);
if (txRate < 0) if (txRate < 0)
{ {
DMSG(0, "NormApp::ProcessCommand(rate) invalid txRate!\n"); DMSG(0, "NormApp::OnCommand(rate) invalid txRate!\n");
return false; return false;
} }
tx_rate = txRate; tx_rate = txRate;
@ -294,7 +373,7 @@ bool NormApp::ProcessCommand(const char* cmd, const char* val)
} }
else else
{ {
DMSG(0, "NormApp::ProcessCommand(cc) invalid option!\n"); DMSG(0, "NormApp::OnCommand(cc) invalid option!\n");
return false; return false;
} }
if (session) session->SetCongestionControl(cc_enable); if (session) session->SetCongestionControl(cc_enable);
@ -307,7 +386,7 @@ bool NormApp::ProcessCommand(const char* cmd, const char* val)
} }
else if (!(input = fopen(val, "rb"))) else if (!(input = fopen(val, "rb")))
{ {
DMSG(0, "NormApp::ProcessCommand(input) fopen() error: %s\n", DMSG(0, "NormApp::OnCommand(input) fopen() error: %s\n",
strerror(errno)); strerror(errno));
return false; return false;
} }
@ -316,7 +395,7 @@ bool NormApp::ProcessCommand(const char* cmd, const char* val)
#ifdef UNIX #ifdef UNIX
// Set input non-blocking // Set input non-blocking
if(-1 == fcntl(fileno(input), F_SETFL, fcntl(fileno(input), F_GETFL, 0) | O_NONBLOCK)) if(-1 == fcntl(fileno(input), F_SETFL, fcntl(fileno(input), F_GETFL, 0) | O_NONBLOCK))
perror("NormApp::ProcessCommand(input) fcntl(F_SETFL(O_NONBLOCK)) error"); perror("NormApp::OnCommand(input) fcntl(F_SETFL(O_NONBLOCK)) error");
#endif // UNIX #endif // UNIX
} }
else if (!strncmp("output", cmd, len)) else if (!strncmp("output", cmd, len))
@ -327,7 +406,7 @@ bool NormApp::ProcessCommand(const char* cmd, const char* val)
} }
else if (!(output = fopen(val, "wb"))) else if (!(output = fopen(val, "wb")))
{ {
DMSG(0, "NormApp::ProcessCommand(output) fopen() error: %s\n", DMSG(0, "NormApp::OnCommand(output) fopen() error: %s\n",
strerror(errno)); strerror(errno));
return false; return false;
} }
@ -343,7 +422,7 @@ bool NormApp::ProcessCommand(const char* cmd, const char* val)
} }
else if (!(input = fopen(val, "rb"))) else if (!(input = fopen(val, "rb")))
{ {
DMSG(0, "NormApp::ProcessCommand(input) fopen() error: %s\n", DMSG(0, "NormApp::OnCommand(input) fopen() error: %s\n",
strerror(errno)); strerror(errno));
return false; return false;
} }
@ -352,7 +431,7 @@ bool NormApp::ProcessCommand(const char* cmd, const char* val)
#ifdef UNIX #ifdef UNIX
// Set input non-blocking // Set input non-blocking
if(-1 == fcntl(fileno(input), F_SETFL, fcntl(fileno(input), F_GETFL, 0) | O_NONBLOCK)) if(-1 == fcntl(fileno(input), F_SETFL, fcntl(fileno(input), F_GETFL, 0) | O_NONBLOCK))
perror("NormApp::ProcessCommand(input) fcntl(F_SETFL(O_NONBLOCK)) error"); perror("NormApp::OnCommand(input) fcntl(F_SETFL(O_NONBLOCK)) error");
#endif // UNIX #endif // UNIX
} }
else if (!strncmp("moutput", cmd, len)) else if (!strncmp("moutput", cmd, len))
@ -363,7 +442,7 @@ bool NormApp::ProcessCommand(const char* cmd, const char* val)
} }
else if (!(output = fopen(val, "wb"))) else if (!(output = fopen(val, "wb")))
{ {
DMSG(0, "NormApp::ProcessCommand(output) fopen() error: %s\n", DMSG(0, "NormApp::OnCommand(output) fopen() error: %s\n",
strerror(errno)); strerror(errno));
return false; return false;
} }
@ -375,7 +454,7 @@ bool NormApp::ProcessCommand(const char* cmd, const char* val)
{ {
if (!tx_file_list.Append(val)) if (!tx_file_list.Append(val))
{ {
DMSG(0, "NormApp::ProcessCommand(sendfile) Error appending \"%s\" " DMSG(0, "NormApp::OnCommand(sendfile) Error appending \"%s\" "
"to tx file list.\n", val); "to tx file list.\n", val);
return false; return false;
} }
@ -386,7 +465,7 @@ bool NormApp::ProcessCommand(const char* cmd, const char* val)
tx_object_interval = -1.0; tx_object_interval = -1.0;
if (tx_object_interval < 0.0) if (tx_object_interval < 0.0)
{ {
DMSG(0, "NormApp::ProcessCommand(interval) Invalid tx object interval: %s\n", DMSG(0, "NormApp::OnCommand(interval) Invalid tx object interval: %s\n",
val); val);
tx_object_interval = 0.0; tx_object_interval = 0.0;
return false; return false;
@ -402,12 +481,16 @@ bool NormApp::ProcessCommand(const char* cmd, const char* val)
tx_repeat_interval = -1.0; tx_repeat_interval = -1.0;
if (tx_repeat_interval < 0.0) if (tx_repeat_interval < 0.0)
{ {
DMSG(0, "NormApp::ProcessCommand(rinterval) Invalid tx repeat interval: %s\n", DMSG(0, "NormApp::OnCommand(rinterval) Invalid tx repeat interval: %s\n",
val); val);
tx_repeat_interval = 0.0; tx_repeat_interval = 0.0;
return false; return false;
} }
} }
else if (!strncmp("updatesOnly", cmd, len))
{
tx_file_list.InitUpdateTime(true);
}
else if (!strncmp("rxcachedir", cmd, len)) else if (!strncmp("rxcachedir", cmd, len))
{ {
unsigned int length = strlen(val); unsigned int length = strlen(val);
@ -418,7 +501,7 @@ bool NormApp::ProcessCommand(const char* cmd, const char* val)
length += 1; length += 1;
if (!(rx_cache_path = new char[length])) if (!(rx_cache_path = new char[length]))
{ {
DMSG(0, "NormApp::ProcessCommand(rxcachedir) alloc error: %s\n", DMSG(0, "NormApp::OnCommand(rxcachedir) alloc error: %s\n",
strerror(errno)); strerror(errno));
return false; return false;
} }
@ -431,7 +514,7 @@ bool NormApp::ProcessCommand(const char* cmd, const char* val)
int segmentSize = atoi(val); int segmentSize = atoi(val);
if ((segmentSize < 0) || (segmentSize > 8000)) if ((segmentSize < 0) || (segmentSize > 8000))
{ {
DMSG(0, "NormApp::ProcessCommand(segment) invalid segment size!\n"); DMSG(0, "NormApp::OnCommand(segment) invalid segment size!\n");
return false; return false;
} }
segment_size = segmentSize; segment_size = segmentSize;
@ -441,7 +524,7 @@ bool NormApp::ProcessCommand(const char* cmd, const char* val)
int blockSize = atoi(val); int blockSize = atoi(val);
if ((blockSize < 1) || (blockSize > 255)) if ((blockSize < 1) || (blockSize > 255))
{ {
DMSG(0, "NormApp::ProcessCommand(block) invalid block size!\n"); DMSG(0, "NormApp::OnCommand(block) invalid block size!\n");
return false; return false;
} }
ndata = blockSize; ndata = blockSize;
@ -451,7 +534,7 @@ bool NormApp::ProcessCommand(const char* cmd, const char* val)
int numParity = atoi(val); int numParity = atoi(val);
if ((numParity < 0) || (numParity > 254)) if ((numParity < 0) || (numParity > 254))
{ {
DMSG(0, "NormApp::ProcessCommand(parity) invalid value!\n"); DMSG(0, "NormApp::OnCommand(parity) invalid value!\n");
return false; return false;
} }
nparity = numParity; nparity = numParity;
@ -461,7 +544,7 @@ bool NormApp::ProcessCommand(const char* cmd, const char* val)
int autoParity = atoi(val); int autoParity = atoi(val);
if ((autoParity < 0) || (autoParity > 254)) if ((autoParity < 0) || (autoParity > 254))
{ {
DMSG(0, "NormApp::ProcessCommand(auto) invalid value!\n"); DMSG(0, "NormApp::OnCommand(auto) invalid value!\n");
return false; return false;
} }
auto_parity = autoParity; auto_parity = autoParity;
@ -472,7 +555,7 @@ bool NormApp::ProcessCommand(const char* cmd, const char* val)
int extraParity = atoi(val); int extraParity = atoi(val);
if ((extraParity < 0) || (extraParity > 254)) if ((extraParity < 0) || (extraParity > 254))
{ {
DMSG(0, "NormApp::ProcessCommand(extra) invalid value!\n"); DMSG(0, "NormApp::OnCommand(extra) invalid value!\n");
return false; return false;
} }
extra_parity = extraParity; extra_parity = extraParity;
@ -483,7 +566,7 @@ bool NormApp::ProcessCommand(const char* cmd, const char* val)
double backoffFactor = atof(val); double backoffFactor = atof(val);
if (backoffFactor < 0) if (backoffFactor < 0)
{ {
DMSG(0, "NormSimAgent::ProcessCommand(backoff) invalid txRate!\n"); DMSG(0, "NormSimAgent::OnCommand(backoff) invalid txRate!\n");
return false; return false;
} }
backoff_factor = backoffFactor; backoff_factor = backoffFactor;
@ -493,7 +576,7 @@ bool NormApp::ProcessCommand(const char* cmd, const char* val)
{ {
if (1 != sscanf(val, "%lu", &tx_buffer_size)) if (1 != sscanf(val, "%lu", &tx_buffer_size))
{ {
DMSG(0, "NormApp::ProcessCommand(txbuffer) invalid value!\n"); DMSG(0, "NormApp::OnCommand(txbuffer) invalid value!\n");
return false; return false;
} }
} }
@ -501,7 +584,7 @@ bool NormApp::ProcessCommand(const char* cmd, const char* val)
{ {
if (1 != sscanf(val, "%lu", &rx_buffer_size)) if (1 != sscanf(val, "%lu", &rx_buffer_size))
{ {
DMSG(0, "NormApp::ProcessCommand(rxbuffer) invalid value!\n"); DMSG(0, "NormApp::OnCommand(rxbuffer) invalid value!\n");
return false; return false;
} }
} }
@ -530,7 +613,7 @@ bool NormApp::ProcessCommand(const char* cmd, const char* val)
msg_flush_mode = NormStreamObject::FLUSH_ACTIVE; msg_flush_mode = NormStreamObject::FLUSH_ACTIVE;
else else
{ {
DMSG(0, "NormApp::ProcessCommand(flush) invalid msg flush mode!\n"); DMSG(0, "NormApp::OnCommand(flush) invalid msg flush mode!\n");
return false; return false;
} }
} }
@ -538,12 +621,31 @@ bool NormApp::ProcessCommand(const char* cmd, const char* val)
{ {
if (!post_processor->SetCommand(val)) if (!post_processor->SetCommand(val))
{ {
DMSG(0, "NormApp::ProcessCommand(processor) error!\n"); DMSG(0, "NormApp::OnCommand(processor) error!\n");
return false; return false;
} }
} }
else if (!strncmp("instance", cmd, len))
{
// First, try to connect to see if instance is already running
if (control_pipe.Connect(val))
{
TRACE("connected to \"%s\"\n", val);
control_remote = true;
}
else if (control_pipe.Listen(val))
{
TRACE("listening as \"%s\"\n", val);
control_remote = false;
}
else
{
DMSG(0, "NormApp::OnCommand(instance) error opening control pipe\n");
return false;
}
}
return true; return true;
} // end NormApp::ProcessCommand() } // end NormApp::OnCommand()
NormApp::CmdType NormApp::CommandType(const char* cmd) NormApp::CmdType NormApp::CommandType(const char* cmd)
@ -590,18 +692,18 @@ bool NormApp::ProcessCommands(int argc, const char*const* argv)
argv[i]); argv[i]);
return false; return false;
case CMD_NOARG: case CMD_NOARG:
if (!ProcessCommand(argv[i], NULL)) if (!OnCommand(argv[i], NULL))
{ {
DMSG(0, "NormApp::ProcessCommands() ProcessCommand(%s) error\n", DMSG(0, "NormApp::ProcessCommands() OnCommand(%s) error\n",
argv[i]); argv[i]);
return false; return false;
} }
i++; i++;
break; break;
case CMD_ARG: case CMD_ARG:
if (!ProcessCommand(argv[i], argv[i+1])) if (!OnCommand(argv[i], argv[i+1]))
{ {
DMSG(0, "NormApp::ProcessCommands() ProcessCommand(%s, %s) error\n", DMSG(0, "NormApp::ProcessCommands() OnCommand(%s, %s) error\n",
argv[i], argv[i+1]); argv[i], argv[i+1]);
return false; return false;
} }
@ -1085,6 +1187,7 @@ bool NormApp::OnIntervalTimeout(ProtoTimer& /*theTimer*/)
char fileName[PATH_MAX]; char fileName[PATH_MAX];
if (tx_file_list.GetNextFile(fileName)) if (tx_file_list.GetNextFile(fileName))
{ {
tx_repeat_clear = true;
char pathName[PATH_MAX]; char pathName[PATH_MAX];
tx_file_list.GetCurrentBasePath(pathName); tx_file_list.GetCurrentBasePath(pathName);
unsigned int len = strlen(pathName); unsigned int len = strlen(pathName);
@ -1132,11 +1235,24 @@ bool NormApp::OnIntervalTimeout(ProtoTimer& /*theTimer*/)
} }
else else
{ {
return OnIntervalTimeout(interval_timer); if (tx_repeat_clear)
{
tx_repeat_clear = false;
return OnIntervalTimeout(interval_timer);
}
else
{
tx_repeat_clear = true;
if (interval_timer.IsActive()) interval_timer.Deactivate();
interval_timer.SetInterval(tx_repeat_interval);
ActivateTimer(interval_timer);
return false;
}
} }
} }
else else
{ {
tx_repeat_clear = true;
DMSG(0, "norm: End of tx file list reached.\n"); DMSG(0, "norm: End of tx file list reached.\n");
} }
return true; return true;
@ -1160,6 +1276,8 @@ bool NormApp::OnStartup(int argc, const char*const* argv)
#ifdef UNIX #ifdef UNIX
signal(SIGCHLD, SignalHandler); signal(SIGCHLD, SignalHandler);
#endif // UNIX #endif // UNIX
if (control_remote) return false;
// Validate our application settings // Validate our application settings
if (!address) if (!address)
@ -1183,7 +1301,7 @@ bool NormApp::OnStartup(int argc, const char*const* argv)
session->SetTrace(tracing); session->SetTrace(tracing);
session->SetTxLoss(tx_loss); session->SetTxLoss(tx_loss);
session->SetRxLoss(rx_loss); session->SetRxLoss(rx_loss);
session->SetBackoffFactor(backoff_factor); session->SetBackoffFactor(backoff_factor);
if (input || !tx_file_list.IsEmpty()) if (input || !tx_file_list.IsEmpty())
{ {

6
common/normFile.h
View File

@ -137,6 +137,12 @@ class NormFileList
{ {
updates_only = updatesOnly; updates_only = updatesOnly;
last_time = this_time = big_time = initTime; last_time = this_time = big_time = initTime;
if (0 == initTime)
{
struct timeval currentTime;
ProtoSystemTime(currentTime);
this_time = currentTime.tv_sec;
}
} }
bool Append(const char* path); bool Append(const char* path);

4
common/normSession.cpp
View File

@ -1710,7 +1710,9 @@ void NormSession::ServerHandleNackMessage(const struct timeval& currentTime, Nor
} }
else else
{ {
block->HandleSegmentRequest(nextSegmentId, lastSegmentId, ndata, nparity, numErasures); block->HandleSegmentRequest(nextSegmentId, lastSegmentId,
object->GetBlockSize(block->Id()),
nparity, numErasures);
startTimer = true; startTimer = true;
} // end if/else (holdoff) } // end if/else (holdoff)
inRange = false; inRange = false;

35
common/normSimAgent.cpp
View File

@ -19,6 +19,7 @@ NormSimAgent::NormSimAgent(ProtoTimerMgr& timerMgr,
tx_object_size(0), tx_object_interval(0.0), tx_object_size(0), tx_object_interval(0.0),
tx_repeat_count(0), tx_repeat_interval(0.0), tx_repeat_count(0), tx_repeat_interval(0.0),
stream(NULL), auto_stream(false), push_stream(false), stream(NULL), auto_stream(false), push_stream(false),
flush_mode(NormStreamObject::FLUSH_PASSIVE),
mgen(NULL), msg_sync(false), mgen_bytes(0), mgen_pending_bytes(0), mgen(NULL), msg_sync(false), mgen_bytes(0), mgen_pending_bytes(0),
tracing(false), tx_loss(0.0), rx_loss(0.0) tracing(false), tx_loss(0.0), rx_loss(0.0)
{ {
@ -69,7 +70,8 @@ const char* const NormSimAgent::cmd_list[] =
"+sendStream", // send a simulated NORM stream "+sendStream", // send a simulated NORM stream
"+openStream", // open a stream object for messaging "+openStream", // open a stream object for messaging
"+push", // "on" means real-time push stream advancement (non-blocking) "+push", // "on" means real-time push stream advancement (non-blocking)
"-flushStream", // flush output stream "+flush", // stream flush mode
"-doFlush", // invoke flushing of stream
"+unicastNacks", // clients will unicast feedback "+unicastNacks", // clients will unicast feedback
"+silentClient", // clients will not transmit "+silentClient", // clients will not transmit
NULL NULL
@ -397,7 +399,22 @@ bool NormSimAgent::ProcessCommand(const char* cmd, const char* val)
return false; return false;
} }
} }
else if (!strncmp("flushStream", cmd, len)) else if (!strncmp("flush", cmd, len))
{
int valLen = strlen(val);
if (!strncmp("none", val, valLen))
flush_mode = NormStreamObject::FLUSH_NONE;
else if (!strncmp("passive", val, valLen))
flush_mode = NormStreamObject::FLUSH_PASSIVE;
else if (!strncmp("active", val, valLen))
flush_mode = NormStreamObject::FLUSH_ACTIVE;
else
{
DMSG(0, "NormApp::OnCommand(flush) invalid msg flush mode!\n");
return false;
}
}
else if (!strncmp("doFlush", cmd, len))
{ {
if (session) if (session)
{ {
@ -535,12 +552,13 @@ void NormSimAgent::OnInputReady()
{ {
unsigned int bytesWrote = stream->Write(tx_msg_buffer+tx_msg_index, unsigned int bytesWrote = stream->Write(tx_msg_buffer+tx_msg_index,
tx_msg_len - tx_msg_index, tx_msg_len - tx_msg_index,
false, false, push_stream); NormStreamObject::FLUSH_NONE,
false, push_stream);
tx_msg_index += bytesWrote; tx_msg_index += bytesWrote;
if (tx_msg_index == tx_msg_len) if (tx_msg_index == tx_msg_len)
{ {
// Provide EOM indication to norm stream // Provide EOM indication to norm stream
stream->Write(NULL, 0, false, true, false); stream->Write(NULL, 0, flush_mode, true, false);
tx_msg_index = tx_msg_len = 0; tx_msg_index = tx_msg_len = 0;
} }
} }
@ -550,7 +568,7 @@ bool NormSimAgent::FlushStream()
{ {
if (stream && session && session->IsServer()) if (stream && session && session->IsServer())
{ {
stream->Write(NULL, 0, true, false, false); stream->Write(NULL, 0, flush_mode, false, false);
return true; return true;
} }
else else
@ -576,7 +594,10 @@ void NormSimAgent::Notify(NormController::Event event,
{ {
// sending a dummy byte stream // sending a dummy byte stream
char buffer[NormMsg::MAX_SIZE]; char buffer[NormMsg::MAX_SIZE];
unsigned int count = stream->Write(buffer, segment_size, false, false, false); unsigned int count =
stream->Write(buffer, segment_size,
NormStreamObject::FLUSH_NONE,
false, false);
} }
else else
{ {
@ -659,7 +680,7 @@ void NormSimAgent::Notify(NormController::Event event,
{ {
// Read the stream when it's updated // Read the stream when it's updated
if (mgen) if (mgen)
{ {
bool dataReady = true; bool dataReady = true;
while (dataReady) while (dataReady)
{ {

81
common/normSimAgent.h
View File

@ -46,51 +46,52 @@ class NormSimAgent : public NormController
static const char* const cmd_list[]; static const char* const cmd_list[];
NormSessionMgr session_mgr; NormSessionMgr session_mgr;
NormSession* session; NormSession* session;
// session parameters // session parameters
char* address; // session address char* address; // session address
UINT16 port; // session port number UINT16 port; // session port number
UINT8 ttl; UINT8 ttl;
double tx_rate; // bits/sec double tx_rate; // bits/sec
bool cc_enable; bool cc_enable;
bool unicast_nacks; bool unicast_nacks;
bool silent_client; bool silent_client;
double backoff_factor; double backoff_factor;
UINT16 segment_size; UINT16 segment_size;
UINT8 ndata; UINT8 ndata;
UINT8 nparity; UINT8 nparity;
UINT8 auto_parity; UINT8 auto_parity;
UINT8 extra_parity; UINT8 extra_parity;
double group_size; double group_size;
unsigned long tx_buffer_size; // bytes unsigned long tx_buffer_size; // bytes
unsigned long rx_buffer_size; // bytes unsigned long rx_buffer_size; // bytes
// for simulated transmission (streams or files) // for simulated transmission (streams or files)
unsigned long tx_object_size; unsigned long tx_object_size;
double tx_object_interval; double tx_object_interval;
int tx_repeat_count; int tx_repeat_count;
double tx_repeat_interval; double tx_repeat_interval;
NormStreamObject* stream; NormStreamObject* stream;
bool auto_stream; bool auto_stream;
bool push_stream; bool push_stream;
char* tx_msg_buffer; NormStreamObject::FlushType flush_mode;
unsigned int tx_msg_len; char* tx_msg_buffer;
unsigned int tx_msg_index; unsigned int tx_msg_len;
Mgen* mgen; unsigned int tx_msg_index;
char mgen_buffer[64]; Mgen* mgen;
bool msg_sync; char mgen_buffer[64];
unsigned int mgen_bytes; bool msg_sync;
unsigned int mgen_pending_bytes; unsigned int mgen_bytes;
unsigned int mgen_pending_bytes;
ProtoTimer interval_timer; ProtoTimer interval_timer;
// protocol debug parameters // protocol debug parameters
bool tracing; bool tracing;
double tx_loss; double tx_loss;
double rx_loss; double rx_loss;
}; // end class NormSimAgent }; // end class NormSimAgent

2
common/normVersion.h
View File

@ -32,6 +32,6 @@
#ifndef _NORM_VERSION #ifndef _NORM_VERSION
#define _NORM_VERSION #define _NORM_VERSION
#define VERSION "1.1b4" #define VERSION "1.1b5"
#endif // _NORM_VERSION #endif // _NORM_VERSION