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

v1.2b5

pull/2/head
Jeff Weston 2019-09-11 11:42:54 -04:00
parent f451a006f6
commit 700519ec12
18 changed files with 3895 additions and 638 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

2885
NormDeveloperGuide.html Normal file
View File

File diff suppressed because it is too large Load Diff

BIN
NormDeveloperGuide.pdf Normal file
View File

Binary file not shown.

8
VERSION.TXT
View File

@ -1,9 +1,14 @@
NORM Version History
Version 1.2b5
=============
- Initial release with theoretically-working API
(including "Norm Developer's Guide!")
Version 1.2b4
=============
- Finally remembered to update this file!
- Added "help" command to "norm" demo app (Thanks Marinho Barcellos)
- Added "help" command to "norm" demo app (Thanks Marinho Barcellos!)
- New rxbuffer mgmnt scheme to better use limited buffer space.
- Removed explicit rate limit on receiver feedback messages
- API more complete and established better API naming conventions
@ -12,6 +17,7 @@ Version 1.2b4
- NORM RFC 3940 was released and this version is in compliance.
- Added optional positive acknowledgements for sender-defined
"watermarks".
- Fixed multicast leave problem when interfaceName was used for join
Version 1.1b9
=============

945
common/normApi.cpp
View File

File diff suppressed because it is too large Load Diff

384
common/normApi.h
View File

