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

v1.2b2

pull/2/head
Jeff Weston 2019-09-11 11:40:31 -04:00
parent 718455beb5
commit 7fdb889444
29 changed files with 4151 additions and 924 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

1131
common/normApi.cpp Normal file
View File

File diff suppressed because it is too large Load Diff

258
common/normApi.h Normal file
View File

@ -0,0 +1,258 @@
#ifndef _NORM_API
#define _NORM_API
#ifdef WIN32
#include <winsock2.h>
#include <windows.h>
#endif // WIN32
////////////////////////////////////////////////////////////
// IMPORTANT NOTICE
// The NORM API is _very_ much in a developmental phase
// right now. So, although this code is available in
// source code distribution, it is very much subject
// to change in future revisions. The goal of the NORM
// API _will_ be to provide a stable base of function calls
// for applications to use, even as the underlying NORM
// C++ code continues to evolve. But, until this notice
// is removed, the API shouldn't be considered final.
typedef const void* NormInstanceHandle;
extern const NormInstanceHandle NORM_INSTANCE_INVALID;
NormInstanceHandle NormCreateInstance();
void NormDestroyInstance(NormInstanceHandle instanceHandle);
// NORM session creation and control
typedef const void* NormSessionHandle;
extern const NormSessionHandle NORM_SESSION_INVALID;
typedef unsigned long NormNodeId;
extern const NormNodeId NORM_NODE_NONE;
extern const NormNodeId NORM_NODE_ANY;
NormSessionHandle NormCreateSession(NormInstanceHandle instanceHandle,
const char* sessionAddress,
unsigned short sessionPort,
NormNodeId localNodeId);
void NormDestroySession(NormSessionHandle sessionHandle);
// Session management and parameters
void NormSetTransmitRate(NormSessionHandle sessionHandle,
double bitsPerSecond);
void NormSetGrttEstimate(NormSessionHandle sessionHandle,
double grttEstimate);
NormNodeId NormGetLocalNodeId(NormSessionHandle sessionHandle);
bool NormSetLoopback(NormSessionHandle sessionHandle, bool state);
// Debug parameters
void NormSetMessageTrace(NormSessionHandle sessionHandle, bool state);
void NormSetTxLoss(NormSessionHandle sessionHandle, double percent);
void NormSetRxLoss(NormSessionHandle sessionHandle, double percent);
// Sender control & parameters
bool NormStartSender(NormSessionHandle sessionHandle,
unsigned long bufferSpace,
unsigned short segmentSize,
unsigned char numData,
unsigned char numParity);
void NormStopSender(NormSessionHandle sessionHandle);
bool NormAddAckingNode(NormSessionHandle sessionHandle,
NormNodeId nodeId);
void NormRemoveAckingNode(NormSessionHandle sessionHandle,
NormNodeId nodeId);
// Receiver control & parameters
bool NormStartReceiver(NormSessionHandle sessionHandle,
unsigned long bufferSpace);
void NormStopReceiver(NormSessionHandle sessionHandle);
void NormSetDefaultUnicastNack(NormSessionHandle sessionHandle,
bool state);
typedef const void* NormNodeHandle;
extern const NormNodeHandle NORM_NODE_INVALID;
typedef const void* NormObjectHandle;
extern const NormObjectHandle NORM_OBJECT_INVALID;
typedef unsigned short NormObjectTransportId;
enum NormNackingMode
{
NORM_NACK_NONE,
NORM_NACK_INFO_ONLY,
NORM_NACK_NORMAL
};
enum NormRepairBoundary
{
NORM_BOUNDARY_BLOCK,
NORM_BOUNDARY_OBJECT
};
NormNodeId NormGetNodeId(NormNodeHandle nodeHandle);
NormRepairBoundary NormGetNodeRepairBoundary(NormNodeHandle nodeHandle);
void NormSetNodeRepairBoundary(NormNodeHandle nodeHandle,
NormRepairBoundary repairBoundary);
void NormSetDefaultRepairBoundary(NormSessionHandle sessionHandle,
NormRepairBoundary repairBoundary);
// General NormObject functions
enum NormObjectType
{
NORM_OBJECT_NONE,
NORM_OBJECT_DATA,
NORM_OBJECT_FILE,
NORM_OBJECT_STREAM
};
NormObjectType NormGetObjectType(NormObjectHandle objectHandle);
bool NormGetObjectInfo(NormObjectHandle objectHandle,
char* infoBuffer,
unsigned short* infoLen);
NormObjectTransportId NormGetObjectTransportId(NormObjectHandle objectHandle);
void NormCancelObject(NormObjectHandle objectHandle);
void NormRetainObject(NormObjectHandle objectHandle);
void NormReleaseObject(NormObjectHandle objectHandle);
// Receiver-only NormObject functions
NormNackingMode NormGetObjectNackingMode(NormObjectHandle objectHandle);
void NormSetObjectNackingMode(NormObjectHandle objectHandle,
NormNackingMode nackingMode);
void NormSetNodeNackingMode(NormNodeHandle nodeHandle,
NormNackingMode nackingMode);
void NormSetDefaultNackingMode(NormSessionHandle sessionHandle,
NormNackingMode nackingMode);
// Sender-only NormObject functions
bool NormSetWatermark(NormSessionHandle sessionHandle,
NormObjectHandle objectHandle);
// NormStreamObject functions
NormObjectHandle NormOpenStream(NormSessionHandle sessionHandle,
unsigned long bufferSize);
void NormCloseStream(NormObjectHandle streamHandle);
enum NormFlushMode
{
NORM_FLUSH_NONE,
NORM_FLUSH_PASSIVE,
NORM_FLUSH_ACTIVE
};
void NormSetStreamFlushMode(NormObjectHandle streamHandle,
NormFlushMode flushMode);
void NormSetStreamPushMode(NormObjectHandle streamHandle,
bool state);
unsigned int NormWriteStream(NormObjectHandle streamHandle,
const char* buffer,
unsigned int numBytes);
void NormFlushStream(NormObjectHandle streamHandle);
void NormMarkStreamEom(NormObjectHandle streamHandle);
bool NormReadStream(NormObjectHandle streamHandle,
char* buffer,
unsigned int* numBytes);
// NormFileObject Functions
NormObjectHandle NormQueueFile(NormSessionHandle sessionHandle,
const char* fileName,
const char* infoPtr = (const char*)0,
unsigned int infoLen = 0);
bool NormSetCacheDirectory(NormInstanceHandle instanceHandle,
const char* cachePath);
bool NormGetFileName(NormObjectHandle fileHandle,
char* nameBuffer,
unsigned int bufferLen);
bool NormSetFileName(NormObjectHandle fileHandle,
const char* fileName);
// NormDataObject Functions
bool NormQueueData(const char* dataPtr,
unsigned long dataLen,
const char* infoPtr = (const char*)0,
unsigned int infoLen = 0);
// NORM Event Notification Routines
enum NormEventType
{
NORM_EVENT_INVALID = 0,
NORM_TX_QUEUE_VACANCY,
NORM_TX_QUEUE_EMPTY,
NORM_TX_OBJECT_SENT,
NORM_TX_OBJECT_PURGED,
NORM_LOCAL_SERVER_CLOSED,
NORM_REMOTE_SERVER_NEW,
NORM_REMOTE_SERVER_INACTIVE,
NORM_REMOTE_SERVER_ACTIVE,
NORM_RX_OBJECT_NEW,
NORM_RX_OBJECT_INFO,
NORM_RX_OBJECT_UPDATE,
NORM_RX_OBJECT_COMPLETED,
NORM_RX_OBJECT_ABORTED
};
typedef struct
{
NormEventType type;
NormSessionHandle session;
NormNodeHandle sender;
NormObjectHandle object;
} NormEvent;
// This call blocks until the next NormEvent is ready unless asynchronous
// notification is used (see below)
bool NormGetNextEvent(NormInstanceHandle instanceHandle, NormEvent* theEvent);
// The "NormGetDescriptor()" function returns a HANDLE (WIN32) or
// a file descriptor (UNIX) which can be used for async notification
// of pending NORM events. On WIN32, the returned HANDLE can be used
// with system calls like "WaitForSingleEvent()", and on Unix the
// returned descriptor can be used in a "select()" call. If this type
// of asynchronous notification is used, calls to "NormGetNextEvent()" will
// not block.
#ifdef WIN32
HANDLE NormGetDescriptor(NormInstanceHandle instanceHandle);
#else
int NormGetDescriptor(NormInstanceHandle instanceHandle);
#endif // if/else WIN32/UNIX
#endif // _NORM_API

169
common/normApp.cpp
View File

@ -11,6 +11,11 @@
#include <stdlib.h> #include <stdlib.h>
#include <ctype.h> // for "isspace()" #include <ctype.h> // for "isspace()"
#ifdef UNIX
#include <sys/types.h>
#include <sys/wait.h> // for "waitpid()"
#endif // UNIX
// Command-line application using Protolib EventDispatcher // Command-line application using Protolib EventDispatcher
class NormApp : public NormController, public ProtoApp class NormApp : public NormController, public ProtoApp
{ {
@ -64,8 +69,8 @@ class NormApp : public NormController, public ProtoApp
unsigned int input_index; unsigned int input_index;
unsigned int input_length; unsigned int input_length;
bool input_active; bool input_active;
bool push_stream; bool push_mode;
NormStreamObject::FlushType msg_flush_mode; NormStreamObject::FlushMode msg_flush_mode;
bool input_messaging; // stream input mode bool input_messaging; // stream input mode
UINT16 input_msg_length; UINT16 input_msg_length;
UINT16 input_msg_index; UINT16 input_msg_index;
@ -79,6 +84,7 @@ class NormApp : public NormController, public ProtoApp
char* address; // session address char* address; // session address
UINT16 port; // session port number UINT16 port; // session port number
UINT8 ttl; UINT8 ttl;
bool loopback;
char* interface_name; // for multi-home hosts char* interface_name; // for multi-home hosts
double tx_rate; // bits/sec double tx_rate; // bits/sec
bool cc_enable; bool cc_enable;
@ -90,6 +96,8 @@ class NormApp : public NormController, public ProtoApp
UINT8 auto_parity; UINT8 auto_parity;
UINT8 extra_parity; UINT8 extra_parity;
double backoff_factor; double backoff_factor;
double grtt_estimate; // initial grtt estimate
double group_size;
unsigned long tx_buffer_size; // bytes unsigned long tx_buffer_size; // bytes
NormFileList tx_file_list; NormFileList tx_file_list;
double tx_object_interval; double tx_object_interval;
@ -118,13 +126,15 @@ NormApp::NormApp()
session_mgr(GetTimerMgr(), GetSocketNotifier()), 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_mode(false), msg_flush_mode(NormStreamObject::FLUSH_PASSIVE),
input_messaging(false), input_msg_length(0), input_msg_index(0), input_messaging(false), input_msg_length(0), input_msg_index(0),
output_index(0), output_messaging(false), output_msg_length(0), output_msg_sync(false), output_index(0), output_messaging(false), output_msg_length(0), output_msg_sync(false),
address(NULL), port(0), ttl(3), interface_name(NULL), address(NULL), port(0), ttl(3), loopback(false), interface_name(NULL),
tx_rate(64000.0), cc_enable(false), tx_rate(64000.0), cc_enable(false),
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),
grtt_estimate(NormSession::DEFAULT_GRTT_ESTIMATE),
group_size(NormSession::DEFAULT_GSIZE_ESTIMATE),
tx_buffer_size(1024*1024), tx_buffer_size(1024*1024),
tx_object_interval(0.0), tx_repeat_count(0), tx_repeat_interval(2.0), tx_repeat_clear(true), 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),
@ -153,16 +163,33 @@ NormApp::~NormApp()
if (post_processor) delete post_processor; if (post_processor) delete post_processor;
} }
// NOTE on message flushing mode:
//
// "none" - messages are aggregated to make NORM_DATA
// payloads a full "segmentSize"
//
// "passive" - At the end-of-message, short NORM_DATA
// payloads are sent as needed so that
// the message is immediately sent
// (i.e. not held for aggregation)
// "active" - Same as "passive", plus NORM_CMD(FLUSH)
// messages are generated until new
// message is written to stream
// (This helps keep receivers in sync
// with sender when intermittent message
// traffic is sent)
const char* const NormApp::cmd_list[] = const char* const NormApp::cmd_list[] =
{ {
"+debug", // debug level "+debug", // debug level
"+log", // log file name "+log", // log file name
"-trace", // message tracing on "+trace", // message tracing on
"+txloss", // tx packet loss percent (for testing) "+txloss", // tx packet loss percent (for testing)
"+rxloss", // rx packet loss percent (for testing) "+rxloss", // rx packet loss percent (for testing)
"+address", // session destination address "+address", // session destination address
"+ttl", // multicast hop count scope "+ttl", // multicast hop count scope
"+loopback", // "off" or "on" to recv our own packets
"+interface", // multicast interface name to use "+interface", // multicast interface name to use
"+cc", // congestion control on/off "+cc", // congestion control on/off
"+rate", // tx date rate (bps) "+rate", // tx date rate (bps)
@ -184,18 +211,19 @@ const char* const NormApp::cmd_list[] =
"+auto", // Number of FEC packets to proactively send (<= nparity) "+auto", // Number of FEC packets to proactively send (<= nparity)
"+extra", // Number of extra FEC packets sent in response to repair requests "+extra", // Number of extra FEC packets sent in response to repair requests
"+backoff", // Backoff factor to use "+backoff", // Backoff factor to use
"+grtt", // Set sender's initial GRTT estimate
"+gsize", // Set sender's group size estimate
"+txbuffer", // Size of sender's buffer "+txbuffer", // Size of sender's buffer
"+rxbuffer", // Size receiver allocates for buffering each sender "+rxbuffer", // Size receiver allocates for buffering each sender
"-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 "+instance", // specify norm instance name for remote control commands
NULL NULL
}; };
void NormApp::OnControlEvent(ProtoSocket& /*theSocket*/, ProtoSocket::Event theEvent) void NormApp::OnControlEvent(ProtoSocket& /*theSocket*/, ProtoSocket::Event theEvent)
{ {
TRACE("NormApp::OnControlEvent() ...\n");
switch (theEvent) switch (theEvent)
{ {
case ProtoSocket::RECV: case ProtoSocket::RECV:
@ -205,7 +233,7 @@ void NormApp::OnControlEvent(ProtoSocket& /*theSocket*/, ProtoSocket::Event theE
if (control_pipe.Recv(buffer, len)) if (control_pipe.Recv(buffer, len))
{ {
buffer[len] = '\0'; buffer[len] = '\0';
TRACE("norm: received command \"%s\"\n", buffer); DMSG(0, "norm: received command \"%s\"\n", buffer);
char* cmd = buffer; char* cmd = buffer;
char* val = NULL; char* val = NULL;
for (unsigned int i = 0; i < len; i++) for (unsigned int i = 0; i < len; i++)
@ -214,6 +242,7 @@ void NormApp::OnControlEvent(ProtoSocket& /*theSocket*/, ProtoSocket::Event theE
{ {
buffer[i++] = '\0'; buffer[i++] = '\0';
val = buffer + i; val = buffer + i;
break;
} }
} }
if (!OnCommand(cmd, val)) if (!OnCommand(cmd, val))
@ -288,8 +317,16 @@ bool NormApp::OnCommand(const char* cmd, const char* val)
} }
else if (!strncmp("trace", cmd, len)) else if (!strncmp("trace", cmd, len))
{ {
tracing = true; if (!strcmp("on", val))
if (session) session->SetTrace(true); tracing = true;
else if (!strcmp("off", val))
tracing = false;
else
{
DMSG(0, "NormApp::OnCommand(trace) invalid argument!\n");
return false;
}
if (session) session->SetTrace(tracing);
} }
else if (!strncmp("txloss", cmd, len)) else if (!strncmp("txloss", cmd, len))
{ {
@ -346,12 +383,38 @@ bool NormApp::OnCommand(const char* cmd, const char* val)
else if (!strncmp("ttl", cmd, len)) else if (!strncmp("ttl", cmd, len))
{ {
int ttlTemp = atoi(val); int ttlTemp = atoi(val);
if ((ttlTemp < 1) || (ttlTemp > 255)) if ((ttlTemp < 0) || (ttlTemp > 255))
{ {
DMSG(0, "NormApp::OnCommand(ttl) invalid value!\n"); DMSG(0, "NormApp::OnCommand(ttl) invalid value!\n");
return false; return false;
} }
ttl = ttlTemp; bool result = session ? session->SetTTL((UINT8)ttlTemp) : true;
ttl = result ? ttlTemp : ttl;
if (!result)
{
DMSG(0, "NormApp::OnCommand(ttl) error setting socket ttl!\n");
return false;
}
}
else if (!strncmp("loopback", cmd, len))
{
bool loopTemp = loopback;
if (!strcmp("on", val))
loopTemp = true;
else if (!strcmp("off", val))
loopTemp = false;
else
{
DMSG(0, "NormApp::OnCommand(loopback) invalid argument!\n");
return false;
}
bool result = session ? session->SetLoopback(loopTemp) : true;
loopback = result ? loopTemp : loopback;
if (!result)
{
DMSG(0, "NormApp::OnCommand(loopback) error setting socket loopback!\n");
return false;
}
} }
else if (!strncmp("interface", cmd, len)) else if (!strncmp("interface", cmd, len))
{ {
@ -580,12 +643,34 @@ bool NormApp::OnCommand(const char* cmd, const char* val)
double backoffFactor = atof(val); double backoffFactor = atof(val);
if (backoffFactor < 0) if (backoffFactor < 0)
{ {
DMSG(0, "NormSimAgent::OnCommand(backoff) invalid txRate!\n"); DMSG(0, "NormApp::OnCommand(backoff) invalid value!\n");
return false; return false;
} }
backoff_factor = backoffFactor; backoff_factor = backoffFactor;
if (session) session->SetBackoffFactor(backoffFactor); if (session) session->SetBackoffFactor(backoffFactor);
} }
else if (!strncmp("grtt", cmd, len))
{
double grttEstimate = atof(val);
if (grttEstimate < 0)
{
DMSG(0, "NormApp::OnCommand(grtt) invalid value!\n");
return false;
}
grtt_estimate = grttEstimate;
if (session) session->ServerSetGrtt(grttEstimate);
}
else if (!strncmp("gsize", cmd, len))
{
double groupSize = atof(val);
if (groupSize < 0)
{
DMSG(0, "NormApp::OnCommand(gsize) invalid value!\n");
return false;
}
group_size = groupSize;
if (session) session->ServerSetGroupSize(groupSize);
}
else if (!strncmp("txbuffer", cmd, len)) else if (!strncmp("txbuffer", cmd, len))
{ {
if (1 != sscanf(val, "%lu", &tx_buffer_size)) if (1 != sscanf(val, "%lu", &tx_buffer_size))
@ -614,7 +699,8 @@ bool NormApp::OnCommand(const char* cmd, const char* val)
} }
else if (!strncmp("push", cmd, len)) else if (!strncmp("push", cmd, len))
{ {
push_stream = true; push_mode = true;
if (tx_stream) tx_stream->SetPushMode(push_mode);
} }
else if (!strncmp("flush", cmd, len)) else if (!strncmp("flush", cmd, len))
{ {
@ -630,6 +716,7 @@ bool NormApp::OnCommand(const char* cmd, const char* val)
DMSG(0, "NormApp::OnCommand(flush) invalid msg flush mode!\n"); DMSG(0, "NormApp::OnCommand(flush) invalid msg flush mode!\n");
return false; return false;
} }
if (tx_stream) tx_stream->SetFlushMode(msg_flush_mode);
} }
else if (!strncmp("processor", cmd, len)) else if (!strncmp("processor", cmd, len))
{ {
@ -644,12 +731,10 @@ bool NormApp::OnCommand(const char* cmd, const char* val)
// First, try to connect to see if instance is already running // First, try to connect to see if instance is already running
if (control_pipe.Connect(val)) if (control_pipe.Connect(val))
{ {
TRACE("connected to \"%s\"\n", val);
control_remote = true; control_remote = true;
} }
else if (control_pipe.Listen(val)) else if (control_pipe.Listen(val))
{ {
TRACE("listening as \"%s\"\n", val);
control_remote = false; control_remote = false;
} }
else else
@ -740,7 +825,7 @@ void NormApp::OnInputReady()
{ {
//DMSG(0, "NormApp::OnInputReady() ...\n"); //DMSG(0, "NormApp::OnInputReady() ...\n");
NormStreamObject::FlushType flushType = NormStreamObject::FLUSH_NONE; bool endOfStream = false;
// write to the stream while input is available _and_ // write to the stream while input is available _and_
// the stream has buffer space for input // the stream has buffer space for input
while (input) while (input)
@ -816,7 +901,8 @@ void NormApp::OnInputReady()
} }
if (stdin != input) fclose(input); if (stdin != input) fclose(input);
input = NULL; input = NULL;
flushType = NormStreamObject::FLUSH_ACTIVE; endOfStream = true;
tx_stream->SetFlushMode(NormStreamObject::FLUSH_ACTIVE);
} }
else if (ferror(input)) else if (ferror(input))
{ {
@ -839,11 +925,10 @@ void NormApp::OnInputReady()
unsigned int writeLength = input_length;// ? input_length - input_index : 0; unsigned int writeLength = input_length;// ? input_length - input_index : 0;
if (writeLength || (NormStreamObject::FLUSH_NONE != flushType)) if (writeLength || endOfStream)
{ {
unsigned int wroteLength = tx_stream->Write(input_buffer+input_index, unsigned int wroteLength = tx_stream->Write(input_buffer+input_index,
writeLength, flushType, false, writeLength, false);
push_stream);
input_length -= wroteLength; input_length -= wroteLength;
if (0 == input_length) if (0 == input_length)
input_index = 0; input_index = 0;
@ -857,7 +942,9 @@ void NormApp::OnInputReady()
input_msg_index = 0; input_msg_index = 0;
input_msg_length = 0; input_msg_length = 0;
// Mark EOM _and_ flush // Mark EOM _and_ flush
tx_stream->Write(NULL, 0, msg_flush_mode, true, false); tx_stream->Write(NULL, 0, true);
// No need to explicitly flush with "flush mode" set.
//tx_stream->Flush();
} }
} }
if (wroteLength < writeLength) if (wroteLength < writeLength)
@ -890,7 +977,7 @@ void NormApp::OnInputReady()
#endif // if/else WIN32/UNIX #endif // if/else WIN32/UNIX
input_active = true; input_active = true;
else else
DMSG(0, "NormApp::Notify(TX_QUEUE_EMPTY) error adding input notification!\n"); DMSG(0, "NormApp::OnInputReady() error adding input notification!\n");
} }
break; break;
} }
@ -992,6 +1079,9 @@ void NormApp::Notify(NormController::Event event,
case NormObject::DATA: case NormObject::DATA:
DMSG(0, "NormApp::Notify() FILE/DATA objects not _yet_ supported...\n"); DMSG(0, "NormApp::Notify() FILE/DATA objects not _yet_ supported...\n");
break; break;
case NormObject::NONE:
break;
} }
break; break;
} }
@ -1032,6 +1122,7 @@ void NormApp::Notify(NormController::Event event,
} }
case NormObject::DATA: case NormObject::DATA:
case NormObject::STREAM: case NormObject::STREAM:
case NormObject::NONE:
break; break;
} // end switch(object->GetType()) } // end switch(object->GetType())
break; break;
@ -1042,6 +1133,8 @@ void NormApp::Notify(NormController::Event event,
{ {
case NormObject::FILE: case NormObject::FILE:
// (TBD) update reception progress display when applicable // (TBD) update reception progress display when applicable
// Call object->SetNotifyOnUpdate(true) here to keep
// the update notifications coming. Otherwise they stop!
break; break;
case NormObject::STREAM: case NormObject::STREAM:
@ -1161,17 +1254,22 @@ void NormApp::Notify(NormController::Event event,
case NormObject::DATA: case NormObject::DATA:
DMSG(0, "NormApp::Notify() DATA objects not _yet_ supported...\n"); DMSG(0, "NormApp::Notify() DATA objects not _yet_ supported...\n");
break; break;
case NormObject::NONE:
break;
} // end switch (object->GetType()) } // end switch (object->GetType())
break; break;
case RX_OBJECT_COMPLETE: case RX_OBJECT_COMPLETED:
{ {
// (TBD) if we're not archiving files we should
// manage our cache, deleting the cache
// on shutdown ...
//DMSG(0, "NormApp::Notify(RX_OBJECT_COMPLETE) ...\n"); //DMSG(0, "NormApp::Notify(RX_OBJECT_COMPLETE) ...\n");
switch(object->GetType()) switch(object->GetType())
{ {
case NormObject::FILE: case NormObject::FILE:
{ {
const char* filePath = ((NormFileObject*)object)->Path(); const char* filePath = ((NormFileObject*)object)->GetPath();
//DMSG(0, "norm: Completed rx file: %s\n", filePath); //DMSG(0, "norm: Completed rx file: %s\n", filePath);
if (post_processor->IsEnabled()) if (post_processor->IsEnabled())
{ {
@ -1189,9 +1287,15 @@ void NormApp::Notify(NormController::Event event,
case NormObject::DATA: case NormObject::DATA:
ASSERT(0); ASSERT(0);
break; break;
case NormObject::NONE:
break;
} }
break; break;
} }
default:
{
DMSG(4, "NormApp::Notify() unhandled event: %d\n", event);
}
} // end switch(event) } // end switch(event)
} // end NormApp::Notify() } // end NormApp::Notify()
@ -1315,14 +1419,17 @@ 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->SetTTL(ttl);
session->SetLoopback(loopback);
if (input || !tx_file_list.IsEmpty()) if (input || !tx_file_list.IsEmpty())
{ {
NormObjectId baseId = (unsigned short)(rand() * (65535.0/ (double)RAND_MAX)); NormObjectId baseId = (unsigned short)(rand() * (65535.0/ (double)RAND_MAX));
session->ServerSetBaseObjectId(baseId); session->ServerSetBaseObjectId(baseId);
session->SetCongestionControl(cc_enable); session->SetCongestionControl(cc_enable);
session->SetBackoffFactor(backoff_factor);
session->ServerSetGrtt(grtt_estimate);
session->ServerSetGroupSize(group_size);
if (!session->StartServer(tx_buffer_size, segment_size, ndata, nparity, interface_name)) if (!session->StartServer(tx_buffer_size, segment_size, ndata, nparity, interface_name))
{ {
DMSG(0, "NormApp::OnStartup() start server error!\n"); DMSG(0, "NormApp::OnStartup() start server error!\n");
@ -1341,6 +1448,8 @@ bool NormApp::OnStartup(int argc, const char*const* argv)
session_mgr.Destroy(); session_mgr.Destroy();
return false; return false;
} }
tx_stream->SetFlushMode(msg_flush_mode);
tx_stream->SetPushMode(push_mode);
} }
} }
@ -1407,6 +1516,10 @@ void NormApp::SignalHandler(int sigNum)
{ {
NormApp* app = static_cast<NormApp*>(ProtoApp::GetApp()); NormApp* app = static_cast<NormApp*>(ProtoApp::GetApp());
if (app->post_processor) app->post_processor->OnDeath(); if (app->post_processor) app->post_processor->OnDeath();
// The use of "waitpid()" here is a work-around for
// an IRIX SIGCHLD issue
int status;
while (waitpid(-1, &status, WNOHANG) > 0);
signal(SIGCHLD, SignalHandler); signal(SIGCHLD, SignalHandler);
break; break;
} }

12
common/normBitmask.cpp
View File

@ -378,7 +378,9 @@ bool NormBitmask::Add(const NormBitmask& b)
if (b.num_bits > num_bits) return false; if (b.num_bits > num_bits) return false;
for(unsigned int i = 0; i < b.mask_len; i++) for(unsigned int i = 0; i < b.mask_len; i++)
mask[i] |= b.mask[i]; mask[i] |= b.mask[i];
if (b.first_set < first_set) first_set = b.first_set; if ((b.first_set < first_set) &&
(b.first_set < b.num_bits))
first_set = b.first_set;
return true; return true;
} // end NormBitmask::Add() } // end NormBitmask::Add()
@ -406,13 +408,14 @@ bool NormBitmask::XCopy(const NormBitmask& b)
for (unsigned int i = begin; i < len; i++) for (unsigned int i = begin; i < len; i++)
mask[i] = b.mask[i] & ~mask[i]; mask[i] = b.mask[i] & ~mask[i];
if (len < mask_len) memset(&mask[len], 0, mask_len - len); if (len < mask_len) memset(&mask[len], 0, mask_len - len);
if (b.first_set < first_set) UINT32 theFirst = (b.first_set < b.num_bits) ? b.first_set : num_bits;
if (theFirst < first_set)
{ {
first_set = b.first_set; first_set = b.first_set;
} }
else else
{ {
first_set = b.first_set; first_set = theFirst;
if (!GetNextSet(first_set)) first_set = num_bits; if (!GetNextSet(first_set)) first_set = num_bits;
} }
return true; return true;
@ -425,6 +428,7 @@ bool NormBitmask::Multiply(const NormBitmask& b)
for(unsigned int i = 0; i < len; i++) for(unsigned int i = 0; i < len; i++)
mask[i] |= b.mask[i]; mask[i] |= b.mask[i];
if (len < mask_len) memset(&mask[len], 0, mask_len - len); if (len < mask_len) memset(&mask[len], 0, mask_len - len);
if (b.first_set > first_set) if (b.first_set > first_set)
{ {
first_set = b.first_set; first_set = b.first_set;
@ -557,7 +561,7 @@ bool NormSlidingMask::CanSet(UINT32 index) const
bool NormSlidingMask::Set(UINT32 index) bool NormSlidingMask::Set(UINT32 index)
{ {
ASSERT(CanSet(index)); //ASSERT(CanSet(index));
if (IsSet()) if (IsSet())
{ {
// Determine position with respect to current start // Determine position with respect to current start

4
common/normEncoder.cpp
View File

@ -218,8 +218,8 @@ void NormEncoder::Encode(const char *data, char **pVec)
*/ */
NormDecoder::NormDecoder() NormDecoder::NormDecoder()
: npar(0), vector_size(0), : npar(0), vector_size(0), Lambda(NULL),
Lambda(NULL), sVec(NULL), oVec(NULL) sVec(NULL), oVec(NULL), scratch(NULL)
{ {
} }

51
common/normFile.cpp
View File

@ -1,4 +1,3 @@
#include "normFile.h" #include "normFile.h"
#include <string.h> // for strerror() #include <string.h> // for strerror()
@ -63,7 +62,7 @@ bool NormFile::Open(const char* thePath, int theFlags)
if (mkdir(tempPath, 0755)) if (mkdir(tempPath, 0755))
#endif // if/else WIN32/UNIX #endif // if/else WIN32/UNIX
{ {
DMSG(0, "NormFile::Rename() mkdir(%s) error: %s\n", DMSG(0, "NormFile::Open() mkdir(%s) error: %s\n",
tempPath, strerror(errno)); tempPath, strerror(errno));
return false; return false;
} }
@ -159,10 +158,21 @@ bool NormFile::Rename(const char* oldName, const char* newName)
if (!strcmp(oldName, newName)) return true; // no change required if (!strcmp(oldName, newName)) return true; // no change required
// Make sure the new file name isn't an existing "busy" file // Make sure the new file name isn't an existing "busy" file
// (This also builds sub-directories as needed) // (This also builds sub-directories as needed)
if (NormFile::IsLocked(newName)) return false; if (NormFile::IsLocked(newName))
{
DMSG(0, "NormFile::Rename() error: file is locked\n");
return false;
}
#ifdef WIN32 #ifdef WIN32
// In Win32, the new file can't already exist // In Win32, the new file can't already exist
if (NormFile::Exists(newName)) _unlink(newName); if (NormFile::Exists(newName))
{
if (0 != _unlink(newName))
{
DMSG(0, "NormFile::Rename() _unlink() error: %s\n", GetErrorString());
return false;
}
}
// In Win32, the old file can't be open // In Win32, the old file can't be open
int oldFlags = 0; int oldFlags = 0;
if (IsOpen()) if (IsOpen())
@ -171,7 +181,7 @@ bool NormFile::Rename(const char* oldName, const char* newName)
oldFlags &= ~(O_CREAT | O_TRUNC); // unset these oldFlags &= ~(O_CREAT | O_TRUNC); // unset these
Close(); Close();
} }
#else #endif // WIN32
// Create sub-directories as needed. // Create sub-directories as needed.
char tempPath[PATH_MAX]; char tempPath[PATH_MAX];
strncpy(tempPath, newName, PATH_MAX); strncpy(tempPath, newName, PATH_MAX);
@ -198,29 +208,44 @@ bool NormFile::Rename(const char* oldName, const char* newName)
{ {
ptr = strchr(ptr, PROTO_PATH_DELIMITER); ptr = strchr(ptr, PROTO_PATH_DELIMITER);
if (ptr) *ptr = '\0'; if (ptr) *ptr = '\0';
#ifdef WIN32
if (0 != _mkdir(tempPath))
#else
if (mkdir(tempPath, 0755)) if (mkdir(tempPath, 0755))
#endif // if/else WIN32/UNIX
{ {
DMSG(0, "NormFile::Rename() mkdir(%s) error: %s\n", DMSG(0, "NormFile::Rename() mkdir(%s) error: %s\n",
tempPath, strerror(errno)); tempPath, GetErrorString());
return false; return false;
} }
if (ptr) *ptr++ = PROTO_PATH_DELIMITER; if (ptr) *ptr++ = PROTO_PATH_DELIMITER;
} }
#endif // if/else WIN32
if (rename(oldName, newName)) if (rename(oldName, newName))
{ {
DMSG(0, "NormFile::Rename() rename() error: %s\n", strerror(errno)); DMSG(0, "NormFile::Rename() rename() error: %s\n", strerror(errno));
#ifdef WIN32 #ifdef WIN32
//if (oldFlags) Open(oldName, oldFlags); if (oldFlags)
{
if (Open(oldName, oldFlags))
offset = 0;
else
DMSG(0, "NormFile::Rename() error re-opening file w/ old name\n");
}
#endif // WIN32 #endif // WIN32
return false; return false;
} }
else else
{ {
#ifdef WIN32 #ifdef WIN32
// (TBD) Is the file offset OK doing this??? // (TBD) Is the file offset OK doing this???
//if (oldFlags) Open(newName, oldFlags); if (oldFlags)
{
if (Open(newName, oldFlags))
offset = 0;
else
DMSG(0, "NormFile::Rename() error opening file w/ new name\n");
}
#endif // WIN32 #endif // WIN32
return true; return true;
} }
@ -234,7 +259,7 @@ int NormFile::Read(char* buffer, int len)
#else #else
int result = read(fd, buffer, len); int result = read(fd, buffer, len);
#endif // if/else WIN32 #endif // if/else WIN32
if (result > 0) offset += result; if (result > 0) offset += (off_t)result;
return result; return result;
} // end NormFile::Read() } // end NormFile::Read()
@ -246,7 +271,7 @@ int NormFile::Write(const char* buffer, int len)
#else #else
int result = write(fd, buffer, len); int result = write(fd, buffer, len);
#endif // if/else WIN32 #endif // if/else WIN32
if (result > 0) offset += result; if (result > 0) offset += (off_t)result;
return result; return result;
} // end NormFile::Write() } // end NormFile::Write()

8
common/normMessage.cpp
View File

@ -29,14 +29,14 @@ bool NormMsg::InitFromBuffer(UINT16 msgLength)
case DATA: case DATA:
// (TBD) look at "fec_id" to determine "fec_payload_id" size // (TBD) look at "fec_id" to determine "fec_payload_id" size
// (we _assume_ "fec_id" == 129 here // (we _assume_ "fec_id" == 129 here
if ((unsigned char)buffer[NormDataMsg::FEC_ID_OFFSET] == 129) if ((unsigned char)buffer[NormObjectMsg::FEC_ID_OFFSET] == 129)
{ {
header_length_base = 24; header_length_base = 24;
} }
else else
{ {
DMSG(0, "NormMsg::InitFromBuffer(DATA) unknown fec_id value: %u\n", DMSG(0, "NormMsg::InitFromBuffer(DATA) unknown fec_id value: %u\n",
buffer[NormDataMsg::FEC_ID_OFFSET]); buffer[NormObjectMsg::FEC_ID_OFFSET]);
return false; return false;
} }
break; break;
@ -54,7 +54,7 @@ bool NormMsg::InitFromBuffer(UINT16 msgLength)
else else
{ {
DMSG(0, "NormMsg::InitFromBuffer(FLUSH|SQUELCH) unknown fec_id value: %u\n", DMSG(0, "NormMsg::InitFromBuffer(FLUSH|SQUELCH) unknown fec_id value: %u\n",
buffer[NormDataMsg::FEC_ID_OFFSET]); buffer[NormCmdFlushMsg::FEC_ID_OFFSET]);
return false; return false;
} }
break; break;
@ -420,7 +420,7 @@ unsigned char NormQuantizeRtt(double rtt)
else if (rtt < NORM_RTT_MIN) else if (rtt < NORM_RTT_MIN)
rtt = NORM_RTT_MIN; rtt = NORM_RTT_MIN;
if (rtt < 3.3e-05) if (rtt < 3.3e-05)
return ((unsigned char)ceil((rtt*NORM_RTT_MIN)) - 1); return ((unsigned char)((rtt/NORM_RTT_MIN)) - 1);
else else
return ((unsigned char)(ceil(255.0 - (13.0*log(NORM_RTT_MAX/rtt))))); return ((unsigned char)(ceil(255.0 - (13.0*log(NORM_RTT_MAX/rtt)))));
} // end NormQuantizeRtt() } // end NormQuantizeRtt()

129
common/normMessage.h
View File

@ -20,7 +20,6 @@ const UINT8 NORM_PROTOCOL_VERSION = 1;
const int NORM_ROBUST_FACTOR = 20; // default robust factor const int NORM_ROBUST_FACTOR = 20; // default robust factor
// Pick a random number from 0..max // Pick a random number from 0..max
inline double UniformRand(double max) inline double UniformRand(double max)
{ {
@ -46,7 +45,7 @@ const double NORM_RTT_MAX = 1000.0;
inline double NormUnquantizeRtt(unsigned char qrtt) inline double NormUnquantizeRtt(unsigned char qrtt)
{ {
return ((qrtt < 31) ? return ((qrtt < 31) ?
(((double)(qrtt+1))/(double)NORM_RTT_MIN) : (((double)(qrtt+1))*(double)NORM_RTT_MIN) :
(NORM_RTT_MAX/exp(((double)(255-qrtt))/(double)13.0))); (NORM_RTT_MAX/exp(((double)(255-qrtt))/(double)13.0)));
} }
unsigned char NormQuantizeRtt(double rtt); unsigned char NormQuantizeRtt(double rtt);
@ -95,6 +94,7 @@ inline double NormUnquantizeRate(unsigned short rate)
} }
// This class is used to describe object "size" and/or "offset" // This class is used to describe object "size" and/or "offset"
// (TBD) This hokey implementation should use "off_t"
class NormObjectSize class NormObjectSize
{ {
public: public:
@ -103,6 +103,13 @@ class NormObjectSize
NormObjectSize(UINT32 size) : msb(0), lsb(size) {} NormObjectSize(UINT32 size) : msb(0), lsb(size) {}
NormObjectSize(UINT16 size) : msb(0), lsb(size) {} NormObjectSize(UINT16 size) : msb(0), lsb(size) {}
NormObjectSize(const NormObjectSize& size) : msb(size.msb), lsb(size.lsb) {} NormObjectSize(const NormObjectSize& size) : msb(size.msb), lsb(size.lsb) {}
NormObjectSize(off_t size)
{
lsb = size & 0x00000000ffffffff;
size >>= 32;
ASSERT(0 == (size & 0xffff0000));
msb = size & 0x0000ffff;
}
UINT16 MSB() const {return msb;} UINT16 MSB() const {return msb;}
UINT32 LSB() const {return lsb;} UINT32 LSB() const {return lsb;}
@ -150,15 +157,24 @@ class NormObjectSize
} }
NormObjectSize operator*(const NormObjectSize& b) const; NormObjectSize operator*(const NormObjectSize& b) const;
NormObjectSize operator/(const NormObjectSize& b) const; NormObjectSize operator/(const NormObjectSize& b) const;
off_t GetOffset() const
{
off_t offset = msb;
offset <<= 32;
offset += lsb;
return offset;
}
private: private:
UINT16 msb; UINT16 msb;
UINT32 lsb; UINT32 lsb;
}; // end class NormObjectSize }; // end class NormObjectSize
typedef UINT32 NormNodeId; #ifndef _NORM_API
const NormNodeId NORM_NODE_INVALID = 0x00000000; typedef unsigned long NormNodeId;
const NormNodeId NORM_NODE_ANY = 0xffffffff; const NormNodeId NORM_NODE_NONE = 0x00000000;
const NormNodeId NORM_NODE_ANY = 0xffffffff;
#endif // !_NORM_API
class NormObjectId class NormObjectId
{ {
@ -167,18 +183,12 @@ class NormObjectId
NormObjectId(UINT16 id) {value = id;} NormObjectId(UINT16 id) {value = id;}
NormObjectId(const NormObjectId& id) {value = id.value;} NormObjectId(const NormObjectId& id) {value = id.value;}
operator UINT16() const {return value;} operator UINT16() const {return value;}
int operator-(const NormObjectId& id) const INT16 operator-(const NormObjectId& id) const
{ {return ((INT16)(value - id.value));}
int result = value - id.value;
return ((0 == (result & 0x00008000)) ?
(result & 0x0000ffff) :
(((result != 0x00008000) || (value < id.value)) ?
result | 0xffff0000 : result));
}
bool operator<(const NormObjectId& id) const bool operator<(const NormObjectId& id) const
{return ((*this - id) < 0);} {return (((short)(value - id.value)) < 0);}
bool operator>(const NormObjectId& id) const bool operator>(const NormObjectId& id) const
{return ((*this - id) > 0);} {return (((INT16)(value - id.value)) > 0);}
bool operator<=(const NormObjectId& id) const bool operator<=(const NormObjectId& id) const
{return ((value == id.value) || (*this<id));} {return ((value == id.value) || (*this<id));}
bool operator>=(const NormObjectId& id) const bool operator>=(const NormObjectId& id) const
@ -187,7 +197,7 @@ class NormObjectId
{return (value == id.value);} {return (value == id.value);}
bool operator!=(const NormObjectId& id) const bool operator!=(const NormObjectId& id) const
{return (value != id.value);} {return (value != id.value);}
NormObjectId& operator++(int ) {value++; return *this;} NormObjectId& operator++(int) {value++; return *this;}
private: private:
UINT16 value; UINT16 value;
@ -204,17 +214,17 @@ class NormBlockId
{return (value == (UINT32)id);} {return (value == (UINT32)id);}
bool operator!=(const NormBlockId& id) const bool operator!=(const NormBlockId& id) const
{return (value != (UINT32)id);} {return (value != (UINT32)id);}
long operator-(const NormBlockId& id) const INT32 operator-(const NormBlockId& id) const
{return ((long)value - id.value);} {return ((INT32)value - id.value);}
bool operator<(const NormBlockId& id) const bool operator<(const NormBlockId& id) const
{return ((*this - id) < 0);} {return (((INT32)(value-id.value)) < 0);}
bool operator>(const NormBlockId& id) const bool operator>(const NormBlockId& id) const
{return ((*this - id) > 0);} {return (((INT32)(value-id.value)) > 0);}
NormBlockId& operator++(int ) {value++; return *this;} NormBlockId& operator++(int ) {value++; return *this;}
private: private:
UINT32 value; UINT32 value;
}; // end class NormObjectId }; // end class NormBlockId
typedef UINT16 NormSymbolId; typedef UINT16 NormSymbolId;
typedef NormSymbolId NormSegmentId; typedef NormSymbolId NormSegmentId;
@ -361,7 +371,7 @@ class NormMsg
void ExtendHeaderLength(UINT16 len) void ExtendHeaderLength(UINT16 len)
{ {
header_length += len; header_length += len;
length = header_length; length += len;
buffer[HDR_LEN_OFFSET] = header_length >> 2; buffer[HDR_LEN_OFFSET] = header_length >> 2;
} }
@ -624,6 +634,7 @@ class NormDataMsg : public NormObjectMsg
return (length - dataIndex); return (length - dataIndex);
} }
// These routines are only applicable to messages containing NORM_OBJECT_STREAM content // These routines are only applicable to messages containing NORM_OBJECT_STREAM content
// Some static helper routines for reading/writing embedded payload length/offsets // Some static helper routines for reading/writing embedded payload length/offsets
static UINT16 GetStreamPayloadHeaderLength() {return (PAYLOAD_DATA_OFFSET);} static UINT16 GetStreamPayloadHeaderLength() {return (PAYLOAD_DATA_OFFSET);}
@ -1049,7 +1060,7 @@ class NormRepairRequest
void SetForm(NormRepairRequest::Form theForm) {form = theForm;} void SetForm(NormRepairRequest::Form theForm) {form = theForm;}
void ResetFlags() {flags = 0;} void ResetFlags() {flags = 0;}
void SetFlag(NormRepairRequest::Flag theFlag) {flags |= theFlag;} void SetFlag(NormRepairRequest::Flag theFlag) {flags |= theFlag;}
void UnsetFlag(NormRepairRequest::Flag theFlag) {flags &= ~theFlag;} void ClearFlag(NormRepairRequest::Flag theFlag) {flags &= ~theFlag;}
// Returns length (each repair item requires 8 bytes of space) // Returns length (each repair item requires 8 bytes of space)
bool AppendRepairItem(const NormObjectId& objectId, bool AppendRepairItem(const NormObjectId& objectId,
@ -1417,6 +1428,7 @@ class NormAckMsg : public NormMsg
{ {
UINT16 len = MIN(payloadLen, segmentSize); UINT16 len = MIN(payloadLen, segmentSize);
memcpy(buffer+header_length, payload, len); memcpy(buffer+header_length, payload, len);
length += len;
return (payloadLen <= segmentSize); return (payloadLen <= segmentSize);
} }
@ -1439,7 +1451,7 @@ class NormAckMsg : public NormMsg
UINT16 GetPayloadLength() const {return (length - header_length);} UINT16 GetPayloadLength() const {return (length - header_length);}
const char* GetPayload() const {return (buffer + header_length);} const char* GetPayload() const {return (buffer + header_length);}
private: protected:
enum enum
{ {
SERVER_ID_OFFSET = MSG_OFFSET, SERVER_ID_OFFSET = MSG_OFFSET,
@ -1451,6 +1463,75 @@ class NormAckMsg : public NormMsg
}; };
}; // end class NormAckMsg }; // end class NormAckMsg
class NormAckFlushMsg : public NormAckMsg
{
public:
void Init()
{
SetType(ACK);
SetBaseHeaderLength(ACK_HEADER_LEN);
SetAckType(NormAck::FLUSH);
SetFecId(129); // only one supported for the moment
buffer[RESERVED_OFFSET] = 0;
length = ACK_HEADER_LEN+PAYLOAD_LENGTH;
}
// Note: must apply any header exts _before_ the payload is set
void SetFecId(UINT8 fecId)
{buffer[header_length+FEC_ID_OFFSET] = fecId;}
void SetObjectId(NormObjectId objectId)
{*((UINT16*)(buffer+header_length+OBJ_ID_OFFSET)) = htons((UINT16)objectId);}
void SetFecBlockId(const NormBlockId& blockId)
{
UINT32 temp32 = htonl((UINT32)blockId);
memcpy(buffer+header_length+BLOCK_ID_OFFSET, &temp32, 4);
}
void SetFecBlockLen(UINT16 blockLen)
{
blockLen = htons(blockLen);
memcpy(buffer+header_length+BLOCK_LEN_OFFSET, &blockLen, 2);
}
void SetFecSymbolId(UINT16 symbolId)
{
symbolId = htons(symbolId);
memcpy(buffer+header_length+SYMBOL_ID_OFFSET, &symbolId, 2);
}
// Message processing
UINT8 GetFecId() {return buffer[header_length+FEC_ID_OFFSET];}
NormObjectId GetObjectId() const
{
return (ntohs(*((UINT16*)(buffer+header_length+OBJ_ID_OFFSET))));
}
NormBlockId GetFecBlockId() const
{
return (ntohl(*((UINT32*)(buffer+header_length+BLOCK_ID_OFFSET))));
}
UINT16 GetFecBlockLen() const
{
return (ntohs(*((UINT16*)(buffer+header_length+BLOCK_LEN_OFFSET))));
}
UINT16 GetFecSymbolId() const
{
return (ntohs(*((UINT16*)(buffer+header_length+SYMBOL_ID_OFFSET))));
}
private:
// These are the payload offsets for "fec_id" = 129
// "fec_payload_id" field
enum
{
FEC_ID_OFFSET = 0,
RESERVED_OFFSET = FEC_ID_OFFSET + 1,
OBJ_ID_OFFSET = RESERVED_OFFSET + 1,
BLOCK_ID_OFFSET = OBJ_ID_OFFSET + 2,
BLOCK_LEN_OFFSET = BLOCK_ID_OFFSET + 4,
SYMBOL_ID_OFFSET = BLOCK_LEN_OFFSET + 2,
PAYLOAD_LENGTH = SYMBOL_ID_OFFSET + 2
};
}; // end class NormAckFlushMsg
class NormReportMsg : public NormMsg class NormReportMsg : public NormMsg
{ {
// (TBD) // (TBD)

674
common/normNode.cpp
View File

File diff suppressed because it is too large Load Diff

208
common/normNode.h
View File

@ -14,8 +14,10 @@ class NormNode
friend class NormNodeListIterator; friend class NormNodeListIterator;
public: public:
NormNode(class NormSession* theSession, NormNodeId nodeId); NormNode(class NormSession& theSession, NormNodeId nodeId);
virtual ~NormNode(); virtual ~NormNode();
void Retain();
void Release();
const ProtoAddress& GetAddress() const {return addr;} const ProtoAddress& GetAddress() const {return addr;}
void SetAddress(const ProtoAddress& address) {addr = address;} void SetAddress(const ProtoAddress& address) {addr = address;}
@ -24,12 +26,13 @@ class NormNode
inline const NormNodeId& LocalNodeId() const; inline const NormNodeId& LocalNodeId() const;
protected: protected:
class NormSession* session; class NormSession& session;
private: private:
NormNodeId id; NormNodeId id;
ProtoAddress addr; ProtoAddress addr;
// We keep NormNodes in a binary tree unsigned int reference_count;
// We keep NormNodes in a binary tree (TBD) make this a ProtoTree
NormNode* parent; NormNode* parent;
NormNode* right; NormNode* right;
NormNode* left; NormNode* left;
@ -99,24 +102,24 @@ class NormLossEstimator2
unsigned int LastLossInterval() {return history[1];} unsigned int LastLossInterval() {return history[1];}
private: private:
enum {DEPTH = 8}; enum {DEPTH = 8};
// Members // Members
bool init; bool init;
unsigned long lag_mask; unsigned long lag_mask;
unsigned int lag_depth; unsigned int lag_depth;
unsigned long lag_test_bit; unsigned long lag_test_bit;
unsigned short lag_index; unsigned short lag_index;
unsigned short event_window; unsigned short event_window;
unsigned short event_index; unsigned short event_index;
double event_window_time; double event_window_time;
double event_index_time; double event_index_time;
bool seeking_loss_event; bool seeking_loss_event;
bool no_loss; bool no_loss;
double initial_loss; double initial_loss;
double loss_interval; // EWMA of loss event interval double loss_interval; // EWMA of loss event interval
unsigned long history[9]; // loss interval history unsigned long history[9]; // loss interval history
double discount[9]; double discount[9];
@ -137,11 +140,33 @@ class NormLossEstimator2
}; // end class NormLossEstimator2 }; // end class NormLossEstimator2
class NormAckingNode : public NormNode
{
public:
NormAckingNode(class NormSession& theSession, NormNodeId nodeId);
~NormAckingNode();
bool IsPending() const
{return (!ack_received &&( req_count > 0));}
void Reset(unsigned int maxAttempts = NORM_ROBUST_FACTOR)
{
ack_received = false;
req_count = maxAttempts;
}
void DecrementReqCount() {if (req_count > 0) req_count--;}
unsigned int GetReqCount() const {return req_count;}
bool AckReceived() const {return ack_received;}
void MarkAckReceived() {ack_received = true;}
private:
bool ack_received; // was ack received?
unsigned int req_count; // remaining request attempts
}; // end NormAckingNode
class NormCCNode : public NormNode class NormCCNode : public NormNode
{ {
public: public:
NormCCNode(class NormSession* theSession, NormNodeId nodeId); NormCCNode(class NormSession& theSession, NormNodeId nodeId);
~NormCCNode(); ~NormCCNode();
bool IsClr() const {return is_clr;} bool IsClr() const {return is_clr;}
@ -182,9 +207,26 @@ class NormServerNode : public NormNode
public: public:
enum ObjectStatus {OBJ_INVALID, OBJ_NEW, OBJ_PENDING, OBJ_COMPLETE}; enum ObjectStatus {OBJ_INVALID, OBJ_NEW, OBJ_PENDING, OBJ_COMPLETE};
NormServerNode(class NormSession* theSession, NormNodeId nodeId); enum RepairBoundary {BLOCK_BOUNDARY, OBJECT_BOUNDARY};
NormServerNode(class NormSession& theSession, NormNodeId nodeId);
~NormServerNode(); ~NormServerNode();
// Parameters
NormObject::NackingMode GetDefaultNackingMode() const
{return default_nacking_mode;}
void SetDefaultNackingMode(NormObject::NackingMode nackingMode)
{default_nacking_mode = nackingMode;}
NormServerNode::RepairBoundary GetRepairBoundary() const
{return repair_boundary;}
// (TBD) force an appropriate RepairCheck on boundary change???
void SetRepairBoundary(RepairBoundary repairBoundary)
{repair_boundary = repairBoundary;}
bool UnicastNacks() {return unicast_nacks;}
void SetUnicastNacks(bool state) {unicast_nacks = state;}
bool UpdateLossEstimate(const struct timeval& currentTime, bool UpdateLossEstimate(const struct timeval& currentTime,
unsigned short theSequence, unsigned short theSequence,
bool ecnStatus = false); bool ecnStatus = false);
@ -201,10 +243,15 @@ class NormServerNode : public NormNode
void HandleNackMessage(const NormNackMsg& nack); void HandleNackMessage(const NormNackMsg& nack);
void HandleAckMessage(const NormAckMsg& ack); void HandleAckMessage(const NormAckMsg& ack);
bool Open(UINT16 sessionId, UINT16 segmentSize, UINT16 numData, UINT16 numParity); bool Open(UINT16 sessionId);
void Activate(); UINT16 GetSessionId() {return session_id;}
bool IsOpen() const {return is_open;}
void Close(); void Close();
bool IsOpen() const {return is_open;} bool AllocateBuffers(UINT16 segmentSize, UINT16 numData, UINT16 numParity);
bool BuffersAllocated() {return (0 != segment_size);}
void FreeBuffers();
void Activate();
bool SyncTest(const NormObjectMsg& msg) const; bool SyncTest(const NormObjectMsg& msg) const;
void Sync(NormObjectId objectId); void Sync(NormObjectId objectId);
@ -234,7 +281,7 @@ class NormServerNode : public NormNode
} }
void SetPending(NormObjectId objectId); void SetPending(NormObjectId objectId);
void DeleteObject(NormObject* obj); void DeleteObject(NormObject* obj, int which);
UINT16 SegmentSize() {return segment_size;} UINT16 SegmentSize() {return segment_size;}
UINT16 BlockSize() {return ndata;} UINT16 BlockSize() {return ndata;}
@ -287,6 +334,9 @@ class NormServerNode : public NormNode
private: private:
bool PassiveRepairCheck(NormObjectId objectId,
NormBlockId blockId,
NormSegmentId segmentId);
void RepairCheck(NormObject::CheckLevel checkLevel, void RepairCheck(NormObject::CheckLevel checkLevel,
NormObjectId objectId, NormObjectId objectId,
NormBlockId blockId, NormBlockId blockId,
@ -295,66 +345,80 @@ class NormServerNode : public NormNode
bool OnActivityTimeout(ProtoTimer& theTimer); bool OnActivityTimeout(ProtoTimer& theTimer);
bool OnRepairTimeout(ProtoTimer& theTimer); bool OnRepairTimeout(ProtoTimer& theTimer);
bool OnCCTimeout(ProtoTimer& theTimer); bool OnCCTimeout(ProtoTimer& theTimer);
bool OnAckTimeout(ProtoTimer& theTimer);
void AttachCCFeedback(NormAckMsg& ack);
void HandleRepairContent(const char* buffer, UINT16 bufferLen); void HandleRepairContent(const char* buffer, UINT16 bufferLen);
UINT16 session_id; UINT16 session_id;
bool synchronized; bool synchronized;
NormObjectId sync_id; // only valid if(synchronized) NormObjectId sync_id; // only valid if(synchronized)
NormObjectId next_id; // only valid if(synchronized) NormObjectId next_id; // only valid if(synchronized)
NormObjectId max_pending_object; // index for NACK construction
NormObjectId current_object_id; // index for suppression
UINT16 max_pending_range; // max range of pending objs allowed
bool is_open; bool is_open;
UINT16 segment_size; UINT16 segment_size;
UINT16 ndata; UINT16 ndata;
UINT16 nparity; UINT16 nparity;
NormObjectTable rx_table; NormObjectTable rx_table;
NormSlidingMask rx_pending_mask; NormSlidingMask rx_pending_mask;
NormSlidingMask rx_repair_mask; NormSlidingMask rx_repair_mask;
NormBlockPool block_pool; RepairBoundary repair_boundary;
NormSegmentPool segment_pool; NormObject::NackingMode default_nacking_mode;
NormDecoder decoder; bool unicast_nacks;
UINT16* erasure_loc; NormBlockPool block_pool;
NormSegmentPool segment_pool;
NormDecoder decoder;
UINT16* erasure_loc;
ProtoTimer activity_timer; bool server_active;
ProtoTimer repair_timer; ProtoTimer activity_timer;
NormObjectId current_object_id; // index for suppression ProtoTimer repair_timer;
NormObjectId max_pending_object; // index for NACK construction
// Watermark acknowledgement
ProtoTimer ack_timer;
NormObjectId watermark_object_id;
NormBlockId watermark_block_id;
NormSegmentId watermark_segment_id;
// Remote server grtt measurement state // Remote server grtt measurement state
double grtt_estimate; double grtt_estimate;
UINT8 grtt_quantized; UINT8 grtt_quantized;
struct timeval grtt_send_time; struct timeval grtt_send_time;
struct timeval grtt_recv_time; struct timeval grtt_recv_time;
double gsize_estimate; double gsize_estimate;
UINT8 gsize_quantized; UINT8 gsize_quantized;
double backoff_factor; double backoff_factor;
// Remote server congestion control state // Remote server congestion control state
NormLossEstimator2 loss_estimator; NormLossEstimator2 loss_estimator;
UINT16 cc_sequence; UINT16 cc_sequence;
bool cc_enable; bool cc_enable;
double cc_rate; // ccRate at start of cc_timer double cc_rate; // ccRate at start of cc_timer
ProtoTimer cc_timer; ProtoTimer cc_timer;
double rtt_estimate; double rtt_estimate;
UINT8 rtt_quantized; UINT8 rtt_quantized;
bool rtt_confirmed; bool rtt_confirmed;
bool is_clr; bool is_clr;
bool is_plr; bool is_plr;
bool slow_start; bool slow_start;
double send_rate; // sender advertised rate double send_rate; // sender advertised rate
double recv_rate; // measured recv rate double recv_rate; // measured recv rate
struct timeval prev_update_time; // for recv_rate measurement struct timeval prev_update_time; // for recv_rate measurement
unsigned long recv_accumulator; // for recv_rate measurement unsigned long recv_accumulator; // for recv_rate measurement
double nominal_packet_size; double nominal_packet_size;
// For statistics tracking // For statistics tracking
unsigned long recv_total; // total recvd accumulator unsigned long recv_total; // total recvd accumulator
unsigned long recv_goodput; // goodput recvd accumulator unsigned long recv_goodput; // goodput recvd accumulator
unsigned long resync_count; unsigned long resync_count;
unsigned long nack_count; unsigned long nack_count;
unsigned long suppress_count; unsigned long suppress_count;
unsigned long completion_count; unsigned long completion_count;
unsigned long failure_count; // usually due to re-syncs unsigned long failure_count; // usually due to re-syncs
}; // end class NormServerNode }; // end class NormServerNode

849
common/normObject.cpp
View File

File diff suppressed because it is too large Load Diff

143
common/normObject.h
View File

@ -14,6 +14,7 @@ class NormObject
public: public:
enum Type enum Type
{ {
NONE,
DATA, DATA,
FILE, FILE,
STREAM STREAM
@ -21,6 +22,7 @@ class NormObject
enum CheckLevel enum CheckLevel
{ {
TO_OBJECT,
THRU_INFO, THRU_INFO,
TO_BLOCK, TO_BLOCK,
THRU_SEGMENT, THRU_SEGMENT,
@ -28,7 +30,21 @@ class NormObject
THRU_OBJECT THRU_OBJECT
}; };
enum NackingMode
{
NACK_NONE,
NACK_INFO_ONLY,
NACK_NORMAL
};
virtual ~NormObject(); virtual ~NormObject();
void Retain();
void Release();
unsigned int GetReferenceCount() {return reference_count;}
// This must be reset after each update
void SetNotifyOnUpdate(bool state)
{notify_on_update = state;}
// Object information // Object information
NormObject::Type GetType() const {return type;} NormObject::Type GetType() const {return type;}
@ -39,19 +55,33 @@ class NormObject
UINT16 GetInfoLength() const {return info_len;} UINT16 GetInfoLength() const {return info_len;}
bool IsStream() const {return (STREAM == type);} bool IsStream() const {return (STREAM == type);}
class NormSession& GetSession() const {return session;}
NormNodeId LocalNodeId() const; NormNodeId LocalNodeId() const;
class NormServerNode* GetServer() {return server;} class NormServerNode* GetServer() const {return server;}
NormNodeId GetServerNodeId() const;
bool IsOpen() {return (0 != segment_size);} bool IsOpen() {return (0 != segment_size);}
// Opens (inits) object for tx operation // Opens (inits) object for tx operation
bool Open(const NormObjectSize& objectSize, bool Open(const NormObjectSize& objectSize,
const char* infoPtr, const char* infoPtr,
UINT16 infoLen); UINT16 infoLen,
UINT16 segmentSize,
UINT16 numData,
UINT16 numParity);
// Opens (inits) object for rx operation // Opens (inits) object for rx operation
bool Open(const NormObjectSize& objectSize, bool hasInfo) bool Open(const NormObjectSize& objectSize,
{return Open(objectSize, (char*)NULL, hasInfo ? 1 : 0);} bool hasInfo,
UINT16 segmentSize,
UINT16 numData,
UINT16 numParity)
{
return Open(objectSize, (char*)NULL, hasInfo ? 1 : 0,
segmentSize, numData, numParity);
}
void Close(); void Close();
virtual bool WriteSegment(NormBlockId blockId, virtual bool WriteSegment(NormBlockId blockId,
NormSegmentId segmentId, NormSegmentId segmentId,
const char* buffer) = 0; const char* buffer) = 0;
@ -59,9 +89,17 @@ class NormObject
NormSegmentId segmentId, NormSegmentId segmentId,
char* buffer) = 0; char* buffer) = 0;
NackingMode GetNackingMode() const {return nacking_mode;}
void SetNackingMode(NackingMode nackingMode)
{
nacking_mode = nackingMode;
// (TBD) initiate an appropriate NormServerNode::RepairCheck
// to prompt repair process if needed
}
// These are only valid after object is open // These are only valid after object is open
NormBlockId GetFinalBlockId() const {return final_block_id;} NormBlockId GetFinalBlockId() const {return final_block_id;}
UINT32 GetBlockSize(NormBlockId blockId) UINT32 GetBlockSize(NormBlockId blockId) const
{ {
return (((UINT32)blockId < large_block_count) ? large_block_size : return (((UINT32)blockId < large_block_count) ? large_block_size :
small_block_size); small_block_size);
@ -114,7 +152,8 @@ class NormObject
return result; return result;
} }
bool FindRepairIndex(NormBlockId& blockId, NormSegmentId& segmentId) const;
// Methods available to server for transmission // Methods available to server for transmission
bool NextServerMsg(NormObjectMsg* msg); bool NextServerMsg(NormObjectMsg* msg);
NormBlock* ServerRecoverBlock(NormBlockId blockId); NormBlock* ServerRecoverBlock(NormBlockId blockId);
@ -141,6 +180,7 @@ class NormObject
NormBlock* FindBlock(NormBlockId blockId) {return block_buffer.Find(blockId);} NormBlock* FindBlock(NormBlockId blockId) {return block_buffer.Find(blockId);}
bool ActivateRepairs(); bool ActivateRepairs();
bool IsRepairSet(NormBlockId blockId) {return repair_mask.Test(blockId);} bool IsRepairSet(NormBlockId blockId) {return repair_mask.Test(blockId);}
bool IsPendingSet(NormBlockId blockId) {return pending_mask.Test(blockId);}
bool AppendRepairAdv(NormCmdRepairAdvMsg& cmd); bool AppendRepairAdv(NormCmdRepairAdvMsg& cmd);
NormBlockId GetMaxPendingBlockId() const {return max_pending_block;} NormBlockId GetMaxPendingBlockId() const {return max_pending_block;}
@ -161,7 +201,8 @@ class NormObject
// Used by receiver for resource management scheme // Used by receiver for resource management scheme
NormBlock* StealNewestBlock(bool excludeBlock, NormBlockId excludeId = 0); NormBlock* StealNewestBlock(bool excludeBlock, NormBlockId excludeId = 0);
NormBlock* StealOldestBlock(bool excludeBlock, NormBlockId excludeId = 0); NormBlock* StealOldestBlock(bool excludeBlock, NormBlockId excludeId = 0);
bool PassiveRepairCheck(NormBlockId blockId,
NormSegmentId segmentId);
bool ClientRepairCheck(CheckLevel level, bool ClientRepairCheck(CheckLevel level,
NormBlockId blockId, NormBlockId blockId,
NormSegmentId segmentId, NormSegmentId segmentId,
@ -178,15 +219,16 @@ class NormObject
protected: protected:
NormObject(Type theType, NormObject(Type theType,
class NormSession* theSession, class NormSession& theSession,
class NormServerNode* theServer, class NormServerNode* theServer,
const NormObjectId& objectId); const NormObjectId& objectId);
void Accept() {accepted = true;} void Accept() {accepted = true;}
NormObject::Type type; NormObject::Type type;
class NormSession* session; class NormSession& session;
class NormServerNode* server; // NULL value indicates local (tx) object class NormServerNode* server; // NULL value indicates local (tx) object
unsigned int reference_count;
NormObjectId transport_id; NormObjectId transport_id;
NormObjectSize object_size; NormObjectSize object_size;
@ -194,9 +236,9 @@ class NormObject
UINT16 ndata; UINT16 ndata;
UINT16 nparity; UINT16 nparity;
NormBlockBuffer block_buffer; NormBlockBuffer block_buffer;
bool pending_info; bool pending_info; // set when we need to send or recv info
NormSlidingMask pending_mask; NormSlidingMask pending_mask;
bool repair_info; bool repair_info; // client: set when
NormSlidingMask repair_mask; NormSlidingMask repair_mask;
NormBlockId current_block_id; // for suppression NormBlockId current_block_id; // for suppression
NormSegmentId next_segment_id; // for suppression NormSegmentId next_segment_id; // for suppression
@ -208,11 +250,14 @@ class NormObject
UINT32 small_block_size; UINT32 small_block_size;
NormBlockId final_block_id; NormBlockId final_block_id;
UINT16 final_segment_size; UINT16 final_segment_size;
NackingMode nacking_mode;
char* info; char* info;
UINT16 info_len; UINT16 info_len;
// Here are some members used to let us know
// our status with respect to the rest of the world
bool accepted; bool accepted;
bool notify_on_update;
NormObject* next; NormObject* next;
}; // end class NormObject }; // end class NormObject
@ -221,7 +266,7 @@ class NormObject
class NormFileObject : public NormObject class NormFileObject : public NormObject
{ {
public: public:
NormFileObject(class NormSession* theSession, NormFileObject(class NormSession& theSession,
class NormServerNode* theServer, class NormServerNode* theServer,
const NormObjectId& objectId); const NormObjectId& objectId);
~NormFileObject(); ~NormFileObject();
@ -231,7 +276,8 @@ class NormFileObject : public NormObject
UINT16 infoLen = 0); UINT16 infoLen = 0);
bool Accept(const char* thePath); bool Accept(const char* thePath);
void Close(); void Close();
const char* Path() {return path;}
const char* GetPath() {return path;}
bool Rename(const char* newPath) bool Rename(const char* newPath)
{ {
bool result = file.Rename(path, newPath); bool result = file.Rename(path, newPath);
@ -254,11 +300,44 @@ class NormFileObject : public NormObject
NormObjectSize small_block_length; NormObjectSize small_block_length;
}; // end class NormFileObject }; // end class NormFileObject
class NormDataObject : public NormObject
{
// (TBD) allow support of greater than 4GB size data objects
public:
NormDataObject(class NormSession& theSession,
class NormServerNode* theServer,
const NormObjectId& objectId);
~NormDataObject();
bool Open(char* dataPtr,
UINT32 dataLen,
const char* infoPtr = NULL,
UINT16 infoLen = 0);
bool Accept(char* dataPtr, UINT32 dataMax);
void Close();
const char* GetData() {return data_ptr;}
virtual bool WriteSegment(NormBlockId blockId,
NormSegmentId segmentId,
const char* buffer);
virtual UINT16 ReadSegment(NormBlockId blockId,
NormSegmentId segmentId,
char* buffer);
private:
NormObjectSize large_block_length;
NormObjectSize small_block_length;
char* data_ptr;
UINT32 data_max;
}; // end class NormDataObject
class NormStreamObject : public NormObject class NormStreamObject : public NormObject
{ {
public: public:
NormStreamObject(class NormSession* theSession, NormStreamObject(class NormSession& theSession,
class NormServerNode* theServer, class NormServerNode* theServer,
const NormObjectId& objectId); const NormObjectId& objectId);
~NormStreamObject(); ~NormStreamObject();
@ -269,19 +348,35 @@ class NormStreamObject : public NormObject
void Close(); void Close();
bool Accept(UINT32 bufferSize); bool Accept(UINT32 bufferSize);
enum FlushType enum FlushMode
{ {
FLUSH_NONE, // no flush action taken FLUSH_NONE, // no flush action taken
FLUSH_PASSIVE, // pending queued data is transmitted, but no CMD(FLUSH) sent FLUSH_PASSIVE, // pending queued data is transmitted, but no CMD(FLUSH) sent
FLUSH_ACTIVE // pending queued data is transmitted, _and_ active CMD(FLUSH) FLUSH_ACTIVE // pending queued data is transmitted, _and_ active CMD(FLUSH)
}; };
void SetFlushMode(FlushMode flushMode) {flush_mode = flushMode;}
void Flush(bool eom = false)
{
FlushMode oldFlushMode = flush_mode;
SetFlushMode(FLUSH_ACTIVE);
Write(NULL, 0, eom);
SetFlushMode(oldFlushMode);
}
void SetPushMode(bool state) {push_mode = state;}
bool Read(char* buffer, unsigned int* buflen, bool findMsgStart = false); bool Read(char* buffer, unsigned int* buflen, bool findMsgStart = false);
UINT32 Write(const char* buffer, UINT32 len, FlushType flushType, bool eom, bool push); UINT32 Write(const char* buffer, UINT32 len, bool eom = false);
bool SyncOffsetIsValid() {return sync_offset_valid;}
UINT32 GetSyncOffset() {return sync_offset;}
UINT32 GetCurrentReadOffset() {return read_offset;}
bool StreamUpdateStatus(NormBlockId blockId); bool StreamUpdateStatus(NormBlockId blockId);
void StreamResync(NormBlockId nextBlockId) void StreamResync(NormBlockId nextBlockId)
{stream_next_id = nextBlockId;} {
stream_next_id = nextBlockId;
}
void StreamAdvance(); void StreamAdvance();
virtual bool WriteSegment(NormBlockId blockId, virtual bool WriteSegment(NormBlockId blockId,
@ -308,7 +403,11 @@ class NormStreamObject : public NormObject
NormSegmentId FlushSegmentId() NormSegmentId FlushSegmentId()
{return (write_index.segment ? (write_index.segment-1) : {return (write_index.segment ? (write_index.segment-1) :
(ndata-1));} (ndata-1));}
NormBlockId GetNextBlockId() const
{return (server ? read_index.block : write_index.block);}
NormSegmentId GetNextSegmentId() const
{return (server ? read_index.segment : write_index.segment);}
private: private:
class Index class Index
{ {
@ -320,6 +419,8 @@ class NormStreamObject : public NormObject
bool stream_sync; bool stream_sync;
NormBlockId stream_sync_id; NormBlockId stream_sync_id;
NormBlockId stream_next_id; NormBlockId stream_next_id;
UINT32 sync_offset;
bool sync_offset_valid;
NormBlockPool block_pool; NormBlockPool block_pool;
NormSegmentPool segment_pool; NormSegmentPool segment_pool;
@ -330,6 +431,8 @@ class NormStreamObject : public NormObject
UINT32 read_offset; UINT32 read_offset;
bool flush_pending; bool flush_pending;
bool msg_start; bool msg_start;
FlushMode flush_mode;
bool push_mode;
}; // end class NormStreamObject }; // end class NormStreamObject
#ifdef SIMULATE #ifdef SIMULATE
@ -338,7 +441,7 @@ class NormStreamObject : public NormObject
class NormSimObject : public NormObject class NormSimObject : public NormObject
{ {
public: public:
NormSimObject(class NormSession* theSession, NormSimObject(class NormSession& theSession,
class NormServerNode* theServer, class NormServerNode* theServer,
const NormObjectId& objectId); const NormObjectId& objectId);
~NormSimObject(); ~NormSimObject();

6
common/normPostProcess.cpp
View File

@ -56,7 +56,6 @@ void NormPostProcessor::GetCommand(char* buffer, unsigned int buflen)
bool NormPostProcessor::SetCommand(const char* cmd) bool NormPostProcessor::SetCommand(const char* cmd)
{ {
// 1) Delete old command resources // 1) Delete old command resources
if (process_argv) if (process_argv)
{ {
@ -81,10 +80,10 @@ bool NormPostProcessor::SetCommand(const char* cmd)
{ {
argCount++; argCount++;
while (!isspace(*ptr) && ('\0' != *ptr)) ptr++; while (!isspace(*ptr) && ('\0' != *ptr)) ptr++;
while (isspace(*ptr) && ('\0' != *ptr)) ptr++;
} }
if (!argCount) return true; // post processing disabled if (!argCount) return true; // post processing disabled
// 3) Allocate new process_argv array (2 extra slots, one for "target", // 3) Allocate new process_argv array (2 extra slots, one for "target",
// and one for terminating NULL pointer. // and one for terminating NULL pointer.
if (!(process_argv = new char*[argCount+2])) if (!(process_argv = new char*[argCount+2]))
@ -115,7 +114,8 @@ bool NormPostProcessor::SetCommand(const char* cmd)
} }
strncpy(arg, argStart, argLength); strncpy(arg, argStart, argLength);
arg[argLength] = '\0'; arg[argLength] = '\0';
process_argv[index] = arg; process_argv[index++] = arg;
while (isspace(*ptr) && ('\0' != *ptr)) ptr++;
} }
return true; return true;
} // end NormPostProcessor::SetCommand() } // end NormPostProcessor::SetCommand()

97
common/normSegment.cpp
View File

@ -257,6 +257,7 @@ bool NormBlock::ActivateRepairs(UINT16 numParity)
if (repair_mask.IsSet()) if (repair_mask.IsSet())
{ {
pending_mask.Add(repair_mask); pending_mask.Add(repair_mask);
ASSERT(pending_mask.IsSet());
repair_mask.Clear(); repair_mask.Clear();
return true; return true;
} }
@ -390,7 +391,7 @@ bool NormBlock::HandleSegmentRequest(NormSegmentId nextId, NormSegmentId lastId,
return increasedRepair; return increasedRepair;
} // end NormBlock::HandleSegmentRequest() } // end NormBlock::HandleSegmentRequest()
// (TBD) this should return true is something is appending, false otherwise
bool NormBlock::AppendRepairAdv(NormCmdRepairAdvMsg& cmd, bool NormBlock::AppendRepairAdv(NormCmdRepairAdvMsg& cmd,
NormObjectId objectId, NormObjectId objectId,
bool repairInfo, bool repairInfo,
@ -433,7 +434,13 @@ bool NormBlock::AppendRepairAdv(NormCmdRepairAdvMsg& cmd,
if (form != prevForm) if (form != prevForm)
{ {
if (NormRepairRequest::INVALID != prevForm) if (NormRepairRequest::INVALID != prevForm)
cmd.PackRepairRequest(req); // (TBD) error check {
if (0 == cmd.PackRepairRequest(req))
{
DMSG(0, "NormBlock::AppendRepairAdv() warning: full msg\n");
break;
}
}
req.SetForm(form); req.SetForm(form);
cmd.AttachRepairRequest(req, segmentSize); // (TBD) error check cmd.AttachRepairRequest(req, segmentSize); // (TBD) error check
prevForm = form; prevForm = form;
@ -460,12 +467,16 @@ bool NormBlock::AppendRepairAdv(NormCmdRepairAdvMsg& cmd,
} }
} // end while (nextId < totalSize) } // end while (nextId < totalSize)
if (NormRepairRequest::INVALID != prevForm) if (NormRepairRequest::INVALID != prevForm)
cmd.PackRepairRequest(req); // (TBD) error check {
if (0 == cmd.PackRepairRequest(req))
DMSG(0, "NormBlock::AppendRepairAdv() warning: full msg\n");
}
} }
return true; return true;
} // end NormBlock::AppendRepairAdv() } // end NormBlock::AppendRepairAdv()
// Called by client // Called by client
// (TBD) this should return true iff something appended, false otherwise
bool NormBlock::AppendRepairRequest(NormNackMsg& nack, bool NormBlock::AppendRepairRequest(NormNackMsg& nack,
UINT16 numData, UINT16 numData,
UINT16 numParity, UINT16 numParity,
@ -527,7 +538,13 @@ bool NormBlock::AppendRepairRequest(NormNackMsg& nack,
if (form != prevForm) if (form != prevForm)
{ {
if (NormRepairRequest::INVALID != prevForm) if (NormRepairRequest::INVALID != prevForm)
nack.PackRepairRequest(req); // (TBD) error check {
if (0 == nack.PackRepairRequest(req))
{
DMSG(0, "NormBlock::AppendRepairRequest() warning: full NACK msg\n");
break;
}
}
nack.AttachRepairRequest(req, segmentSize); // (TBD) error check nack.AttachRepairRequest(req, segmentSize); // (TBD) error check
req.SetForm(form); req.SetForm(form);
prevForm = form; prevForm = form;
@ -554,76 +571,10 @@ bool NormBlock::AppendRepairRequest(NormNackMsg& nack,
} }
} // end while (nextId < lastId) } // end while (nextId < lastId)
if (NormRepairRequest::INVALID != prevForm) if (NormRepairRequest::INVALID != prevForm)
nack.PackRepairRequest(req); // (TBD) error check
/*
// old code begins here
NormSegmentId prevId = nextId;
while ((nextId <= lastId) || (segmentCount > 0))
{ {
// force break of possible ending consec. series if (0 == nack.PackRepairRequest(req))
if (nextId == lastId) nextId++; DMSG(0, "NormBlock::AppendRepairRequest() warning: full NACK msg\n");
if (segmentCount && (segmentCount == (nextId - prevId))) }
{
segmentCount++; // consecutive series continues
}
else
{
NormRepairRequest::Form nextForm;
switch(segmentCount)
{
case 0:
nextForm = NormRepairRequest::INVALID;
break;
case 1:
case 2:
nextForm = NormRepairRequest::ITEMS;
break;
default:
nextForm = NormRepairRequest::RANGES;
break;
} // end switch(reqCount)
if (prevForm != nextForm)
{
if (NormRepairRequest::INVALID != prevForm)
nack.PackRepairRequest(req); // (TBD) error check
if (NormRepairRequest::INVALID != nextForm)
{
nack.AttachRepairRequest(req, segmentSize);
req.SetForm(nextForm);
req.SetFlag(NormRepairRequest::SEGMENT);
if (pendingInfo) req.SetFlag(NormRepairRequest::INFO);
}
prevForm = nextForm;
}
if (NormRepairRequest::INVALID != nextForm)
DMSG(6, "NormBlock::AppendRepairRequest() SEGMENT request\n");
switch (nextForm)
{
case NormRepairRequest::ITEMS:
req.AppendRepairItem(objectId, id, prevId); // (TBD) error check
if (2 == reqCount)
req.AppendRepairItem(objectId, id, prevId+1); // (TBD) error check
break;
case NormRepairRequest::RANGES:
req.AppendRepairItem(objectId, id, prevId); // (TBD) error check
req.AppendRepairItem(objectId, id, prevId+reqCount-1); // (TBD) error check
break;
default:
break;
} // end switch(nextForm)
prevId = nextId;
if (nextId < lastId)
segmentCount = 1;
else
segmentCount = 0;
} // end if/else (reqCount && (reqCount == (nextId - prevId)))
nextId++;
if (nextId <= lastId) nextId = pending_mask.NextSet(nextId);
} // end while(nextId <= lastId)
if (NormRepairRequest::INVALID != prevForm)
nack.PackRepairRequest(req); // (TBD) error check
*/
return true; return true;
} // end NormBlock::AppendRepairRequest() } // end NormBlock::AppendRepairRequest()

650
common/normSession.cpp
View File

File diff suppressed because it is too large Load Diff

279
common/normSession.h
View File

@ -13,11 +13,20 @@ class NormController
public: public:
enum Event enum Event
{ {
EVENT_INVALID = 0,
TX_QUEUE_VACANCY,
TX_QUEUE_EMPTY, TX_QUEUE_EMPTY,
TX_OBJECT_SENT,
TX_OBJECT_PURGED,
LOCAL_SERVER_CLOSED,
REMOTE_SERVER_NEW,
REMOTE_SERVER_INACTIVE,
REMOTE_SERVER_ACTIVE,
RX_OBJECT_NEW, RX_OBJECT_NEW,
RX_OBJECT_INFO, RX_OBJECT_INFO,
RX_OBJECT_UPDATE, RX_OBJECT_UPDATE,
RX_OBJECT_COMPLETE, RX_OBJECT_COMPLETED,
RX_OBJECT_ABORTED
}; };
virtual void Notify(NormController::Event event, virtual void Notify(NormController::Event event,
@ -54,9 +63,10 @@ class NormSessionMgr
} }
void ActivateTimer(ProtoTimer& timer) {timer_mgr.ActivateTimer(timer);} void ActivateTimer(ProtoTimer& timer) {timer_mgr.ActivateTimer(timer);}
ProtoTimerMgr& GetTimerMgr() {return timer_mgr;} ProtoTimerMgr& GetTimerMgr() const {return timer_mgr;}
ProtoSocket::Notifier& GetSocketNotifier() {return socket_notifier;} ProtoSocket::Notifier& GetSocketNotifier() const {return socket_notifier;}
NormController* GetController() const {return controller;}
private: private:
ProtoTimerMgr& timer_mgr; ProtoTimerMgr& timer_mgr;
ProtoSocket::Notifier& socket_notifier; ProtoSocket::Notifier& socket_notifier;
@ -86,26 +96,34 @@ class NormSession
static const UINT16 DEFAULT_NPARITY; static const UINT16 DEFAULT_NPARITY;
// General methods // General methods
const NormNodeId& LocalNodeId() {return local_node_id;} const NormNodeId& LocalNodeId() const {return local_node_id;}
bool Open(const char* interfaceName = NULL); bool Open(const char* interfaceName = NULL);
void Close(); void Close();
bool IsOpen() {return (rx_socket.IsOpen() || tx_socket.IsOpen());} bool IsOpen() {return (rx_socket.IsOpen() || tx_socket.IsOpen());}
const ProtoAddress& Address() {return address;} const ProtoAddress& Address() {return address;}
void SetAddress(const ProtoAddress& addr) {address = addr;} void SetAddress(const ProtoAddress& addr) {address = addr;}
bool SetTTL(UINT8 theTTL)
{
bool result = tx_socket.IsOpen() ? tx_socket.SetTTL(theTTL) : true;
ttl = result ? theTTL : ttl;
return result;
}
bool SetLoopback(bool state)
{
bool result = tx_socket.IsOpen() ? tx_socket.SetLoopback(state) : true;
loopback = result ? state : loopback;
return result;
}
static double CalculateRate(double size, double rtt, double loss); static double CalculateRate(double size, double rtt, double loss);
NormSessionMgr& GetSessionMgr() {return session_mgr;}
// Session parameters // Session parameters
double TxRate() {return (tx_rate * 8.0);} double TxRate() {return (tx_rate * 8.0);}
// (TBD) watch timer scheduling and min/max bounds // (TBD) watch timer scheduling and min/max bounds
void SetTxRate(double txRate) {tx_rate = txRate / 8.0;} void SetTxRate(double txRate);
double BackoffFactor() {return backoff_factor;} double BackoffFactor() {return backoff_factor;}
void SetBackoffFactor(double value) {backoff_factor = value;} void SetBackoffFactor(double value) {backoff_factor = value;}
void SetLoopback(bool state)
{
rx_socket.SetLoopback(state);
tx_socket.SetLoopback(state);
}
bool CongestionControl() {return cc_enable;} bool CongestionControl() {return cc_enable;}
void SetCongestionControl(bool state) {cc_enable = state;} void SetCongestionControl(bool state) {cc_enable = state;}
@ -130,11 +148,12 @@ class NormSession
next_tx_object_id = IsServer() ? next_tx_object_id : baseId; next_tx_object_id = IsServer() ? next_tx_object_id : baseId;
session_id = IsServer() ? session_id : (UINT16)baseId; session_id = IsServer() ? session_id : (UINT16)baseId;
} }
bool StartServer(unsigned long bufferSpace, bool IsServer() {return is_server;}
UINT16 segmentSize, bool StartServer(UINT32 bufferSpace,
UINT16 numData, UINT16 segmentSize,
UINT16 numParity, UINT16 numData,
const char* interfaceName = NULL); UINT16 numParity,
const char* interfaceName = NULL);
void StopServer(); void StopServer();
NormStreamObject* QueueTxStream(UINT32 bufferSize, NormStreamObject* QueueTxStream(UINT32 bufferSize,
const char* infoPtr = NULL, const char* infoPtr = NULL,
@ -142,15 +161,27 @@ class NormSession
NormFileObject* QueueTxFile(const char* path, NormFileObject* QueueTxFile(const char* path,
const char* infoPtr = NULL, const char* infoPtr = NULL,
UINT16 infoLen = 0); UINT16 infoLen = 0);
NormDataObject* QueueTxData(const char* dataPtr,
bool IsServer() {return is_server;} UINT32 dataLen,
UINT16 ServerSegmentSize() {return segment_size;} const char* infoPtr = NULL,
UINT16 ServerBlockSize() {return ndata;} UINT16 infoLen = 0);
UINT16 ServerNumParity() {return nparity;} void DeleteTxObject(NormObject* obj);
UINT16 ServerAutoParity() {return auto_parity;}
// postive ack mgmnt
void ServerSetWatermark(NormObjectId objectId,
NormBlockId blockId,
NormSegmentId segmentId);
bool ServerAddAckingNode(NormNodeId nodeId);
void ServerRemoveAckingNode(NormNodeId nodeId);
UINT16 ServerSegmentSize() const {return segment_size;}
UINT16 ServerBlockSize() const {return ndata;}
UINT16 ServerNumParity() const {return nparity;}
UINT16 ServerAutoParity() const {return auto_parity;}
void ServerSetAutoParity(UINT16 autoParity) void ServerSetAutoParity(UINT16 autoParity)
{ASSERT(autoParity <= nparity); auto_parity = autoParity;} {ASSERT(autoParity <= nparity); auto_parity = autoParity;}
UINT16 ServerExtraParity() {return extra_parity;} UINT16 ServerExtraParity() const {return extra_parity;}
void ServerSetExtraParity(UINT16 extraParity) void ServerSetExtraParity(UINT16 extraParity)
{extra_parity = extraParity;} {extra_parity = extraParity;}
@ -161,6 +192,13 @@ class NormSession
objectId = (UINT16)index; objectId = (UINT16)index;
return result; return result;
} }
bool ServerGetFirstRepairPending(NormObjectId& objectId)
{
UINT32 index;
bool result = tx_repair_mask.GetFirstSet(index);
objectId = (UINT16)index;
return result;
}
double ServerGrtt() {return grtt_advertised;} double ServerGrtt() {return grtt_advertised;}
void ServerSetGrtt(double grttValue) void ServerSetGrtt(double grttValue)
@ -192,15 +230,7 @@ class NormSession
void ServerPutFreeSegment(char* segment) {segment_pool.Put(segment);} void ServerPutFreeSegment(char* segment) {segment_pool.Put(segment);}
void PromptServer() void PromptServer() {QueueMessage(NULL);}
{
if (!tx_timer.IsActive())
{
tx_timer.SetInterval(0.0);
ActivateTimer(tx_timer);
}
}
void TouchServer() void TouchServer()
{ {
@ -213,13 +243,23 @@ class NormSession
bool StartClient(unsigned long bufferSpace, bool StartClient(unsigned long bufferSpace,
const char* interfaceName = NULL); const char* interfaceName = NULL);
void StopClient(); void StopClient();
bool IsClient() {return is_client;} bool IsClient() const {return is_client;}
unsigned long RemoteServerBufferSize() unsigned long RemoteServerBufferSize() const
{return remote_server_buffer_size;} {return remote_server_buffer_size;}
void SetUnicastNacks(bool state) {unicast_nacks = state;} void SetUnicastNacks(bool state) {unicast_nacks = state;}
bool UnicastNacks() {return unicast_nacks;} bool UnicastNacks() const {return unicast_nacks;}
void ClientSetSilent(bool state) {client_silent = state;} void ClientSetSilent(bool state) {client_silent = state;}
bool ClientIsSilent() {return client_silent;} bool ClientIsSilent() const {return client_silent;}
NormObject::NackingMode ClientGetDefaultNackingMode() const
{return default_nacking_mode;}
void ClientSetDefaultNackingMode(NormObject::NackingMode nackingMode)
{default_nacking_mode = nackingMode;}
NormServerNode::RepairBoundary ClientGetDefaultRepairBoundary() const
{return default_repair_boundary;}
void ClientSetDefaultRepairBoundary(NormServerNode::RepairBoundary repairBoundary)
{default_repair_boundary = repairBoundary;}
// Debug settings // Debug settings
void SetTrace(bool state) {trace = state;} void SetTrace(bool state) {trace = state;}
@ -239,13 +279,12 @@ class NormSession
~NormSession(); ~NormSession();
void Serve(); void Serve();
bool QueueTxObject(NormObject* obj, bool touchServer = true); bool QueueTxObject(NormObject* obj);
void DeleteTxObject(NormObject* obj);
bool OnTxTimeout(ProtoTimer& theTimer); bool OnTxTimeout(ProtoTimer& theTimer);
bool OnRepairTimeout(ProtoTimer& theTimer); bool OnRepairTimeout(ProtoTimer& theTimer);
bool OnFlushTimeout(ProtoTimer& theTimer); bool OnFlushTimeout(ProtoTimer& theTimer);
bool OnWatermarkTimeout(ProtoTimer& theTimer);
bool OnProbeTimeout(ProtoTimer& theTimer); bool OnProbeTimeout(ProtoTimer& theTimer);
bool OnReportTimeout(ProtoTimer& theTimer); bool OnReportTimeout(ProtoTimer& theTimer);
@ -271,6 +310,7 @@ class NormSession
void AdjustRate(bool onResponse); void AdjustRate(bool onResponse);
bool ServerQueueSquelch(NormObjectId objectId); bool ServerQueueSquelch(NormObjectId objectId);
void ServerQueueFlush(); void ServerQueueFlush();
bool ServerQueueWatermarkFlush();
bool ServerBuildRepairAdv(NormCmdRepairAdvMsg& cmd); bool ServerBuildRepairAdv(NormCmdRepairAdvMsg& cmd);
void ServerUpdateGroupSize(); void ServerUpdateGroupSize();
@ -282,102 +322,111 @@ class NormSession
void ClientHandleNackMessage(const NormNackMsg& nack); void ClientHandleNackMessage(const NormNackMsg& nack);
void ClientHandleAckMessage(const NormAckMsg& ack); void ClientHandleAckMessage(const NormAckMsg& ack);
NormSessionMgr& session_mgr; NormSessionMgr& session_mgr;
bool notify_pending; bool notify_pending;
ProtoTimer tx_timer; ProtoTimer tx_timer;
ProtoSocket tx_socket; ProtoSocket tx_socket;
ProtoSocket rx_socket; ProtoSocket rx_socket;
NormMessageQueue message_queue; NormMessageQueue message_queue;
NormMessageQueue message_pool; NormMessageQueue message_pool;
ProtoTimer report_timer; ProtoTimer report_timer;
UINT16 tx_sequence; UINT16 tx_sequence;
// General session parameters // General session parameters
NormNodeId local_node_id; NormNodeId local_node_id;
ProtoAddress address; // session destination address & port ProtoAddress address; // session destination address & port
UINT8 ttl; // session multicast ttl UINT8 ttl; // session multicast ttl
double tx_rate; // bytes per second bool loopback; // to receive own traffic
double backoff_factor; double tx_rate; // bytes per second
double backoff_factor;
// Server parameters and state // Server parameters and state
bool is_server; bool is_server;
UINT16 session_id; UINT16 session_id;
UINT16 segment_size; UINT16 segment_size;
UINT16 ndata; UINT16 ndata;
UINT16 nparity; UINT16 nparity;
UINT16 auto_parity; UINT16 auto_parity;
UINT16 extra_parity; UINT16 extra_parity;
NormObjectTable tx_table; NormObjectTable tx_table;
NormSlidingMask tx_pending_mask; NormSlidingMask tx_pending_mask;
NormSlidingMask tx_repair_mask; NormSlidingMask tx_repair_mask;
ProtoTimer repair_timer; ProtoTimer repair_timer;
NormBlockPool block_pool; NormBlockPool block_pool;
NormSegmentPool segment_pool; NormSegmentPool segment_pool;
NormEncoder encoder; NormEncoder encoder;
NormObjectId next_tx_object_id; NormObjectId next_tx_object_id;
unsigned int tx_cache_count_min; unsigned int tx_cache_count_min;
unsigned int tx_cache_count_max; unsigned int tx_cache_count_max;
NormObjectSize tx_cache_size_max; NormObjectSize tx_cache_size_max;
ProtoTimer flush_timer; ProtoTimer flush_timer;
int flush_count; int flush_count;
bool posted_tx_queue_empty; bool posted_tx_queue_empty;
ProtoTimer watermark_timer;
int watermark_count; // For postive acknowledgement collection
// (TBD) watermark_object_id, watermark_block_id, watermark_symbol_id NormNodeTree acking_node_tree;
unsigned int acking_node_count;
bool watermark_pending;
NormObjectId watermark_object_id;
NormBlockId watermark_block_id;
NormSegmentId watermark_segment_id;
unsigned int acks_collected;
// for unicast nack/cc feedback suppression // for unicast nack/cc feedback suppression
bool advertise_repairs; bool advertise_repairs;
bool suppress_nonconfirmed; bool suppress_nonconfirmed;
double suppress_rate; double suppress_rate;
double suppress_rtt; double suppress_rtt;
ProtoTimer probe_timer; // GRTT/congestion control probes ProtoTimer probe_timer; // GRTT/congestion control probes
bool probe_proactive; bool probe_proactive;
bool probe_pending; // true while CMD(CC) enqueued bool probe_pending; // true while CMD(CC) enqueued
bool probe_reset; bool probe_reset;
double grtt_interval; // current GRTT update interval double grtt_interval; // current GRTT update interval
double grtt_interval_min; // minimum GRTT update interval double grtt_interval_min; // minimum GRTT update interval
double grtt_interval_max; // maximum GRTT update interval double grtt_interval_max; // maximum GRTT update interval
double grtt_max; double grtt_max;
unsigned int grtt_decrease_delay_count; unsigned int grtt_decrease_delay_count;
bool grtt_response; bool grtt_response;
double grtt_current_peak; double grtt_current_peak;
double grtt_measured; double grtt_measured;
double grtt_age; double grtt_age;
double grtt_advertised; double grtt_advertised;
UINT8 grtt_quantized; UINT8 grtt_quantized;
double gsize_measured; double gsize_measured;
double gsize_advertised; double gsize_advertised;
UINT8 gsize_quantized; UINT8 gsize_quantized;
// Server congestion control parameters // Server congestion control parameters
bool cc_enable; bool cc_enable;
UINT8 cc_sequence; UINT8 cc_sequence;
NormNodeList cc_node_list; NormNodeList cc_node_list;
bool cc_slow_start; bool cc_slow_start;
double sent_rate; // measured sent rate double sent_rate; // measured sent rate
struct timeval prev_update_time; // for sent_rate measurement struct timeval prev_update_time; // for sent_rate measurement
unsigned long sent_accumulator; // for sent_rate measurement unsigned long sent_accumulator; // for sent_rate measurement
double nominal_packet_size; double nominal_packet_size;
// Client parameters // Client parameters
bool is_client; bool is_client;
NormNodeTree server_tree; NormNodeTree server_tree;
unsigned long remote_server_buffer_size; unsigned long remote_server_buffer_size;
bool unicast_nacks; bool unicast_nacks;
bool client_silent; bool client_silent;
NormServerNode::RepairBoundary default_repair_boundary;
NormObject::NackingMode default_nacking_mode;
// Protocol test/debug parameters // Protocol test/debug parameters
bool trace; bool trace;
double tx_loss_rate; // for correlated loss double tx_loss_rate; // for correlated loss
double rx_loss_rate; // for uncorrelated loss double rx_loss_rate; // for uncorrelated loss
// Linkers // Linkers
NormSession* next; NormSession* next;
}; // end class NormSession }; // end class NormSession

2
common/normSimAgent.cpp
View File

@ -831,7 +831,7 @@ void NormSimAgent::Notify(NormController::Event event,
break; break;
} // end switch (object->GetType()) } // end switch (object->GetType())
break; break;
case RX_OBJECT_COMPLETE: case RX_OBJECT_COMPLETED:
{ {
switch(object->GetType()) switch(object->GetType())
{ {

231
common/normTest.cpp Normal file
View File

@ -0,0 +1,231 @@
////////////////////////////////////////////////////
// This is a test application for experimenting
// with the NORM API implementation during its
// development. A better-documented and complete
// example of the NORM API usage will be provided
// when the NORM API is more complete.
#include "normApi.h"
#include "protokit.h" // for protolib debug, stuff, etc
#include <stdio.h>
#ifdef UNIX
#include <unistd.h> // for "sleep()"
#endif // UNIX
int main(int argc, char* argv[])
{
printf("normTest starting ...\n");
SetDebugLevel(4);
NormInstanceHandle instance = NormCreateInstance();
#ifdef WIN32
const char* cachePath = "C:\\Adamson\\Temp\\cache\\";
#else
const char* cachePath = "/tmp/";
#endif // if/else WIN32/UNIX
NormSetCacheDirectory(instance, cachePath);
NormSessionHandle session = NormCreateSession(instance,
"224.1.2.3",
6003,
NORM_NODE_ANY);
NormSetMessageTrace(session, true);
NormSetTxLoss(session, 10.0); // 1% packet loss
NormSetGrttEstimate(session, 0.1);//0.001); // 1 msec initial grtt
NormSetTransmitRate(session, 1.0e+05); // in bits/second
NormSetDefaultRepairBoundary(session, NORM_BOUNDARY_BLOCK);
// Uncomment to receive own traffic
NormSetLoopback(session, true);
// Uncomment this line to participate as a receiver
NormStartReceiver(session, 1024*1024);
// Uncomment the following line to start sender
NormStartSender(session, 1024*1024, 1024, 64, 0);
NormAddAckingNode(session, NormGetLocalNodeId(session));
NormObjectHandle stream = NORM_OBJECT_INVALID;
const char* filePath = "/home/adamson/images/art/giger/giger205.jpg";
const char* fileName = "ferrari.jpg";
// Uncomment this line to send a stream instead of the file
stream = NormOpenStream(session, 1024*1024);
NormSetStreamFlushMode(stream, NORM_FLUSH_NONE);
int index = -1; // used to monitor reliable stream reception
int sendCount = 0;
int sendMax = 200;
NormEvent theEvent;
while (NormGetNextEvent(instance, &theEvent))
{
switch (theEvent.type)
{
case NORM_TX_QUEUE_EMPTY:
if (NORM_OBJECT_INVALID != stream)
{
// Write a message to the "stream"
char buffer[1024];
sprintf(buffer, "normTest says hello %d ...\n", sendCount++);
unsigned int len = strlen(buffer);
TRACE("writing to stream ...\n");
if (len != NormWriteStream(stream, buffer, len))
TRACE("incomplete write:%u\n", len);
else
{
NormFlushStream(stream);
NormSetWatermark(session, stream);
}
}
else
{
NormObjectHandle txFile =
NormQueueFile(session,
filePath,
fileName,
strlen(fileName));
// Repeatedly queue our file for sending
if (NORM_OBJECT_INVALID == txFile)
{
DMSG(0, "normTest: error queuing file: %s\n", filePath);
break;
}
else
{
TRACE("QUEUED FILE ...\n");
NormSetWatermark(session, txFile);
}
sendCount++;
}
break;
case NORM_TX_OBJECT_PURGED:
DMSG(3, "normTest: NORM_TX_OBJECT_PURGED event ...\n");
break;
case NORM_RX_OBJECT_NEW:
DMSG(3, "normTest: NORM_RX_OBJECT_NEW event ...\n");
break;
case NORM_RX_OBJECT_INFO:
// Assume info contains '/' delimited <path/fileName> string
if (NORM_OBJECT_FILE == NormGetObjectType(theEvent.object))
{
NormObjectTransportId id = NormGetObjectTransportId(theEvent.object);
if (0 != (id & 0x01))
{
//NormCancelObject(theEvent.object);
//break;
}
char fileName[PATH_MAX];
strcpy(fileName, cachePath);
int pathLen = strlen(fileName);
unsigned short nameLen = PATH_MAX - pathLen;
NormGetObjectInfo(theEvent.object, fileName+pathLen, &nameLen);
fileName[nameLen + pathLen] = '\0';
char* ptr = fileName + 5;
while ('\0' != *ptr)
{
if ('/' == *ptr) *ptr = PROTO_PATH_DELIMITER;
ptr++;
}
if (!NormSetFileName(theEvent.object, fileName))
TRACE("normTest: NormSetFileName(%s) error\n", fileName);
DMSG(3, "normTest: recv'd info for file: %s\n", fileName);
}
break;
case NORM_RX_OBJECT_UPDATE:
{
//TRACE("normTest: NORM_RX_OBJECT_UPDATE event ...\n");
if (NORM_OBJECT_STREAM != NormGetObjectType(theEvent.object))
break;
char buffer[1024];
unsigned int len = 1023;
do
{
len = 1023;
if (NormReadStream(theEvent.object, buffer, &len))
{
buffer[len] = '\0';
if (len)
{
TRACE("normTest: recvd(%u):\n\"%s\"\n", len, buffer);
// This while() loop is cheesy test, looking for broken stream
//
char* ptr = buffer;
while ((ptr = strstr(ptr, "hello")))
{
int value;
if (1 == sscanf(ptr, "hello %d", &value))
{
if (index >= 0)
{
if (1 != (value - index))
TRACE("WARNING! possible break? value:%d index:%d\n",
value, index);
}
index = value;
}
else
{
TRACE("couldn't find index\n");
}
ptr += 5;
}
}
}
else
{
TRACE("normTest: error reading stream\n");
break;
}
} while (0 != len);
break;
}
case NORM_RX_OBJECT_COMPLETED:
TRACE("normTest: NORM_RX_OBJECT_COMPLETED event ...\n");
break;
case NORM_RX_OBJECT_ABORTED:
TRACE("normTest: NORM_RX_OBJECT_ABORTED event ...\n");
break;
default:
TRACE("Got event type: %d\n", theEvent.type);
} // end switch(theEvent.type)
// Break after sending "sendMax" messages or files
if (sendCount >= sendMax) break;
} // end while (NormGetNextEvent())
fprintf(stderr, "normTest shutting down in 30 sec ...\n");
#ifdef WIN32
Sleep(30000);
#else
sleep(30); // allows time for cleanup if we're sending to someone else
#endif // if/else WIN32/UNIX
NormCloseStream(stream);
NormStopReceiver(session);
NormStopSender(session);
NormDestroySession(session);
NormDestroyInstance(instance);
fprintf(stderr, "normTest: Done.\n");
return 0;
} // end main()

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.2b1" #define VERSION "1.2b2"
#endif // _NORM_VERSION #endif // _NORM_VERSION

4
common/raft.cpp
View File

@ -61,7 +61,7 @@ class RaftApp : public ProtoApp
const char* const RaftApp::CMD_LIST[] = const char* const RaftApp::CMD_LIST[] =
{ {
"+debug", // debug <level> "+debug", // debug <level>
"+listen", // recv [<mcastAddr>/]<port> "+listen", // recv [<mcastAddr>/]<port>
"+dest", // send <addr>/<port> "+dest", // send <addr>/<port>
"+rtspProxy", // rtsp <rtspUrl> "+rtspProxy", // rtsp <rtspUrl>
NULL NULL
@ -467,7 +467,7 @@ void RaftApp::OnClientSocketEvent(ProtoSocket& /*theSocket*/,
while (rtsp_client_socket.Recv(buffer, buflen)) while (rtsp_client_socket.Recv(buffer, buflen))
{ {
buffer[buflen] = '\0'; buffer[buflen] = '\0';
TRACE("%s"); TRACE("%s", buffer);
} }
TRACE("\n"); TRACE("\n");
break; break;

2
ns/example.tcl
View File

@ -82,7 +82,7 @@ for {set i 2} {$i <= $numNodes} {incr i} {
} }
$ns_ at 0.0 "$norm(1) sendFile 64000" $ns_ at 0.0 "$norm(1) sendFile 64000"
$ns_ at 3000.0 "finish $ns_ $f $nf" $ns_ at 300.0 "finish $ns_ $f $nf"
proc finish {ns_ f nf} { proc finish {ns_ f nf} {
$ns_ flush-trace $ns_ flush-trace

12
unix/Makefile.common
View File

@ -13,7 +13,7 @@ NS = ../ns
INCLUDES = $(TCL_INCL_PATH) $(SYSTEM_INCLUDES) -I$(UNIX) -I$(COMMON) -I$(PROTOLIB)/common INCLUDES = $(TCL_INCL_PATH) $(SYSTEM_INCLUDES) -I$(UNIX) -I$(COMMON) -I$(PROTOLIB)/common
CFLAGS = -DPROTO_DEBUG -DUNIX -O -fPIC $(SYSTEM_HAVES) $(INCLUDES) CFLAGS = -g -DPROTO_DEBUG -DUNIX -D_FILE_OFFSET_BITS=64 -O -fPIC $(SYSTEM_HAVES) $(INCLUDES)
LDFLAGS = $(SYSTEM_LDFLAGS) LDFLAGS = $(SYSTEM_LDFLAGS)
@ -83,7 +83,15 @@ APP_OBJ = $(APP_SRC:.cpp=.o)
norm: $(APP_OBJ) libnorm.a $(LIBPROTO) norm: $(APP_OBJ) libnorm.a $(LIBPROTO)
$(CC) $(CFLAGS) -o $@ $(APP_OBJ) $(LDFLAGS) libnorm.a $(LIBPROTO) $(LIBS) $(CC) $(CFLAGS) -o $@ $(APP_OBJ) $(LDFLAGS) libnorm.a $(LIBPROTO) $(LIBS)
# (normTest) test of NORM API
TEST_SRC = $(COMMON)/normApi.cpp $(COMMON)/normTest.cpp
TEST_OBJ = $(TEST_SRC:.cpp=.o)
normTest: $(TEST_OBJ) libnorm.a $(LIBPROTO)
$(CC) $(CFLAGS) -o $@ $(TEST_OBJ) $(LDFLAGS) libnorm.a $(LIBPROTO) $(LIBS)
# (raft) command-line reliable tunnel helper # (raft) command-line reliable tunnel helper
RAFT_SRC = $(COMMON)/raft.cpp RAFT_SRC = $(COMMON)/raft.cpp
RAFT_OBJ = $(RAFT_SRC:.cpp=.o) RAFT_OBJ = $(RAFT_SRC:.cpp=.o)

4
unix/Makefile.linux
View File

@ -62,8 +62,8 @@ SYSTEM_LIBS = -ldl
# (We export these for other Makefiles as needed) # (We export these for other Makefiles as needed)
# #
SYSTEM_HAVES = -DLINUX -DHAVE_IPV6 -DHAVE_GETLOGIN -DHAVE_LOCKF \ SYSTEM_HAVES = -DLINUX -DHAVE_IPV6 -DHAVE_GETLOGIN -D_FILE_OFFSET_BITS=64 -DHAVE_LOCKF \
-DHAVE_OLD_SIGNALHANDLER -DHAVE_DIRFD -DHAVE_ASSERT $(NETSEC) -DHAVE_OLD_SIGNALHANDLER -DHAVE_DIRFD -DHAVE_ASSERT
SYSTEM_SRC = SYSTEM_SRC =

15
unix/Makefile.macosx
View File

@ -6,9 +6,9 @@
# (Where to find X11 libraries, etc) # (Where to find X11 libraries, etc)
# #
SYSTEM_INCLUDES = -I/usr/X11R6/include SYSTEM_INCLUDES =
SYSTEM_LDFLAGS = -L/usr/X11R6/lib SYSTEM_LDFLAGS =
SYSTEM_LIBS = SYSTEM_LIBS = -lresolv
# 2) System specific capabilities # 2) System specific capabilities
# Must choose appropriate for the following: # Must choose appropriate for the following:
@ -36,15 +36,16 @@ SYSTEM_LIBS =
# (We export these for other Makefiles as needed) # (We export these for other Makefiles as needed)
# #
export SYSTEM_HAVES = -DMACOSX -DHAVE_IPV6 -DHAVE_ASSERT -DHAVE_GETLOGIN -DHAVE_FLOCK -DHAVE_DIRFD $(DNETSEC) -DSOCKLEN_T=int SYSTEM_HAVES = -DMACOSX -DHAVE_IPV6 -DHAVE_ASSERT -DHAVE_GETLOGIN -DHAVE_FLOCK \
-D_FILE_OFFSET_BITS=64 -DHAVE_DIRFD -DSOCKLEN_T=int
SYSTEM_SRC = bsdRouteMgr.cpp SYSTEM_SRC = bsdRouteMgr.cpp
# The "SYSTEM" keyword can be used for dependent makes # The "SYSTEM" keyword can be used for dependent makes
SYSTEM = macosx SYSTEM = macosx
export CC = g++ CC = g++
export RANLIB = ranlib RANLIB = ranlib
export AR = ar AR = ar
include Makefile.common include Makefile.common

7
unix/Makefile.solaris
View File

@ -5,8 +5,8 @@
# 1) System specific additional libraries, include paths, etc # 1) System specific additional libraries, include paths, etc
# (Where to find X11 libraries, etc) # (Where to find X11 libraries, etc)
# #
SYSTEM_INCLUDES = -I/usr/openwin/include SYSTEM_INCLUDES =
SYSTEM_LDFLAGS = -L/usr/openwin/lib -R/usr/openwin/lib SYSTEM_LDFLAGS =
SYSTEM_LIBS = -ldl -lnsl -lsocket -lresolv SYSTEM_LIBS = -ldl -lnsl -lsocket -lresolv
# 2) System specific capabilities # 2) System specific capabilities
@ -35,7 +35,8 @@ SYSTEM_LIBS = -ldl -lnsl -lsocket -lresolv
# (We export these for other Makefiles as needed) # (We export these for other Makefiles as needed)
# #
SYSTEM_HAVES = -DHAVE_IPV6 -DHAVE_ASSERT -DHAVE_CUSERID -DHAVE_LOCKF $(NETSEC) -DSOLARIS SYSTEM_HAVES = -DHAVE_IPV6 -DHAVE_ASSERT -DHAVE_CUSERID -DHAVE_LOCKF \
-D_FILE_OFFSET_BITS=64 -D_LARGEFILE_SOURCE -DSOLARIS
SYSTEM = solaris SYSTEM = solaris

4
unix/unixPostProcess.cpp
View File

@ -234,6 +234,10 @@ bool UnixPostProcessor::ProcessFile(const char* path)
// Restore signal handlers for parent // Restore signal handlers for parent
signal(SIGTERM, sigtermHandler); signal(SIGTERM, sigtermHandler);
signal(SIGINT, sigintHandler); signal(SIGINT, sigintHandler);
// The use of "waitpid()" here is a work-around
// for an IRIX SIGCHLD issue
int status;
while (waitpid(-1, &status, WNOHANG) > 0);
signal(SIGCHLD, sigchldHandler); signal(SIGCHLD, sigchldHandler);
break; break;
} }

18
win32/Norm.dsw
View File

@ -45,6 +45,24 @@ Package=<4>
############################################################################### ###############################################################################
Project: "normTest"=..\normTest.dsp - Package Owner=<4>
Package=<5>
{{{
}}}
Package=<4>
{{{
Begin Project Dependency
Project_Dep_Name NormLib
End Project Dependency
Begin Project Dependency
Project_Dep_Name Protokit
End Project Dependency
}}}
###############################################################################
Global: Global:
Package=<5> Package=<5>

BIN
win32/Norm.opt
View File

Binary file not shown.

106
win32/normTest.dsp Normal file
View File

@ -0,0 +1,106 @@
# Microsoft Developer Studio Project File - Name="normTest" - Package Owner=<4>
# Microsoft Developer Studio Generated Build File, Format Version 6.00
# ** DO NOT EDIT **
# TARGTYPE "Win32 (x86) Console Application" 0x0103
CFG=normTest - Win32 Debug
!MESSAGE This is not a valid makefile. To build this project using NMAKE,
!MESSAGE use the Export Makefile command and run
!MESSAGE
!MESSAGE NMAKE /f "normTest.mak".
!MESSAGE
!MESSAGE You can specify a configuration when running NMAKE
!MESSAGE by defining the macro CFG on the command line. For example:
!MESSAGE
!MESSAGE NMAKE /f "normTest.mak" CFG="normTest - Win32 Debug"
!MESSAGE
!MESSAGE Possible choices for configuration are:
!MESSAGE
!MESSAGE "normTest - Win32 Release" (based on "Win32 (x86) Console Application")
!MESSAGE "normTest - Win32 Debug" (based on "Win32 (x86) Console Application")
!MESSAGE
# Begin Project
# PROP AllowPerConfigDependencies 0
# PROP Scc_ProjName ""
# PROP Scc_LocalPath ""
CPP=cl.exe
RSC=rc.exe
!IF "$(CFG)" == "normTest - Win32 Release"
# PROP BASE Use_MFC 0
# PROP BASE Use_Debug_Libraries 0
# PROP BASE Output_Dir "Release"
# PROP BASE Intermediate_Dir "Release"
# PROP BASE Target_Dir ""
# PROP Use_MFC 0
# PROP Use_Debug_Libraries 0
# PROP Output_Dir "Release"
# PROP Intermediate_Dir "Release"
# PROP Ignore_Export_Lib 0
# PROP Target_Dir ""
# ADD BASE CPP /nologo /W3 /GX /O2 /D "WIN32" /D "NDEBUG" /D "_CONSOLE" /D "_MBCS" /YX /FD /c
# ADD CPP /nologo /MD /W3 /GX /O2 /I "..\common" /I "..\protolib\common" /I "..\protolib\win32" /D "NDEBUG" /D "PROTO_DEBUG" /D "HAVE_IPV6" /D "HAVE_ASSERT" /D "WIN32" /D "_CONSOLE" /D "_MBCS" /YX /FD /c
# ADD BASE RSC /l 0x409 /d "NDEBUG"
# ADD RSC /l 0x409 /d "NDEBUG"
BSC32=bscmake.exe
# ADD BASE BSC32 /nologo
# ADD BSC32 /nologo
LINK32=link.exe
# ADD BASE LINK32 kernel32.lib user32.lib gdi32.lib winspool.lib comdlg32.lib advapi32.lib shell32.lib ole32.lib oleaut32.lib uuid.lib odbc32.lib odbccp32.lib kernel32.lib user32.lib gdi32.lib winspool.lib comdlg32.lib advapi32.lib shell32.lib ole32.lib oleaut32.lib uuid.lib odbc32.lib odbccp32.lib /nologo /subsystem:console /machine:I386
# ADD LINK32 iphlpapi.lib ws2_32.lib kernel32.lib user32.lib gdi32.lib winspool.lib comdlg32.lib advapi32.lib shell32.lib ole32.lib oleaut32.lib uuid.lib odbc32.lib odbccp32.lib /nologo /subsystem:console /machine:I386
!ELSEIF "$(CFG)" == "normTest - Win32 Debug"
# PROP BASE Use_MFC 0
# PROP BASE Use_Debug_Libraries 1
# PROP BASE Output_Dir "Debug"
# PROP BASE Intermediate_Dir "Debug"
# PROP BASE Target_Dir ""
# PROP Use_MFC 0
# PROP Use_Debug_Libraries 1
# PROP Output_Dir "Debug"
# PROP Intermediate_Dir "Debug"
# PROP Ignore_Export_Lib 0
# PROP Target_Dir ""
# ADD BASE CPP /nologo /W3 /Gm /GX /ZI /Od /D "WIN32" /D "_DEBUG" /D "_CONSOLE" /D "_MBCS" /YX /FD /GZ /c
# ADD CPP /nologo /MDd /W3 /GX /Od /I "..\common" /I "..\protolib\common" /I "..\protolib\win32" /D "_DEBUG" /D "PROTO_DEBUG" /D "HAVE_IPV6" /D "HAVE_ASSERT" /D "WIN32" /D "_CONSOLE" /D "_MBCS" /YX /FD /GZ /c
# ADD BASE RSC /l 0x409 /d "_DEBUG"
# ADD RSC /l 0x409 /d "_DEBUG"
BSC32=bscmake.exe
# ADD BASE BSC32 /nologo
# ADD BSC32 /nologo
LINK32=link.exe
# ADD BASE LINK32 kernel32.lib user32.lib gdi32.lib winspool.lib comdlg32.lib advapi32.lib shell32.lib ole32.lib oleaut32.lib uuid.lib odbc32.lib odbccp32.lib kernel32.lib user32.lib gdi32.lib winspool.lib comdlg32.lib advapi32.lib shell32.lib ole32.lib oleaut32.lib uuid.lib odbc32.lib odbccp32.lib /nologo /subsystem:console /debug /machine:I386 /pdbtype:sept
# ADD LINK32 iphlpapi.lib ws2_32.lib kernel32.lib user32.lib gdi32.lib winspool.lib comdlg32.lib advapi32.lib shell32.lib ole32.lib oleaut32.lib uuid.lib odbc32.lib odbccp32.lib /nologo /subsystem:console /debug /machine:I386 /pdbtype:sept
!ENDIF
# Begin Target
# Name "normTest - Win32 Release"
# Name "normTest - Win32 Debug"
# Begin Group "Source Files"
# PROP Default_Filter "cpp;c;cxx;rc;def;r;odl;idl;hpj;bat"
# Begin Source File
SOURCE=..\common\normApi.cpp
# End Source File
# Begin Source File
SOURCE=..\common\normTest.cpp
# End Source File
# End Group
# Begin Group "Header Files"
# PROP Default_Filter "h;hpp;hxx;hm;inl"
# End Group
# Begin Group "Resource Files"
# PROP Default_Filter "ico;cur;bmp;dlg;rc2;rct;bin;rgs;gif;jpg;jpeg;jpe"
# End Group
# End Target
# End Project