@ -6,6 +6,8 @@
#include <windows.h>
#endif // WIN32
#include <sys/types.h> // for "off_t" type
////////////////////////////////////////////////////////////
// IMPORTANT NOTICE
// The NORM API is _very_ much in a developmental phase
@ -18,106 +20,27 @@
// is removed, the API shouldn't be considered final.
/** NORM API Data Types and Constants
* These are data types and constants defined
* for the NORM API
*/
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 const void* NormNodeHandle;
extern const NormNodeHandle NORM_NODE_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
};
void NormSetDefaultNackingMode(NormSessionHandle sessionHandle,
NormNackingMode nackingMode);
enum NormRepairBoundary
{
NORM_BOUNDARY_BLOCK,
NORM_BOUNDARY_OBJECT
};
void NormSetDefaultRepairBoundary(NormSessionHandle sessionHandle,
NormRepairBoundary repairBoundary);
NormRepairBoundary NormNodeGetRepairBoundary(NormNodeHandle nodeHandle);
void NormNodeSetRepairBoundary(NormNodeHandle nodeHandle,
NormRepairBoundary repairBoundary);
NormNodeId NormNodeGetId(NormNodeHandle nodeHandle);
void NormNodeSetNackingMode(NormNodeHandle nodeHandle,
NormNackingMode nackingMode);
// General NormObject functions
enum NormObjectType
{
NORM_OBJECT_NONE,
@ -126,44 +49,6 @@ enum NormObjectType
NORM_OBJECT_STREAM
};
NormObjectType NormObjectGetType(NormObjectHandle objectHandle);
bool NormObjectGetInfo(NormObjectHandle objectHandle,
char* infoBuffer,
unsigned short* infoLen);
NormObjectTransportId NormObjectGetTransportId(NormObjectHandle objectHandle);
void NormObjectCancel(NormObjectHandle objectHandle);
void NormObjectReNormObjectRetaintain(NormObjectHandle objectHandle);
void NormObjectRelease(NormObjectHandle objectHandle);
// Receiver-only NormObject functions
NormNackingMode NormObjectGetNackingMode(NormObjectHandle objectHandle);
void NormObjectSetNackingMode(NormObjectHandle objectHandle,
NormNackingMode nackingMode);
// Sender-only NormObject functions
bool NormSetWatermark(NormSessionHandle sessionHandle,
NormObjectHandle objectHandle);
// NormStreamObject functions
NormObjectHandle NormStreamOpen(NormSessionHandle sessionHandle,
unsigned long bufferSize);
void NormStreamClose(NormObjectHandle streamHandle);
enum NormFlushMode
{
NORM_FLUSH_NONE,
@ -171,49 +56,19 @@ enum NormFlushMode
NORM_FLUSH_ACTIVE
};
void NormStreamSetFlushMode(NormObjectHandle streamHandle,
NormFlushMode flushMode);
void NormStreamSetPushMode(NormObjectHandle streamHandle,
bool state);
enum NormNackingMode
{
NORM_NACK_NONE,
NORM_NACK_INFO_ONLY,
NORM_NACK_NORMAL
};
unsigned int NormStreamWrite(NormObjectHandle streamHandle,
const char* buffer,
unsigned int numBytes);
void NormStreamFlush(NormObjectHandle streamHandle, bool eom = false);
void NormStreamMarkEom(NormObjectHandle streamHandle);
bool NormStreamRead(NormObjectHandle streamHandle,
char* buffer,
unsigned int* numBytes);
bool NormStreamSeekMsgStart(NormObjectHandle streamHandle);
// NormFileObject Functions
NormObjectHandle NormFileEnqueue(NormSessionHandle sessionHandle,
const char* fileName,
const char* infoPtr = (const char*)0,
unsigned int infoLen = 0);
bool NormSetCacheDirectory(NormInstanceHandle instanceHandle,
const char* cachePath);
bool NormFileGetName(NormObjectHandle fileHandle,
char* nameBuffer,
unsigned int bufferLen);
bool NormFileRename(NormObjectHandle fileHandle,
const char* fileName);
// NormDataObject Functions
bool NormDataEnqueue(const char* dataPtr,
unsigned long dataLen,
const char* infoPtr = (const char*)0,
unsigned int infoLen = 0);
// NORM Event Notification Routines
enum NormRepairBoundary
{
NORM_BOUNDARY_BLOCK,
NORM_BOUNDARY_OBJECT
};
enum NormEventType
{
@ -224,11 +79,12 @@ enum NormEventType
NORM_TX_OBJECT_PURGED,
NORM_LOCAL_SERVER_CLOSED,
NORM_REMOTE_SERVER_NEW,
NORM_REMOTE_SERVER_INACTIVE,
NORM_REMOTE_SERVER_ACTIVE,
NORM_REMOTE_SERVER_INACTIVE,
NORM_REMOTE_SERVER_PURGED,
NORM_RX_OBJECT_NEW,
NORM_RX_OBJECT_INFO,
NORM_RX_OBJECT_UPDATE,
NORM_RX_OBJECT_UPDATED,
NORM_RX_OBJECT_COMPLETED,
NORM_RX_OBJECT_ABORTED
};
@ -242,6 +98,16 @@ typedef struct
} NormEvent;
/** NORM API General Initialization and Operation Functions */
NormInstanceHandle NormCreateInstance();
void NormDestroyInstance(NormInstanceHandle instanceHandle);
// This MUST be set to enable NORM_OBJECT_FILE reception!
bool NormSetCacheDirectory(NormInstanceHandle instanceHandle,
const char* cachePath);
// This call blocks until the next NormEvent is ready unless asynchronous
// notification is used (see below)
bool NormGetNextEvent(NormInstanceHandle instanceHandle, NormEvent* theEvent);
@ -252,7 +118,7 @@ bool NormGetNextEvent(NormInstanceHandle instanceHandle, NormEvent* theEvent);
// 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.
// not block when the notification is posted.
#ifdef WIN32
typedef HANDLE NormDescriptor;
@ -261,4 +127,184 @@ typedef int NormDescriptor;
#endif // if/else WIN32/UNIX
NormDescriptor NormGetDescriptor(NormInstanceHandle instanceHandle);
/** NORM Session Creation and Control Functions */
NormSessionHandle NormCreateSession(NormInstanceHandle instanceHandle,
const char* sessionAddress,
unsigned short sessionPort,
NormNodeId localNodeId);
void NormDestroySession(NormSessionHandle sessionHandle);
NormNodeId NormGetLocalNodeId(NormSessionHandle sessionHandle);
bool NormSetMulticastInterface(NormSessionHandle sessionHandle,
const char* interfaceName);
bool NormSetTTL(NormSessionHandle sessionHandle,
unsigned char ttl);
bool NormSetTOS(NormSessionHandle sessionHandle,
unsigned char tos);
bool NormSetLoopback(NormSessionHandle sessionHandle,
bool loopback);
// Special functions for debug support
void NormSetMessageTrace(NormSessionHandle sessionHandle, bool state);
void NormSetTxLoss(NormSessionHandle sessionHandle, double percent);
void NormSetRxLoss(NormSessionHandle sessionHandle, double percent);
/** NORM Sender Functions */
bool NormStartSender(NormSessionHandle sessionHandle,
unsigned long bufferSpace,
unsigned short segmentSize,
unsigned char numData,
unsigned char numParity);
void NormStopSender(NormSessionHandle sessionHandle);
void NormSetTransmitRate(NormSessionHandle sessionHandle,
double bitsPerSecond);
void NormSetCongestionControl(NormSessionHandle sessionHandle,
bool state);
void NormSetTransmitRateBounds(NormSessionHandle sessionHandle,
double rateMin,
double rateMax);
void NormSetAutoParity(NormSessionHandle sessionHandle,
unsigned char autoParity);
void NormSetGrttEstimate(NormSessionHandle sessionHandle,
double grttEstimate);
bool NormAddAckingNode(NormSessionHandle sessionHandle,
NormNodeId nodeId);
void NormRemoveAckingNode(NormSessionHandle sessionHandle,
NormNodeId nodeId);
NormObjectHandle NormFileEnqueue(NormSessionHandle sessionHandle,
const char* fileName,
const char* infoPtr = (const char*)0,
unsigned int infoLen = 0);
bool NormDataEnqueue(const char* dataPtr,
unsigned long dataLen,
const char* infoPtr = (const char*)0,
unsigned int infoLen = 0);
NormObjectHandle NormStreamOpen(NormSessionHandle sessionHandle,
unsigned long bufferSize);
void NormStreamClose(NormObjectHandle streamHandle);
unsigned int NormStreamWrite(NormObjectHandle streamHandle,
const char* buffer,
unsigned int numBytes);
void NormStreamFlush(NormObjectHandle streamHandle,
bool eom = false,
NormFlushMode flushMode = NORM_FLUSH_PASSIVE);
void NormStreamSetAutoFlush(NormObjectHandle streamHandle,
NormFlushMode flushMode);
void NormStreamSetPushEnable(NormObjectHandle streamHandle,
bool pushEnable);
void NormStreamMarkEom(NormObjectHandle streamHandle);
bool NormSetWatermark(NormSessionHandle sessionHandle,
NormObjectHandle objectHandle);
/** NORM Receiver Functions */
bool NormStartReceiver(NormSessionHandle sessionHandle,
unsigned long bufferSpace);
void NormStopReceiver(NormSessionHandle sessionHandle);
void NormSetSilentReceiver(NormSessionHandle sessionHandle,
bool silent);
void NormSetDefaultUnicastNack(NormSessionHandle sessionHandle,
bool unicastNacks);
void NormNodeSetUnicastNack(NormNodeHandle remoteSender,
bool unicastNacks);
void NormSetDefaultNackingMode(NormSessionHandle sessionHandle,
NormNackingMode nackingMode);
void NormNodeSetNackingMode(NormNodeHandle nodeHandle,
NormNackingMode nackingMode);
void NormObjectSetNackingMode(NormObjectHandle objectHandle,
NormNackingMode nackingMode);
void NormSetDefaultRepairBoundary(NormSessionHandle sessionHandle,
NormRepairBoundary repairBoundary);
void NormNodeSetRepairBoundary(NormNodeHandle nodeHandle,
NormRepairBoundary repairBoundary);
bool NormStreamRead(NormObjectHandle streamHandle,
char* buffer,
unsigned int* numBytes);
bool NormStreamSeekMsgStart(NormObjectHandle streamHandle);
unsigned long NormStreamGetReadOffset(NormObjectHandle streamHandle);
/** NORM Object Functions */
NormObjectType NormObjectGetType(NormObjectHandle objectHandle);
bool NormObjectHasInfo(NormObjectHandle objectHandle);
unsigned short NormObjectGetInfoLength(NormObjectHandle objectHandle);
unsigned short NormObjectGetInfo(NormObjectHandle objectHandle,
char* buffer,
unsigned short bufferLen);
off_t NormObjectGetSize(NormObjectHandle objectHandle);
void NormObjectCancel(NormObjectHandle objectHandle);
void NormObjectRetain(NormObjectHandle objectHandle);
void NormObjectRelease(NormObjectHandle objectHandle);
bool NormFileGetName(NormObjectHandle fileHandle,
char* nameBuffer,
unsigned int bufferLen);
bool NormFileRename(NormObjectHandle fileHandle,
const char* fileName);
const char*volatile NormDataAccessData(NormObjectHandle objectHandle);
char* NormDataDetachData(NormObjectHandle objectHandle);
NormNodeHandle NormObjectGetSender(NormObjectHandle objectHandle);
/** NORM Node Functions */
NormNodeId NormNodeGetId(NormNodeHandle nodeHandle);
void NormNodeRetain(NormNodeHandle nodeHandle);
void NormNodeRelease(NormNodeHandle nodeHandle);
#endif // _NORM_API

10
common/normApp.cpp
View File

@ -761,7 +761,7 @@ bool NormApp::OnCommand(const char* cmd, const char* val)
else if (!strncmp("unicastNacks", cmd, len))
{
unicast_nacks = true;
if (session) session->SetUnicastNacks(true);
if (session) session->ClientSetUnicastNacks(true);
}
else if (!strncmp("silentClient", cmd, len))
{
@ -1198,8 +1198,8 @@ void NormApp::Notify(NormController::Event event,
} // end switch(object->GetType())
break;
case RX_OBJECT_UPDATE:
//DMSG(0, "NormApp::Notify(RX_OBJECT_UPDATE) ...\n");
case RX_OBJECT_UPDATED:
//DMSG(0, "NormApp::Notify(RX_OBJECT_UPDATED) ...\n");
switch (object->GetType())
{
case NormObject::FILE:
@ -1212,7 +1212,7 @@ void NormApp::Notify(NormController::Event event,
{
if (object != rx_stream)
{
DMSG(0, "NormApp::Notify(RX_OBJECT_UPDATE) update for invalid stream\n");
DMSG(0, "NormApp::Notify(RX_OBJECT_UPDATED) update for invalid stream\n");
break;
}
// Read the stream when it's updated
@ -1532,7 +1532,7 @@ bool NormApp::OnStartup(int argc, const char*const* argv)
if (output || rx_cache_path)
{
// StartClient(bufferMax (per-sender))
session->SetUnicastNacks(unicast_nacks);
session->ClientSetUnicastNacks(unicast_nacks);
session->ClientSetSilent(silent_client);
if (!session->StartClient(rx_buffer_size, interface_name))
{

7
common/normMessage.h
View File

@ -8,6 +8,7 @@
#include <string.h> // for memcpy(), etc
#include <math.h>
#include <stdlib.h> // for rand(), etc
#include <sys/types.h> // for off_t
#ifdef SIMULATE
#define SIM_PAYLOAD_MAX (36+8) // MGEN message size + StreamPayloadHeaderLen()
@ -94,7 +95,9 @@ inline double NormUnquantizeRate(unsigned short rate)
}
// This class is used to describe object "size" and/or "offset"
// (TBD) This hokey implementation should use "off_t"
// (TBD) This hokey implementation should use "off_t" as it's
// native storage format and get rid of this custom crap !!!
// (We should just get rid of this class !!!!!!!!)
class NormObjectSize
{
public:
@ -109,6 +112,7 @@ class NormObjectSize
size >>= 32;
ASSERT(0 == (size & 0xffff0000));
msb = size & 0x0000ffff;
msb = 0;
}
UINT16 MSB() const {return msb;}
@ -159,6 +163,7 @@ class NormObjectSize
NormObjectSize operator/(const NormObjectSize& b) const;
off_t GetOffset() const
{
off_t offset = msb;
offset <<= 32;
offset += lsb;

5
common/normNode.cpp
View File

@ -59,7 +59,7 @@ NormServerNode::NormServerNode(class NormSession& theSession, NormNodeId nodeId)
repair_boundary = session.ClientGetDefaultRepairBoundary();
default_nacking_mode = session.ClientGetDefaultNackingMode();
unicast_nacks = session.UnicastNacks();
unicast_nacks = session.ClientGetUnicastNacks();
// (TBD) get "max_pending_range" value from NormSession parameter
repair_timer.SetListener(this, &NormServerNode::OnRepairTimeout);
@ -1706,6 +1706,7 @@ void NormServerNode::Activate()
activity_timer.SetInterval(grtt_estimate*NORM_ROBUST_FACTOR);
session.ActivateTimer(activity_timer);
server_active = false;
session.Notify(NormController::REMOTE_SERVER_ACTIVE, this, NULL);
}
else
{
@ -1725,7 +1726,7 @@ bool NormServerNode::OnActivityTimeout(ProtoTimer& /*theTimer*/)
DMSG(0, "NormServerNode::OnActivityTimeout() node>%lu server>%lu gone inactive?\n",
LocalNodeId(), GetId());
FreeBuffers();
// (TBD) Notify application
session.Notify(NormController::REMOTE_SERVER_INACTIVE, this, NULL);
}
else
{

2
common/normNode.h
View File

@ -16,6 +16,8 @@ class NormNode
public:
NormNode(class NormSession& theSession, NormNodeId nodeId);
virtual ~NormNode();
NormSession& GetSession() const {return session;}
void Retain();
void Release();

93
common/normObject.cpp
View File

@ -11,7 +11,7 @@ NormObject::NormObject(NormObject::Type theType,
transport_id(transportId), segment_size(0), pending_info(false), repair_info(false),
current_block_id(0), next_segment_id(0),
max_pending_block(0), max_pending_segment(0),
info(NULL), info_len(0), accepted(false), notify_on_update(true)
info_ptr(NULL), info_len(0), accepted(false), notify_on_update(true)
{
if (theServer)
nacking_mode = theServer->GetDefaultNackingMode();
@ -22,10 +22,10 @@ NormObject::NormObject(NormObject::Type theType,
NormObject::~NormObject()
{
Close();
if (info)
if (info_ptr)
{
delete info;
info = NULL;
delete info_ptr;
info_ptr = NULL;
}
}
@ -73,7 +73,7 @@ bool NormObject::Open(const NormObjectSize& objectSize,
{
pending_info = true;
info_len = 0;
if (!(info = new char[segmentSize]))
if (!(info_ptr = new char[segmentSize]))
{
DMSG(0, "NormObject::Open() info allocation error\n");
return false;
@ -84,20 +84,20 @@ bool NormObject::Open(const NormObjectSize& objectSize,
{
if (infoPtr)
{
if (info) delete []info;
if (info_ptr) delete []info_ptr;
if (infoLen > segmentSize)
{
DMSG(0, "NormObject::Open() info too big error\n");
info_len = 0;
return false;
}
if (!(info = new char[infoLen]))
if (!(info_ptr = new char[infoLen]))
{
DMSG(0, "NormObject::Open() info allocation error\n");
info_len = 0;
return false;
}
memcpy(info, infoPtr, infoLen);
memcpy(info_ptr, infoPtr, infoLen);
info_len = infoLen;
pending_info = true;
}
@ -207,7 +207,7 @@ void NormObject::Close()
bool NormObject::HandleInfoRequest()
{
bool increasedRepair = false;
if (info)
if (info_ptr)
{
if (!repair_info)
{
@ -1047,7 +1047,7 @@ void NormObject::HandleObjectMessage(const NormObjectMsg& msg,
"Warning! info too long.\n", LocalNodeId(), server->GetId(),
(UINT16)transport_id);
}
memcpy(info, infoMsg.GetInfo(), info_len);
memcpy(info_ptr, infoMsg.GetInfo(), info_len);
pending_info = false;
session.Notify(NormController::RX_OBJECT_INFO, server, this);
}
@ -1219,9 +1219,10 @@ void NormObject::HandleObjectMessage(const NormObjectMsg& msg,
block->SetPending(nextSegment);
block->IncrementErasureCount();
// Clear any erasure/retrieval segments
for (UINT16 i = 0; i < erasureCount; i++)
UINT16 i;
for (i = 0; i < erasureCount; i++)
block->DetachSegment(server->GetErasureLoc(i));
for (UINT16 i = 0; i < retrievalCount; i++)
for (i = 0; i < retrievalCount; i++)
block->DetachSegment(server->GetRetrievalLoc(i));
return;
}
@ -1272,7 +1273,7 @@ void NormObject::HandleObjectMessage(const NormObjectMsg& msg,
if (objectUpdated && notify_on_update)
{
notify_on_update = false;
session.Notify(NormController::RX_OBJECT_UPDATE, server, this);
session.Notify(NormController::RX_OBJECT_UPDATED, server, this);
}
}
else
@ -1418,7 +1419,7 @@ bool NormObject::NextServerMsg(NormObjectMsg* msg)
default:
break;
}
if (info) msg->SetFlag(NormObjectMsg::FLAG_INFO);
if (info_ptr) msg->SetFlag(NormObjectMsg::FLAG_INFO);
msg->SetFecId(129);
msg->SetObjectId(transport_id);
@ -1435,7 +1436,7 @@ bool NormObject::NextServerMsg(NormObjectMsg* msg)
{
// (TBD) set REPAIR_FLAG for retransmitted info
NormInfoMsg* infoMsg = (NormInfoMsg*)msg;
infoMsg->SetInfo(info, info_len);
infoMsg->SetInfo(info_ptr, info_len);
pending_info = false;
return true;
}
@ -1938,7 +1939,7 @@ NormDataObject::NormDataObject(class NormSession& theSession,
const NormObjectId& objectId)
: NormObject(DATA, theSession, theServer, objectId),
large_block_length(0,0), small_block_length(0,0),
data_ptr(NULL), data_max(0)
data_ptr(NULL), data_max(0), data_released(false)
{
}
@ -1946,11 +1947,21 @@ NormDataObject::NormDataObject(class NormSession& theSession,
NormDataObject::~NormDataObject()
{
Close();
if (data_released)
{
if (data_ptr)
{
delete data_ptr;
data_ptr = NULL;
}
data_released = false;
}
}
// Assign data object to data ptr
bool NormDataObject::Open(char* dataPtr,
UINT32 dataLen,
bool dataRelease,
const char* infoPtr,
UINT16 infoLen)
{
@ -1976,15 +1987,16 @@ bool NormDataObject::Open(char* dataPtr,
}
data_ptr = dataPtr;
data_max = dataLen;
data_released = dataRelease;
large_block_length = NormObjectSize(large_block_size) * segment_size;
small_block_length = NormObjectSize(small_block_size) * segment_size;
return true;
} // end NormDataObject::Open()
bool NormDataObject::Accept(char* dataPtr, UINT32 dataMax)
bool NormDataObject::Accept(char* dataPtr, UINT32 dataMax, bool dataRelease)
{
ASSERT(NULL == server);
if (Open(dataPtr, dataMax))
if (Open(dataPtr, dataMax, dataRelease))
{
NormObject::Accept();
return true;
@ -2005,6 +2017,11 @@ bool NormDataObject::WriteSegment(NormBlockId blockId,
const char* buffer,
bool /*msgStart*/)
{
if (NULL == data_ptr)
{
DMSG(0, "NormDataObject::WriteSegment() error: NULL data_ptr\n");
return false;
}
UINT16 len;
if (blockId == final_block_id)
{
@ -2046,6 +2063,11 @@ UINT16 NormDataObject::ReadSegment(NormBlockId blockId,
char* buffer,
bool* /*msgStart*/)
{
if (NULL == data_ptr)
{
DMSG(0, "NormDataObject::ReadSegment() error: NULL data_ptr\n");
return 0;
}
// Determine segment length from blockId::segmentId
UINT16 len;
if (blockId == final_block_id)
@ -2075,14 +2097,23 @@ UINT16 NormDataObject::ReadSegment(NormBlockId blockId,
segmentSize*segmentId;
}
ASSERT(0 == segmentOffset.MSB());
ASSERT(data_max >= (segmentOffset.LSB() + len));
if (data_max <= segmentOffset.LSB())
return 0;
else if (data_max <= (segmentOffset.LSB() + len))
len -= (segmentOffset.LSB() + len - data_max);
memcpy(buffer, data_ptr + segmentOffset.LSB(), len);
return true;
return len;
} // end NormDataObject::ReadSegment()
char* NormDataObject::RetrieveSegment(NormBlockId blockId,
NormSegmentId segmentId)
{
if (NULL == data_ptr)
{
DMSG(0, "NormDataObject::RetrieveSegment() error: NULL data_ptr\n");
return NULL;
}
// Determine segment length from blockId::segmentId
UINT16 len;
if (blockId == final_block_id)
@ -2101,7 +2132,7 @@ char* NormDataObject::RetrieveSegment(NormBlockId blockId,
if (server)
{
char* segment = server->GetRetrievalSegment();
ReadSegment(blockId, segmentId, segment);
len = ReadSegment(blockId, segmentId, segment);
memset(segment+len, 0, segment_size-len);
return segment;
}
@ -2490,7 +2521,7 @@ bool NormStreamObject::WriteSegment(NormBlockId blockId,
block->GetFirstPending(read_index.segment);
NormBlock* tempBlock = block;
UINT32 tempOffset = read_offset;
session.Notify(NormController::RX_OBJECT_UPDATE, server, this);
session.Notify(NormController::RX_OBJECT_UPDATED, server, this);
block = stream_buffer.Find(stream_buffer.RangeLo());
if (tempBlock == block)
{
@ -2954,6 +2985,18 @@ NormSimObject::~NormSimObject()
}
bool NormSimObject::Open(UINT32 objectSize,
const char* infoPtr ,
UINT16 infoLen)
{
return (server ?
true :
NormObject::Open(objectSize, infoPtr, infoLen,
session.ServerSegmentSize(),
session.ServerBlockSize(),
session.ServerNumParity()));
} // end NormSimObject::Open()
UINT16 NormSimObject::ReadSegment(NormBlockId blockId,
NormSegmentId segmentId,
char* buffer,
@ -2975,6 +3018,12 @@ UINT16 NormSimObject::ReadSegment(NormBlockId blockId,
return len;
} // end NormSimObject::ReadSegment()
char* NormSimObject::RetrieveSegment(NormBlockId blockId,
NormSegmentId segmentId)
{
return server ? server->GetRetrievalSegment() : NULL;
} // end NormSimObject::RetrieveSegment()
#endif // SIMULATE
/////////////////////////////////////////////////////////////////

27
common/normObject.h
View File

@ -51,7 +51,9 @@ class NormObject
const NormObjectId& GetId() const {return transport_id;}
const NormObjectSize& GetSize() const {return object_size;}
bool HaveInfo() const {return (info_len > 0);}
const char* GetInfo() const {return info;}
bool HasInfo() const {return (NULL != info_ptr);}
const char* GetInfo() const {return info_ptr;}
UINT16 GetInfoLength() const {return info_len;}
bool IsStream() const {return (STREAM == type);}
@ -257,7 +259,7 @@ class NormObject
NormBlockId final_block_id;
UINT16 final_segment_size;
NackingMode nacking_mode;
char* info;
char* info_ptr;
UINT16 info_len;
// Here are some members used to let us know
@ -321,12 +323,19 @@ class NormDataObject : public NormObject
bool Open(char* dataPtr,
UINT32 dataLen,
bool dataRelease,
const char* infoPtr = NULL,
UINT16 infoLen = 0);
bool Accept(char* dataPtr, UINT32 dataMax);
bool Accept(char* dataPtr, UINT32 dataMax, bool dataRelease);
void Close();
const char* GetData() {return data_ptr;}
char* DetachData()
{
char* dataPtr = data_ptr;
data_ptr = NULL;
return dataPtr;
}
virtual bool WriteSegment(NormBlockId blockId,
NormSegmentId segmentId,
@ -346,6 +355,8 @@ class NormDataObject : public NormObject
NormObjectSize small_block_length;
char* data_ptr;
UINT32 data_max;
bool data_released; // when true, data_ptr is deleted
// on NormDataObject destruction
}; // end class NormDataObject
@ -479,10 +490,7 @@ class NormSimObject : public NormObject
bool Open(UINT32 objectSize,
const char* infoPtr = NULL,
UINT16 infoLen = 0)
{
return server ? true : NormObject::Open(objectSize, infoPtr, infoLen);
}
UINT16 infoLen = 0);
bool Accept() {NormObject::Accept(); return true;}
void Close() {NormObject::Close();}
@ -497,10 +505,7 @@ class NormSimObject : public NormObject
bool* msgStart = NULL);
virtual char* RetrieveSegment(NormBlockId blockId,
NormSegmentId segmentId)
{
return server ? server->GetRetrievalSegment() : NULL;
}
NormSegmentId segmentId);
}; // end class NormSimObject
#endif // SIMULATE

100
common/normSession.cpp
View File

@ -20,7 +20,7 @@ NormSession::NormSession(NormSessionMgr& sessionMgr, NormNodeId localNodeId)
tx_socket(ProtoSocket::UDP), rx_socket(ProtoSocket::UDP),
local_node_id(localNodeId),
ttl(DEFAULT_TTL), loopback(false),
tx_rate(DEFAULT_TRANSMIT_RATE/8.0),
tx_rate(DEFAULT_TRANSMIT_RATE/8.0), tx_rate_min(-1.0), tx_rate_max(-1.0),
backoff_factor(DEFAULT_BACKOFF_FACTOR), is_server(false), session_id(0),
ndata(DEFAULT_NDATA), nparity(DEFAULT_NPARITY), auto_parity(0), extra_parity(0),
next_tx_object_id(0), tx_cache_count_min(8), tx_cache_count_max(256),
@ -112,12 +112,6 @@ bool NormSession::Open(const char* interfaceName)
if (address.IsMulticast())
{
if (!rx_socket.JoinGroup(address, interfaceName))
{
DMSG(0, "NormSession::Open() rx_socket join group error\n");
Close();
return false;
}
if (!tx_socket.SetTTL(ttl))
{
DMSG(0, "NormSession::Open() tx_socket set ttl error\n");
@ -132,11 +126,27 @@ bool NormSession::Open(const char* interfaceName)
}
if (interfaceName)
{
rx_socket.SetMulticastInterface(interfaceName);
tx_socket.SetMulticastInterface(interfaceName);
strncpy(interface_name, interfaceName, 31);
interface_name[31] = '\0';
}
if ('\0' != interface_name[0])
{
bool result = rx_socket.SetMulticastInterface(interface_name);
result &= tx_socket.SetMulticastInterface(interface_name);
if (!result)
{
DMSG(0, "NormSession::Open() error setting multicast interface\n");
Close();
return false;
}
interfaceName = interface_name;
}
if (!rx_socket.JoinGroup(address, interfaceName))
{
DMSG(0, "NormSession::Open() rx_socket join group error\n");
Close();
return false;
}
}
for (unsigned int i = 0; i < DEFAULT_MESSAGE_POOL_DEPTH; i++)
{
@ -179,6 +189,24 @@ void NormSession::Close()
} // end NormSession::Close()
bool NormSession::SetMulticastInterface(const char* interfaceName)
{
if (interfaceName)
{
bool result = true;
if (rx_socket.IsOpen())
result &= rx_socket.SetMulticastInterface(interfaceName);
if (tx_socket.IsOpen())
result &= tx_socket.SetMulticastInterface(interfaceName);
return result;
}
else
{
interface_name[0] = '\0';
return true;
}
} // end NormSession::SetMulticastInterface()
void NormSession::SetTxRate(double txRate)
{
txRate /= 8.0; // convert to bytes/sec
@ -193,7 +221,7 @@ void NormSession::SetTxRate(double txRate)
}
tx_rate = txRate;
}
else if (0.0 == tx_rate)
else if ((0.0 == tx_rate) && IsOpen())
{
tx_rate = txRate;
tx_timer.SetInterval(0.0);
@ -205,6 +233,37 @@ void NormSession::SetTxRate(double txRate)
}
} // end NormSession::SetTxRate()
void NormSession::SetTxRateBounds(double rateMin, double rateMax)
{
// Make sure min <= max
if ((rateMin >= 0.0) && (rateMax >= 0.0))
{
if (rateMin > rateMax)
{
double temp = rateMin;
rateMin = rateMax;
rateMax = temp;
}
}
if (rateMin < 0.0)
tx_rate_min = -1.0;
else
tx_rate_min = rateMin/8.0; // convert to bytes/second
if (rateMax < 0.0)
tx_rate_max = -1.0;
else
tx_rate_max = rateMax/8.0; // convert to bytes/second
if (cc_enable)
{
if ((tx_rate_min >= 0.0) && (tx_rate < tx_rate_min))
tx_rate = tx_rate_min;
if ((tx_rate_max >= 0.0) && (tx_rate > tx_rate_max))
tx_rate = tx_rate_max;
SetTxRate(tx_rate*8.0);
}
} // end NormSession::SetTxRateBounds()
bool NormSession::StartServer(UINT32 bufferSpace,
UINT16 segmentSize,
UINT16 numData,
@ -285,7 +344,15 @@ bool NormSession::StartServer(UINT32 bufferSpace,
//probe_timer.SetInterval(0.0);
probe_pending = probe_reset = false;
if (cc_enable) tx_rate = segmentSize;
if (cc_enable)
{
tx_rate = segmentSize;
if ((tx_rate_min >= 0.0) && (tx_rate < tx_rate_min))
tx_rate = tx_rate_min;
if ((tx_rate_max >= 0.0) && (tx_rate > tx_rate_max))
tx_rate = tx_rate_max;
}
OnProbeTimeout(probe_timer);
ActivateTimer(probe_timer);
return true;
@ -782,7 +849,7 @@ NormDataObject* NormSession::QueueTxData(const char* dataPtr,
strerror(errno));
return NULL;
}
if (!obj->Open((char*)dataPtr, dataLen, infoPtr, infoLen))
if (!obj->Open((char*)dataPtr, dataLen, false, infoPtr, infoLen))
{
DMSG(0, "NormSession::QueueTxData() object open error\n");
delete obj;
@ -1263,6 +1330,7 @@ void NormSession::ClientHandleObjectMessage(const struct timeval& currentTime,
{
if ((theServer = new NormServerNode(*this, msg.GetSourceId())))
{
Notify(NormController::REMOTE_SERVER_NEW, theServer, NULL);
if (theServer->Open(msg.GetSessionId()))
{
server_tree.AttachNode(theServer);
@ -2821,10 +2889,18 @@ void NormSession::AdjustRate(bool onResponse)
}*/
}
// Don't let tx_rate below MIN(one segment per grtt, one segment per second)
double minRate = ((double)segment_size) / grtt_measured;
minRate = MIN((double)segment_size, minRate);
tx_rate = MAX(tx_rate, minRate);
// Keep "tx_rate" within user set rate bounds (if any)
if ((tx_rate_min >= 0.0) && (tx_rate < tx_rate_min))
tx_rate = tx_rate_min;
if ((tx_rate_max >= 0.0) && (tx_rate > tx_rate_max))
tx_rate = tx_rate_max;
struct timeval currentTime;
::ProtoSystemTime(currentTime);
double theTime = (double)currentTime.tv_sec + 1.0e-06 * ((double)currentTime.tv_usec);

13
common/normSession.h
View File

@ -20,11 +20,12 @@ class NormController
TX_OBJECT_PURGED,
LOCAL_SERVER_CLOSED,
REMOTE_SERVER_NEW,
REMOTE_SERVER_INACTIVE,
REMOTE_SERVER_ACTIVE,
REMOTE_SERVER_INACTIVE,
REMOTE_SERVER_PURGED,
RX_OBJECT_NEW,
RX_OBJECT_INFO,
RX_OBJECT_UPDATE,
RX_OBJECT_UPDATED,
RX_OBJECT_COMPLETED,
RX_OBJECT_ABORTED
};
@ -102,6 +103,7 @@ class NormSession
bool IsOpen() {return (rx_socket.IsOpen() || tx_socket.IsOpen());}
const ProtoAddress& Address() {return address;}
void SetAddress(const ProtoAddress& addr) {address = addr;}
bool SetMulticastInterface(const char* interfaceName);
bool SetTTL(UINT8 theTTL)
{
bool result = tx_socket.IsOpen() ? tx_socket.SetTTL(theTTL) : true;
@ -129,6 +131,7 @@ class NormSession
void SetBackoffFactor(double value) {backoff_factor = value;}
bool CongestionControl() {return cc_enable;}
void SetCongestionControl(bool state) {cc_enable = state;}
void SetTxRateBounds(double rateMin, double rateMax);
void Notify(NormController::Event event,
class NormServerNode* server,
@ -249,8 +252,8 @@ class NormSession
bool IsClient() const {return is_client;}
unsigned long RemoteServerBufferSize() const
{return remote_server_buffer_size;}
void SetUnicastNacks(bool state) {unicast_nacks = state;}
bool UnicastNacks() const {return unicast_nacks;}
void ClientSetUnicastNacks(bool state) {unicast_nacks = state;}
bool ClientGetUnicastNacks() const {return unicast_nacks;}
void ClientSetSilent(bool state) {client_silent = state;}
bool ClientIsSilent() const {return client_silent;}
@ -342,6 +345,8 @@ class NormSession
bool loopback; // to receive own traffic
char interface_name[32];
double tx_rate; // bytes per second
double tx_rate_min;
double tx_rate_max;
double backoff_factor;
// Server parameters and state

23
common/normSimAgent.cpp
View File

@ -20,7 +20,7 @@ NormSimAgent::NormSimAgent(ProtoTimerMgr& timerMgr,
tx_object_size(0), tx_object_interval(0.0),
tx_object_size_min(0), tx_object_size_max(0),
tx_repeat_count(0), tx_repeat_interval(0.0),
stream(NULL), auto_stream(false), push_stream(false),
stream(NULL), auto_stream(false), push_mode(false),
flush_mode(NormStreamObject::FLUSH_PASSIVE),
mgen(NULL), msg_sync(false), mgen_bytes(0), mgen_pending_bytes(0),
tracing(false), tx_loss(0.0), rx_loss(0.0)
@ -397,6 +397,8 @@ bool NormSimAgent::ProcessCommand(const char* cmd, const char* val)
DMSG(0, "NormSimAgent::ProcessCommand(sendStream) error opening stream!\n");
return false;
}
stream->SetFlushMode(flush_mode);
stream->SetPushMode(push_mode);
auto_stream = true;
}
else
@ -430,6 +432,8 @@ bool NormSimAgent::ProcessCommand(const char* cmd, const char* val)
DMSG(0, "NormSimAgent::ProcessCommand(openStream) error opening stream!\n");
return false;
}
stream->SetFlushMode(flush_mode);
stream->SetPushMode(push_mode);
auto_stream = false;
tx_msg_len = tx_msg_index = 0;
}
@ -469,9 +473,9 @@ bool NormSimAgent::ProcessCommand(const char* cmd, const char* val)
else if (!strncmp("push", cmd, len))
{
if (!strcmp(val, "on"))
push_stream = true;
push_mode = true;
else if (!strcmp(val, "off"))
push_stream = false;
push_mode = false;
else
{
DMSG(0, "NormSimAgent::ProcessCommand(push) invalid argument!\n");
@ -592,13 +596,12 @@ void NormSimAgent::OnInputReady()
{
unsigned int bytesWrote = stream->Write(tx_msg_buffer+tx_msg_index,
tx_msg_len - tx_msg_index,
NormStreamObject::FLUSH_NONE,
false, push_stream);
false);
tx_msg_index += bytesWrote;
if (tx_msg_index == tx_msg_len)
{
// Provide EOM indication to norm stream
stream->Write(NULL, 0, flush_mode, true, false);
// Mark EOM _and_ flush (using set flush mode)
stream->Write(NULL, 0, true);
tx_msg_index = tx_msg_len = 0;
}
}
@ -609,7 +612,7 @@ bool NormSimAgent::FlushStream()
{
if (stream && session && session->IsServer())
{
stream->Write(NULL, 0, NormStreamObject::FLUSH_ACTIVE, false, false);
stream->Flush(true);
return true;
}
else
@ -636,9 +639,7 @@ void NormSimAgent::Notify(NormController::Event event,
// sending a dummy byte stream
char buffer[NormMsg::MAX_SIZE];
unsigned int count =
stream->Write(buffer, segment_size,
NormStreamObject::FLUSH_NONE,
false, false);
stream->Write(buffer, segment_size, false);
}
else
{

4
common/normSimAgent.h
View File

@ -78,8 +78,8 @@ class NormSimAgent : public NormController
NormStreamObject* stream;
bool auto_stream;
bool push_stream;
NormStreamObject::FlushType flush_mode;
bool push_mode;
NormStreamObject::FlushMode flush_mode;
char* tx_msg_buffer;
unsigned int tx_msg_len;
unsigned int tx_msg_index;

13
common/normTest.cpp
View File

@ -64,7 +64,7 @@ int main(int argc, char* argv[])
// NORM_FLUSH_PASSIVE automatically flushes full writes to
// the stream.
NormStreamSetFlushMode(stream, NORM_FLUSH_PASSIVE);
NormStreamSetAutoFlush(stream, NORM_FLUSH_PASSIVE);
// Some variable for stream input/output
@ -143,18 +143,11 @@ int main(int argc, char* argv[])
// Assume info contains '/' delimited <path/fileName> string
if (NORM_OBJECT_FILE == NormObjectGetType(theEvent.object))
{
NormObjectTransportId id = NormObjectGetTransportId(theEvent.object);
if (0 != (id & 0x01))
{
//NormObjectCancel(theEvent.object);
//break;
}
char fileName[PATH_MAX];
strcpy(fileName, cachePath);
int pathLen = strlen(fileName);
unsigned short nameLen = PATH_MAX - pathLen;
NormObjectGetInfo(theEvent.object, fileName+pathLen, &nameLen);
nameLen = NormObjectGetInfo(theEvent.object, fileName+pathLen, nameLen);
fileName[nameLen + pathLen] = '\0';
char* ptr = fileName + 5;
while ('\0' != *ptr)
@ -170,7 +163,7 @@ int main(int argc, char* argv[])
}
break;
case NORM_RX_OBJECT_UPDATE:
case NORM_RX_OBJECT_UPDATED:
{
//TRACE("normTest: NORM_RX_OBJECT_UPDATE event ...\n");
if (NORM_OBJECT_STREAM != NormObjectGetType(theEvent.object))

2
common/normVersion.h
View File

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

BIN
win32/Norm.opt
View File

Binary file not shown.