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

v1.2b4

pull/2/head
Jeff Weston 2019-09-11 11:42:03 -04:00
parent a5e5d140e3
commit f451a006f6
17 changed files with 853 additions and 359 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

21
README.md
View File

@ -4,10 +4,21 @@
The NORM code here is still preliminary but a functional "norm" The NORM code here is still preliminary but a functional "norm"
command-line application can be built which can send and receive command-line application can be built which can send and receive
a set of files _or_ a stream piped to/from stdin/stdout. The command- a set of files _or_ a stream piped to/from stdin/stdout. The command-
line options are not yet fully documented. line options are not yet fully documented, but there is abbreviated "help"
available in the "norm" application. And many options are similar to the
preceding NRL MDP work (see http://mdp.pf.itd.nrl.navy.mil).
The current code only has Makefiles for Unix platforms, but the A "normTest.cpp" program is now included which is a simple demonstration
code could be assembled as Win32 project under VC++. (and active test code) of the evolving NORM API. When the API is
further completed, I plan to provide a few different example applications
(file transfer, streaming, etc) which use the API.
The NORM code depends upon the current "Protolib" release. The Win32 code is not yet as well-tested as the Unix version, but
See <http://pf.itd.nrl.navy.mil> for that code. might be OK at this point. I will do more Win32 testing when I get
caught up on the API development and documentation.
The NORM code depends upon the current "Protolib" release. The NORM
source code tarballs contain an appropriate release of "Protolib" as
part of the NORM source tree. If the NORM code is checked out from
our CVS server, it is necessary to also check out "protolib" separately
and provide paths to it for NORM to build.

13
VERSION.TXT
View File

@ -1,5 +1,18 @@
NORM Version History NORM Version History
Version 1.2b4
=============
- Finally remembered to update this file!
- 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
(documentation soon to follow, I promise!)
- There have been some "cc" tweaks and bug fixes.
- NORM RFC 3940 was released and this version is in compliance.
- Added optional positive acknowledgements for sender-defined
"watermarks".
Version 1.1b9 Version 1.1b9
============= =============
- There have been a number of bug fixes since 1.1b3. This release - There have been a number of bug fixes since 1.1b3. This release

169
common/normApi.cpp
View File

@ -217,12 +217,19 @@ void NormInstance::Notify(NormController::Event event,
NormStreamObject* stream = static_cast<NormStreamObject*>(object); NormStreamObject* stream = static_cast<NormStreamObject*>(object);
// (TBD) implement silent_client accept differently // (TBD) implement silent_client accept differently
NormObjectSize size = stream->GetSize(); NormObjectSize size = stream->GetSize();
if (!stream->Accept(size.LSB())) // We double the size to prevent unecessary data loss
// for our threaded API
if (!stream->Accept(2*size.LSB()))
{ {
DMSG(0, "NormInstance::Notify() stream accept error\n"); DMSG(0, "NormInstance::Notify() stream accept error\n");
notify_pool.Append(n); notify_pool.Append(n);
return; return;
} }
// By setting a non-zero "block pool threshold", this
// gives the API a chance to "catch up" on reading
// when the receive stream becomes buffer-constrained
UINT32 blockPoolCount = stream->GetBlockPoolCount();
stream->SetBlockPoolThreshold(blockPoolCount / 2);
break; break;
} }
case NormObject::FILE: case NormObject::FILE:
@ -688,33 +695,6 @@ void NormSetRxLoss(NormSessionHandle sessionHandle, double percent)
} // end NormSetRxLoss() } // end NormSetRxLoss()
NormNodeId NormGetNodeId(NormNodeHandle nodeHandle)
{
NormNode* node = (NormNode*)nodeHandle;
if (node)
return node->GetId();
else
return NORM_NODE_NONE;
} // end NormGetNodeId()
NormRepairBoundary NormGetNodeRepairBoundary(NormNodeHandle nodeHandle)
{
NormServerNode* node = static_cast<NormServerNode*>((NormNode*)nodeHandle);
if (node)
return ((NormRepairBoundary)(node->GetRepairBoundary()));
else
return NORM_BOUNDARY_BLOCK; // return default value
} // end NormGetNodeRepairBoundary()
void NormSetNodeRepairBoundary(NormNodeHandle nodeHandle,
NormRepairBoundary repairBoundary)
{
NormServerNode* node = static_cast<NormServerNode*>((NormNode*)nodeHandle);
if (node)
node->SetRepairBoundary((NormServerNode::RepairBoundary)repairBoundary);
} // end NormSetNodeRepairBoundary()
void NormSetDefaultRepairBoundary(NormSessionHandle sessionHandle, void NormSetDefaultRepairBoundary(NormSessionHandle sessionHandle,
NormRepairBoundary repairBoundary) NormRepairBoundary repairBoundary)
{ {
@ -724,15 +704,44 @@ void NormSetDefaultRepairBoundary(NormSessionHandle sessionHandle,
} // end NormSetDefaultRepairBoundary() } // end NormSetDefaultRepairBoundary()
NormObjectType NormGetObjectType(NormObjectHandle objectHandle) NormRepairBoundary NormNodeGetRepairBoundary(NormNodeHandle nodeHandle)
{
NormServerNode* node = static_cast<NormServerNode*>((NormNode*)nodeHandle);
if (node)
return ((NormRepairBoundary)(node->GetRepairBoundary()));
else
return NORM_BOUNDARY_BLOCK; // return default value
} // end NormNodeGetRepairBoundary()
void NormNodeSetRepairBoundary(NormNodeHandle nodeHandle,
NormRepairBoundary repairBoundary)
{
NormServerNode* node = static_cast<NormServerNode*>((NormNode*)nodeHandle);
if (node)
node->SetRepairBoundary((NormServerNode::RepairBoundary)repairBoundary);
} // end NormNodeSetRepairBoundary()
NormNodeId NormNodeGetId(NormNodeHandle nodeHandle)
{
NormNode* node = (NormNode*)nodeHandle;
if (node)
return node->GetId();
else
return NORM_NODE_NONE;
} // end NormNodeGetId()
NormObjectType NormObjectGetType(NormObjectHandle objectHandle)
{ {
if (NORM_OBJECT_INVALID != objectHandle) if (NORM_OBJECT_INVALID != objectHandle)
return ((NormObjectType)(((NormObject*)objectHandle)->GetType())); return ((NormObjectType)(((NormObject*)objectHandle)->GetType()));
else else
return NORM_OBJECT_NONE; return NORM_OBJECT_NONE;
} // end NormGetObjectType() } // end NormObjectGetType()
NormObjectTransportId NormGetObjectTransportId(NormObjectHandle objectHandle) NormObjectTransportId NormObjectGetTransportId(NormObjectHandle objectHandle)
{ {
// Like many other API calls, these should be changed // Like many other API calls, these should be changed
// to provide an error code result // to provide an error code result
@ -740,9 +749,9 @@ NormObjectTransportId NormGetObjectTransportId(NormObjectHandle objectHandle)
return ((NormObjectTransportId)(((NormObject*)objectHandle)->GetId())); return ((NormObjectTransportId)(((NormObject*)objectHandle)->GetId()));
else else
return 0; return 0;
} // end NormGetFileName() } // end NormObjectGetTransportId()
bool NormGetObjectInfo(NormObjectHandle objectHandle, bool NormObjectGetInfo(NormObjectHandle objectHandle,
char* infoBuffer, char* infoBuffer,
unsigned short* infoLen) unsigned short* infoLen)
{ {
@ -768,9 +777,9 @@ bool NormGetObjectInfo(NormObjectHandle objectHandle,
return result; return result;
} }
return false; return false;
} // end NormGetObjectInfo() } // end NormObjectGetInfo()
void NormCancelObject(NormObjectHandle objectHandle) void NormObjectCancel(NormObjectHandle objectHandle)
{ {
if (NORM_OBJECT_INVALID != objectHandle) if (NORM_OBJECT_INVALID != objectHandle)
{ {
@ -780,16 +789,16 @@ void NormCancelObject(NormObjectHandle objectHandle)
NormObject* obj = (NormObject*)objectHandle; NormObject* obj = (NormObject*)objectHandle;
NormServerNode* server = obj->GetServer(); NormServerNode* server = obj->GetServer();
if (server) if (server)
server->DeleteObject(obj, 8); server->DeleteObject(obj);
else else
obj->GetSession().DeleteTxObject(obj); obj->GetSession().DeleteTxObject(obj);
instance->PurgeObjectNotifications(objectHandle); instance->PurgeObjectNotifications(objectHandle);
instance->dispatcher.ResumeThread(); instance->dispatcher.ResumeThread();
} }
} }
} // end NormCancelObject() } // end NormObjectCancel()
void NormRetainObject(NormObjectHandle objectHandle) void NormObjectRetain(NormObjectHandle objectHandle)
{ {
if (NORM_OBJECT_INVALID != objectHandle) if (NORM_OBJECT_INVALID != objectHandle)
{ {
@ -803,7 +812,7 @@ void NormRetainObject(NormObjectHandle objectHandle)
} }
} // end NormRetainObject() } // end NormRetainObject()
void NormReleaseObject(NormObjectHandle objectHandle) void NormObjectRelease(NormObjectHandle objectHandle)
{ {
if (NORM_OBJECT_INVALID != objectHandle) if (NORM_OBJECT_INVALID != objectHandle)
{ {
@ -812,20 +821,20 @@ void NormReleaseObject(NormObjectHandle objectHandle)
{ {
NormObject* obj = (NormObject*)objectHandle; NormObject* obj = (NormObject*)objectHandle;
obj->Release(); obj->Release();
//instance->PurgeObjectNotifications(objectHandle); instance->PurgeObjectNotifications(objectHandle);
instance->dispatcher.ResumeThread(); instance->dispatcher.ResumeThread();
} }
} }
} // end NormReleaseObject() } // end NormObjectRelease()
NormNackingMode NormGetObjectNackingMode(NormObjectHandle objectHandle) NormNackingMode NormObjectGetNackingMode(NormObjectHandle objectHandle)
{ {
NormObject* object = (NormObject*)objectHandle; NormObject* object = (NormObject*)objectHandle;
if (object) if (object)
return ((NormNackingMode)object->GetNackingMode()); return ((NormNackingMode)object->GetNackingMode());
else else
return NORM_NACK_NONE; return NORM_NACK_NONE;
} // end NormGetObjectNackingMode() } // end NormObjectGetNackingMode()
void NormSetObjectNackingMode(NormObjectHandle objectHandle, void NormSetObjectNackingMode(NormObjectHandle objectHandle,
NormNackingMode nackingMode) NormNackingMode nackingMode)
@ -834,12 +843,12 @@ void NormSetObjectNackingMode(NormObjectHandle objectHandle,
if (object) object->SetNackingMode((NormObject::NackingMode)nackingMode); if (object) object->SetNackingMode((NormObject::NackingMode)nackingMode);
} // end NormSetObjectNackingMode() } // end NormSetObjectNackingMode()
void NormSetNodeNackingMode(NormNodeHandle nodeHandle, void NormObjectSetNackingMode(NormNodeHandle nodeHandle,
NormNackingMode nackingMode) NormNackingMode nackingMode)
{ {
NormServerNode* node = (NormServerNode*)nodeHandle; NormServerNode* node = (NormServerNode*)nodeHandle;
if (node) node->SetDefaultNackingMode((NormObject::NackingMode)nackingMode); if (node) node->SetDefaultNackingMode((NormObject::NackingMode)nackingMode);
} // end NormSetNodeNackingMode() } // end NormObjectSetNackingMode()
void NormSetDefaultNackingMode(NormSessionHandle sessionHandle, void NormSetDefaultNackingMode(NormSessionHandle sessionHandle,
NormNackingMode nackingMode) NormNackingMode nackingMode)
@ -910,7 +919,7 @@ void NormRemoveAckingNode(NormSessionHandle sessionHandle,
} }
} // end NormAddAckingNode() } // end NormAddAckingNode()
NormObjectHandle NormOpenStream(NormSessionHandle sessionHandle, NormObjectHandle NormStreamOpen(NormSessionHandle sessionHandle,
unsigned long bufferSize) unsigned long bufferSize)
{ {
NormObjectHandle objectHandle = NORM_OBJECT_INVALID; NormObjectHandle objectHandle = NORM_OBJECT_INVALID;
@ -927,12 +936,12 @@ NormObjectHandle NormOpenStream(NormSessionHandle sessionHandle,
instance->dispatcher.ResumeThread(); instance->dispatcher.ResumeThread();
} }
return objectHandle; return objectHandle;
} // end NormOpenStream() } // end NormStreamOpen()
void NormCloseStream(NormObjectHandle streamHandle) void NormStreamClose(NormObjectHandle streamHandle)
{ {
NormCancelObject(streamHandle); NormObjectCancel(streamHandle);
} // end NormCloseStream() } // end NormStreamClose()
// (TBD) Some stream i/o performance improvement can be realized // (TBD) Some stream i/o performance improvement can be realized
@ -950,35 +959,35 @@ void NormCloseStream(NormObjectHandle streamHandle)
// events with a "NormDispatchEvents()" call ... // events with a "NormDispatchEvents()" call ...
// //
void NormSetStreamFlushMode(NormObjectHandle streamHandle, void NormStreamSetFlushMode(NormObjectHandle streamHandle,
NormFlushMode flushMode) NormFlushMode flushMode)
{ {
NormStreamObject* stream = NormStreamObject* stream =
static_cast<NormStreamObject*>((NormObject*)streamHandle); static_cast<NormStreamObject*>((NormObject*)streamHandle);
if (stream) if (stream)
stream->SetFlushMode((NormStreamObject::FlushMode)flushMode); stream->SetFlushMode((NormStreamObject::FlushMode)flushMode);
} // end NormSetStreamFlushMode() } // end NormStreamSetFlushMode()
void NormSetStreamPushMode(NormObjectHandle streamHandle, bool state) void NormStreamSetPushMode(NormObjectHandle streamHandle, bool state)
{ {
NormStreamObject* stream = NormStreamObject* stream =
static_cast<NormStreamObject*>((NormObject*)streamHandle); static_cast<NormStreamObject*>((NormObject*)streamHandle);
if (stream) stream->SetPushMode(state); if (stream) stream->SetPushMode(state);
} // end NormSetStreamPushMode() } // end NormStreamSetPushMode()
void NormFlushStream(NormObjectHandle streamHandle) void NormStreamFlush(NormObjectHandle streamHandle, bool eom)
{ {
NormInstance* instance = NormInstance::GetInstanceFromObject(streamHandle); NormInstance* instance = NormInstance::GetInstanceFromObject(streamHandle);
if (instance && instance->dispatcher.SuspendThread()) if (instance && instance->dispatcher.SuspendThread())
{ {
NormStreamObject* stream = NormStreamObject* stream =
static_cast<NormStreamObject*>((NormObject*)streamHandle); static_cast<NormStreamObject*>((NormObject*)streamHandle);
if (stream) stream->Flush(); if (stream) stream->Flush(eom);
instance->dispatcher.ResumeThread(); instance->dispatcher.ResumeThread();
} }
} // end NormFlushStream() } // end NormStreamFlush()
void NormMarkStreamEom(NormObjectHandle streamHandle) void NormStreamMarkEom(NormObjectHandle streamHandle)
{ {
NormInstance* instance = NormInstance::GetInstanceFromObject(streamHandle); NormInstance* instance = NormInstance::GetInstanceFromObject(streamHandle);
if (instance && instance->dispatcher.SuspendThread()) if (instance && instance->dispatcher.SuspendThread())
@ -990,7 +999,7 @@ void NormMarkStreamEom(NormObjectHandle streamHandle)
} }
} // end NormMarkStreamEom() } // end NormMarkStreamEom()
unsigned int NormWriteStream(NormObjectHandle streamHandle, unsigned int NormStreamWrite(NormObjectHandle streamHandle,
const char* buffer, const char* buffer,
unsigned int numBytes) unsigned int numBytes)
{ {
@ -1005,9 +1014,9 @@ unsigned int NormWriteStream(NormObjectHandle streamHandle,
instance->dispatcher.ResumeThread(); instance->dispatcher.ResumeThread();
} }
return result; return result;
} // end NormWriteStream() } // end NormStreamWrite()
bool NormReadStream(NormObjectHandle streamHandle, bool NormStreamRead(NormObjectHandle streamHandle,
char* buffer, char* buffer,
unsigned int* numBytes) unsigned int* numBytes)
{ {
@ -1022,9 +1031,9 @@ bool NormReadStream(NormObjectHandle streamHandle,
instance->dispatcher.ResumeThread(); instance->dispatcher.ResumeThread();
} }
return result; return result;
} // end NormReadStream() } // end NormStreamRead()
bool NormFindStreamMsgStart(NormObjectHandle streamHandle) bool NormStreamSeekMsgStart(NormObjectHandle streamHandle)
{ {
bool result = false; bool result = false;
NormInstance* instance = NormInstance::GetInstanceFromObject(streamHandle); NormInstance* instance = NormInstance::GetInstanceFromObject(streamHandle);
@ -1038,12 +1047,12 @@ bool NormFindStreamMsgStart(NormObjectHandle streamHandle)
instance->dispatcher.ResumeThread(); instance->dispatcher.ResumeThread();
} }
return result; return result;
} // end NormFindStreamMsgStart() } // end NormStreamSeekMsgStart()
NormObjectHandle NormQueueFile(NormSessionHandle sessionHandle, NormObjectHandle NormFileEnqueue(NormSessionHandle sessionHandle,
const char* fileName, const char* fileName,
const char* infoPtr, const char* infoPtr,
unsigned int infoLen) unsigned int infoLen)
{ {
NormObjectHandle objectHandle = NORM_OBJECT_INVALID; NormObjectHandle objectHandle = NORM_OBJECT_INVALID;
NormInstance* instance = NormInstance::GetInstanceFromSession(sessionHandle); NormInstance* instance = NormInstance::GetInstanceFromSession(sessionHandle);
@ -1058,7 +1067,7 @@ NormObjectHandle NormQueueFile(NormSessionHandle sessionHandle,
instance->dispatcher.ResumeThread(); instance->dispatcher.ResumeThread();
} }
return objectHandle; return objectHandle;
} // end NormQueueFile() } // end NormFileEnqueue()
bool NormSetCacheDirectory(NormInstanceHandle instanceHandle, bool NormSetCacheDirectory(NormInstanceHandle instanceHandle,
const char* cachePath) const char* cachePath)
@ -1070,8 +1079,8 @@ bool NormSetCacheDirectory(NormInstanceHandle instanceHandle,
return false; return false;
} // end NormSetCacheDirectory() } // end NormSetCacheDirectory()
bool NormSetFileName(NormObjectHandle fileHandle, bool NormFileRename(NormObjectHandle fileHandle,
const char* fileName) const char* fileName)
{ {
bool result = false; bool result = false;
NormInstance* instance = NormInstance::GetInstanceFromObject(fileHandle); NormInstance* instance = NormInstance::GetInstanceFromObject(fileHandle);
@ -1084,9 +1093,9 @@ bool NormSetFileName(NormObjectHandle fileHandle,
instance->dispatcher.ResumeThread(); instance->dispatcher.ResumeThread();
} }
return result; return result;
} // end NormGetFileName() } // end NormFileRename()
bool NormGetFileName(NormObjectHandle fileHandle, bool NormFileGetName(NormObjectHandle fileHandle,
char* nameBuffer, char* nameBuffer,
unsigned int bufferLen) unsigned int bufferLen)
{ {
@ -1102,13 +1111,13 @@ bool NormGetFileName(NormObjectHandle fileHandle,
result = true; result = true;
} }
return result; return result;
} // end NormGetFileName() } // end NormFileGetName()
NormObjectHandle NormQueueData(NormSessionHandle sessionHandle, NormObjectHandle NormDataEnqueue(NormSessionHandle sessionHandle,
const char* dataPtr, const char* dataPtr,
unsigned long dataLen, unsigned long dataLen,
const char* infoPtr = NULL, const char* infoPtr = NULL,
unsigned int infoLen = 0) unsigned int infoLen = 0)
{ {
NormObjectHandle objectHandle = NORM_OBJECT_INVALID; NormObjectHandle objectHandle = NORM_OBJECT_INVALID;
NormInstance* instance = NormInstance::GetInstanceFromSession(sessionHandle); NormInstance* instance = NormInstance::GetInstanceFromSession(sessionHandle);
@ -1123,7 +1132,7 @@ NormObjectHandle NormQueueData(NormSessionHandle sessionHandle,
instance->dispatcher.ResumeThread(); instance->dispatcher.ResumeThread();
} }
return objectHandle; return objectHandle;
} // end NormQueueData() } // end NormDataEnqueue()
bool NormGetNextEvent(NormInstanceHandle instanceHandle, NormEvent* theEvent) bool NormGetNextEvent(NormInstanceHandle instanceHandle, NormEvent* theEvent)

88
common/normApi.h
View File

@ -93,22 +93,28 @@ enum NormNackingMode
NORM_NACK_NORMAL NORM_NACK_NORMAL
}; };
void NormSetDefaultNackingMode(NormSessionHandle sessionHandle,
NormNackingMode nackingMode);
enum NormRepairBoundary enum NormRepairBoundary
{ {
NORM_BOUNDARY_BLOCK, NORM_BOUNDARY_BLOCK,
NORM_BOUNDARY_OBJECT NORM_BOUNDARY_OBJECT
}; };
NormNodeId NormGetNodeId(NormNodeHandle nodeHandle);
NormRepairBoundary NormGetNodeRepairBoundary(NormNodeHandle nodeHandle);
void NormSetNodeRepairBoundary(NormNodeHandle nodeHandle,
NormRepairBoundary repairBoundary);
void NormSetDefaultRepairBoundary(NormSessionHandle sessionHandle, void NormSetDefaultRepairBoundary(NormSessionHandle sessionHandle,
NormRepairBoundary repairBoundary); 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 // General NormObject functions
@ -121,33 +127,30 @@ enum NormObjectType
}; };
NormObjectType NormGetObjectType(NormObjectHandle objectHandle); NormObjectType NormObjectGetType(NormObjectHandle objectHandle);
bool NormGetObjectInfo(NormObjectHandle objectHandle, bool NormObjectGetInfo(NormObjectHandle objectHandle,
char* infoBuffer, char* infoBuffer,
unsigned short* infoLen); unsigned short* infoLen);
NormObjectTransportId NormGetObjectTransportId(NormObjectHandle objectHandle); NormObjectTransportId NormObjectGetTransportId(NormObjectHandle objectHandle);
void NormCancelObject(NormObjectHandle objectHandle); void NormObjectCancel(NormObjectHandle objectHandle);
void NormRetainObject(NormObjectHandle objectHandle); void NormObjectReNormObjectRetaintain(NormObjectHandle objectHandle);
void NormReleaseObject(NormObjectHandle objectHandle); void NormObjectRelease(NormObjectHandle objectHandle);
// Receiver-only NormObject functions // Receiver-only NormObject functions
NormNackingMode NormGetObjectNackingMode(NormObjectHandle objectHandle); NormNackingMode NormObjectGetNackingMode(NormObjectHandle objectHandle);
void NormSetObjectNackingMode(NormObjectHandle objectHandle, void NormObjectSetNackingMode(NormObjectHandle objectHandle,
NormNackingMode nackingMode); NormNackingMode nackingMode);
void NormSetNodeNackingMode(NormNodeHandle nodeHandle,
NormNackingMode nackingMode);
void NormSetDefaultNackingMode(NormSessionHandle sessionHandle,
NormNackingMode nackingMode);
// Sender-only NormObject functions // Sender-only NormObject functions
bool NormSetWatermark(NormSessionHandle sessionHandle, bool NormSetWatermark(NormSessionHandle sessionHandle,
@ -156,10 +159,10 @@ bool NormSetWatermark(NormSessionHandle sessionHandle,
// NormStreamObject functions // NormStreamObject functions
NormObjectHandle NormOpenStream(NormSessionHandle sessionHandle, NormObjectHandle NormStreamOpen(NormSessionHandle sessionHandle,
unsigned long bufferSize); unsigned long bufferSize);
void NormCloseStream(NormObjectHandle streamHandle); void NormStreamClose(NormObjectHandle streamHandle);
enum NormFlushMode enum NormFlushMode
{ {
@ -168,47 +171,47 @@ enum NormFlushMode
NORM_FLUSH_ACTIVE NORM_FLUSH_ACTIVE
}; };
void NormSetStreamFlushMode(NormObjectHandle streamHandle, void NormStreamSetFlushMode(NormObjectHandle streamHandle,
NormFlushMode flushMode); NormFlushMode flushMode);
void NormSetStreamPushMode(NormObjectHandle streamHandle, void NormStreamSetPushMode(NormObjectHandle streamHandle,
bool state); bool state);
unsigned int NormWriteStream(NormObjectHandle streamHandle, unsigned int NormStreamWrite(NormObjectHandle streamHandle,
const char* buffer, const char* buffer,
unsigned int numBytes); unsigned int numBytes);
void NormFlushStream(NormObjectHandle streamHandle); void NormStreamFlush(NormObjectHandle streamHandle, bool eom = false);
void NormMarkStreamEom(NormObjectHandle streamHandle); void NormStreamMarkEom(NormObjectHandle streamHandle);
bool NormReadStream(NormObjectHandle streamHandle, bool NormStreamRead(NormObjectHandle streamHandle,
char* buffer, char* buffer,
unsigned int* numBytes); unsigned int* numBytes);
bool NormFindStreamMsgStart(NormObjectHandle streamHandle); bool NormStreamSeekMsgStart(NormObjectHandle streamHandle);
// NormFileObject Functions // NormFileObject Functions
NormObjectHandle NormQueueFile(NormSessionHandle sessionHandle, NormObjectHandle NormFileEnqueue(NormSessionHandle sessionHandle,
const char* fileName, const char* fileName,
const char* infoPtr = (const char*)0, const char* infoPtr = (const char*)0,
unsigned int infoLen = 0); unsigned int infoLen = 0);
bool NormSetCacheDirectory(NormInstanceHandle instanceHandle, bool NormSetCacheDirectory(NormInstanceHandle instanceHandle,
const char* cachePath); const char* cachePath);
bool NormGetFileName(NormObjectHandle fileHandle, bool NormFileGetName(NormObjectHandle fileHandle,
char* nameBuffer, char* nameBuffer,
unsigned int bufferLen); unsigned int bufferLen);
bool NormSetFileName(NormObjectHandle fileHandle, bool NormFileRename(NormObjectHandle fileHandle,
const char* fileName); const char* fileName);
// NormDataObject Functions // NormDataObject Functions
bool NormQueueData(const char* dataPtr, bool NormDataEnqueue(const char* dataPtr,
unsigned long dataLen, unsigned long dataLen,
const char* infoPtr = (const char*)0, const char* infoPtr = (const char*)0,
unsigned int infoLen = 0); unsigned int infoLen = 0);
// NORM Event Notification Routines // NORM Event Notification Routines
@ -252,9 +255,10 @@ bool NormGetNextEvent(NormInstanceHandle instanceHandle, NormEvent* theEvent);
// not block. // not block.
#ifdef WIN32 #ifdef WIN32
HANDLE NormGetDescriptor(NormInstanceHandle instanceHandle); typedef HANDLE NormDescriptor;
#else #else
int NormGetDescriptor(NormInstanceHandle instanceHandle); typedef int NormDescriptor;
#endif // if/else WIN32/UNIX #endif // if/else WIN32/UNIX
NormDescriptor NormGetDescriptor(NormInstanceHandle instanceHandle);
#endif // _NORM_API #endif // _NORM_API

65
common/normApp.cpp
View File

@ -35,6 +35,7 @@ class NormApp : public NormController, public ProtoApp
const void* userData); const void* userData);
private: private:
void ShowHelp();
void OnInputReady(); void OnInputReady();
enum CmdType {CMD_INVALID, CMD_NOARG, CMD_ARG}; enum CmdType {CMD_INVALID, CMD_NOARG, CMD_ARG};
@ -184,6 +185,7 @@ NormApp::~NormApp()
const char* const NormApp::cmd_list[] = const char* const NormApp::cmd_list[] =
{ {
"-help", // show this help"
"+debug", // debug level "+debug", // debug level
"+log", // log file name "+log", // log file name
"+trace", // message tracing on "+trace", // message tracing on
@ -225,6 +227,56 @@ const char* const NormApp::cmd_list[] =
NULL NULL
}; };
// Thanks to Marinho Barcellos taking initiative with this help function
// (Someday a "norm" user's guide will be completed!"
void NormApp::ShowHelp()
{
// TBD: this should be taken automatically from the cmd array
fprintf(stderr,
"List of commands taken by \"norm\" (+ indicates that an argument is required):\n"
" -help, // show this help\n"
" +debug, // debug level\n"
" +log, // log file name\n"
" +trace, // message tracing on\n"
" +txloss, // tx packet loss percent (for testing)\n"
" +rxloss, // rx packet loss percent (for testing)\n"
" +address, // session destination address\n"
" +ttl, // multicast hop count scope\n"
" +loopback, // 'on' or 'off' to recv our own packets (default = 'off')\n"
" +interface, // multicast interface name to use\n"
" +cc, // congestion control 'on' or 'off'\n"
" +rate, // tx date rate (bps)\n"
" -push, // push stream writes for real-time messaging\n"
" +flush, // message flushing mode ('none', 'passive', or 'active')\n"
" +input, // send stream input\n"
" +output, // recv stream output\n"
" +minput, // sender message stream input\n"
" +moutput, // receiver message stream output\n"
" +sendfile, // file/directory list to transmit\n"
" +interval, // delay time (sec) between files (0.0 sec default)\n"
" +repeatcount, // How many times to repeat the file/directory list tx\n"
" +rinterval, // Interval (sec) between file/directory list repeats\n"
" -updatesOnly, // only send updated files on repeat transmission\n"
" +rxcachedir, // recv file cache directory\n"
" +segment, // payload segment size (bytes)\n"
" +block, // User data packets per FEC coding block (blockSize)\n"
" +parity, // FEC packets calculated per coding block (nparity)\n"
" +auto, // Number of FEC packets to proactively send (<= nparity)\n"
" +extra, // Number of extra FEC packets sent in response to repair requests\n"
" +backoff, // Backoff factor to use\n"
" +grtt, // Set sender's initial GRTT estimate\n"
" +gsize, // Set sender's group size estimate\n"
" +txbuffer, // Size of sender's buffer\n"
" +rxbuffer, // Size receiver allocates for buffering each sender\n"
" +rxsockbuffer, // Optional recv socket buffer size.\n"
" -unicastNacks, // unicast instead of multicast feedback messages\n"
" -silentClient, // silent (non-nacking) receiver (EMCON mode)\n"
" +processor, // receive file post processing command\n"
" +instance, // specify norm instance name for remote control commands\n"
"\n");
} // end NormApp::ShowHelp()
void NormApp::OnControlEvent(ProtoSocket& /*theSocket*/, ProtoSocket::Event theEvent) void NormApp::OnControlEvent(ProtoSocket& /*theSocket*/, ProtoSocket::Event theEvent)
{ {
switch (theEvent) switch (theEvent)
@ -275,6 +327,12 @@ bool NormApp::OnCommand(const char* cmd, const char* val)
return false; return false;
} }
if (!strncmp("help", cmd, len))
{
ShowHelp();
return false;
}
if (control_remote) if (control_remote)
{ {
// Since the "instance" already exists, we send the command // Since the "instance" already exists, we send the command
@ -1391,7 +1449,12 @@ bool NormApp::OnIntervalTimeout(ProtoTimer& /*theTimer*/)
bool NormApp::OnStartup(int argc, const char*const* argv) bool NormApp::OnStartup(int argc, const char*const* argv)
{ {
if (argc < 3)
{
ShowHelp();
return false;
}
if (!(post_processor = NormPostProcessor::Create())) if (!(post_processor = NormPostProcessor::Create()))
{ {
DMSG(0, "NormApp::OnStartup() error creating post processor\n"); DMSG(0, "NormApp::OnStartup() error creating post processor\n");

2
common/normBitmask.cpp
View File

@ -201,8 +201,6 @@ void NormBitmask::Destroy()
bool NormBitmask::GetNextSet(UINT32& index) const bool NormBitmask::GetNextSet(UINT32& index) const
{ {
//TRACE("NormBitmask::GetNextSet(%lu) first_set:%lu num_bits:%lu\n",
// index, first_set, num_bits);
if (index >= num_bits) return false; if (index >= num_bits) return false;
if (index < first_set) return GetFirstSet(index); if (index < first_set) return GetFirstSet(index);
UINT32 maskIndex = index >> 3; UINT32 maskIndex = index >> 3;

2
common/normEncoder.cpp
View File

@ -409,7 +409,7 @@ int NormDecoder::Decode(char** dVec, int ndata, UINT16 erasureCount, UINT16* era
int j; int j;
for (j = 1; j < degree; j += 2) for (j = 1; j < degree; j += 2)
denom ^= gmult(Lambda[j], gexp(((255-k)*(j-1)) % 255)); denom ^= gmult(Lambda[j], gexp(((255-k)*(j-1)) % 255));
// Invert for use computing errror value below // Invert for use computing error value below
denom = ginv(denom); denom = ginv(denom);
// Now evaluate Omega at alpha^(-i) (numerator) // Now evaluate Omega at alpha^(-i) (numerator)

171
common/normNode.cpp
View File

@ -48,6 +48,7 @@ NormServerNode::NormServerNode(class NormSession& theSession, NormNodeId nodeId)
: NormNode(theSession, nodeId), session_id(0), synchronized(false), sync_id(0), : NormNode(theSession, nodeId), session_id(0), synchronized(false), sync_id(0),
max_pending_range(256), is_open(false), segment_size(0), ndata(0), nparity(0), max_pending_range(256), is_open(false), segment_size(0), ndata(0), nparity(0),
repair_boundary(BLOCK_BOUNDARY), erasure_loc(NULL), repair_boundary(BLOCK_BOUNDARY), erasure_loc(NULL),
retrieval_loc(NULL), retrieval_pool(NULL),
cc_sequence(0), cc_enable(false), cc_rate(0.0), cc_sequence(0), cc_enable(false), cc_rate(0.0),
rtt_confirmed(false), is_clr(false), is_plr(false), rtt_confirmed(false), is_clr(false), is_plr(false),
slow_start(true), send_rate(0.0), recv_rate(0.0), recv_accumulator(0), slow_start(true), send_rate(0.0), recv_rate(0.0), recv_accumulator(0),
@ -140,22 +141,42 @@ void NormServerNode::Close()
is_open = false; is_open = false;
} // end NormServerNode::Close() } // end NormServerNode::Close()
bool NormServerNode::AllocateBuffers(UINT16 segmentSize, UINT16 numData, UINT16 numParity) bool NormServerNode::AllocateBuffers(UINT16 segmentSize,
UINT16 numData,
UINT16 numParity)
{ {
ASSERT(IsOpen()); ASSERT(IsOpen());
// Calculate how much memory each buffered block will require // Calculate how much memory each buffered block will require
UINT16 blockSize = numData + numParity; UINT16 blockSize = numData + numParity;
unsigned long maskSize = blockSize >> 3; unsigned long maskSize = blockSize >> 3;
if (0 != (blockSize & 0x07)) maskSize++; if (0 != (blockSize & 0x07)) maskSize++;
unsigned long blockSpace = sizeof(NormBlock) + unsigned long blockStateSpace = sizeof(NormBlock) + blockSize * sizeof(char*) + 2*maskSize;
blockSize * sizeof(char*) +
2*maskSize +
numData * (segmentSize + NormDataMsg::GetStreamPayloadHeaderLength());
unsigned long bufferSpace = session.RemoteServerBufferSize(); unsigned long bufferSpace = session.RemoteServerBufferSize();
// The "bufferFactor" weight determines the ratio of segment buffers (blockSegmentSpace) to
// allocated NormBlock (blockStateSpace).
// If "bufferFactor = 1.0", this is equivalent to the old scheme, where every allocated
// block can be fully buffered (numData segs) for decoding (no seeking required). If
// "bufferFactor = 0.0", only a guarantee of at least "numParity" segments per block is
// enforced. Note that "bufferFactor" values > 0.0 help reduce "seeking" for decoding,
// but reduce the number of blocks for which NORM can keep state. Note this only comes
// into play when NORM would be "buffer constrained"
// (TBD) perhaps we should keep more "block state" than we can even buffer parity for ???
// (this would reduce requests for full block retransmissions when resource contrained)
double bufferFactor = 0.0;
unsigned long segPerBlock =
(unsigned long) ((bufferFactor * (double)numData) +
((1.0 - bufferFactor) * (double)numParity) + 0.5);
// If there's no parity, no segment buffering for decoding is required at all!
// (Thus, the full rxbuffer space can be used for block state)
if (0 == numParity) segPerBlock = 0;
unsigned long blockSegmentSpace = segPerBlock * (segmentSize + NormDataMsg::GetStreamPayloadHeaderLength());
unsigned long blockSpace = blockStateSpace+blockSegmentSpace;
unsigned long numBlocks = bufferSpace / blockSpace; unsigned long numBlocks = bufferSpace / blockSpace;
if (bufferSpace > (numBlocks*blockSpace)) numBlocks++; if (bufferSpace > (numBlocks*blockSpace)) numBlocks++;
if (numBlocks < 2) numBlocks = 2; if (numBlocks < 2) numBlocks = 2;
unsigned long numSegments = numBlocks * numData; unsigned long numSegments = numBlocks * segPerBlock;
if (!block_pool.Init(numBlocks, blockSize)) if (!block_pool.Init(numBlocks, blockSize))
{ {
@ -164,7 +185,7 @@ bool NormServerNode::AllocateBuffers(UINT16 segmentSize, UINT16 numData, UINT16
return false; return false;
} }
// The extra byte of segments is used for marking segments // The extra byte of segments is used for marking segments (not any more!! TBD remove)
// which are "start segments" for messages encapsulated in // which are "start segments" for messages encapsulated in
// a NormStreamObject // a NormStreamObject
if (!segment_pool.Init(numSegments, segmentSize+NormDataMsg::GetStreamPayloadHeaderLength()+1)) if (!segment_pool.Init(numSegments, segmentSize+NormDataMsg::GetStreamPayloadHeaderLength()+1))
@ -174,17 +195,44 @@ bool NormServerNode::AllocateBuffers(UINT16 segmentSize, UINT16 numData, UINT16
return false; return false;
} }
// The "retrieval_pool" is used for FEC block decoding
// These segments are temporarily used for "retrieved" source symbol segments
// needed for block decoding (new rx buffer mgmt scheme)
if (!(retrieval_pool = new char*[numData]))
{
DMSG(0, "NormServerNode::Open() new retrieval_pool error: %s\n", GetErrorString());
Close();
return false;
}
for (UINT16 i = 0; i < numData; i++)
{
char* s = new char[segmentSize+NormDataMsg::GetStreamPayloadHeaderLength()];
if (NULL == s)
{
DMSG(0, "NormServerNode::Open() new retrieval segment error: %s\n", GetErrorString());
Close();
return false;
}
retrieval_pool[i] = s;
}
retrieval_index = 0;
if (!(retrieval_loc = new UINT16[numData]))
{
DMSG(0, "NormServerNode::Open() retrieval_loc allocation error: %s\n", GetErrorString());
Close();
return false;
}
if (!decoder.Init(numParity, segmentSize+NormDataMsg::GetStreamPayloadHeaderLength())) if (!decoder.Init(numParity, segmentSize+NormDataMsg::GetStreamPayloadHeaderLength()))
{ {
DMSG(0, "NormServerNode::Open() decoder init error: %s\n", DMSG(0, "NormServerNode::Open() decoder init error\n");
strerror(errno));
Close(); Close();
return false; return false;
} }
if (!(erasure_loc = new UINT16[numParity])) if (!(erasure_loc = new UINT16[numParity]))
{ {
DMSG(0, "NormServerNode::Open() erasure_loc allocation error: %s\n", DMSG(0, "NormServerNode::Open() erasure_loc allocation error: %s\n", GetErrorString());
strerror(errno));
Close(); Close();
return false; return false;
} }
@ -197,12 +245,31 @@ bool NormServerNode::AllocateBuffers(UINT16 segmentSize, UINT16 numData, UINT16
void NormServerNode::FreeBuffers() void NormServerNode::FreeBuffers()
{ {
decoder.Destroy();
if (erasure_loc) if (erasure_loc)
{ {
delete []erasure_loc; delete[] erasure_loc;
erasure_loc = NULL; erasure_loc = NULL;
} }
decoder.Destroy();
if (retrieval_loc)
{
delete[] retrieval_loc;
retrieval_loc = NULL;
}
if (retrieval_pool)
{
for (UINT16 i = 0; i < ndata; i++)
{
if (retrieval_pool[i])
{
delete[] retrieval_pool[i];
retrieval_pool[i] = NULL;
}
}
delete[] retrieval_pool;
retrieval_pool = NULL;
}
NormObject* obj; NormObject* obj;
while ((obj = rx_table.Find(rx_table.RangeLo()))) while ((obj = rx_table.Find(rx_table.RangeLo())))
{ {
@ -701,7 +768,6 @@ NormBlock* NormServerNode::GetFreeBlock(NormObjectId objectId, NormBlockId block
if (!b) if (!b)
{ {
if (session.ClientIsSilent()) if (session.ClientIsSilent())
//if (1)
{ {
// forward iteration to find oldest older object with resources // forward iteration to find oldest older object with resources
NormObjectTable::Iterator iterator(rx_table); NormObjectTable::Iterator iterator(rx_table);
@ -757,13 +823,29 @@ NormBlock* NormServerNode::GetFreeBlock(NormObjectId objectId, NormBlockId block
char* NormServerNode::GetFreeSegment(NormObjectId objectId, NormBlockId blockId) char* NormServerNode::GetFreeSegment(NormObjectId objectId, NormBlockId blockId)
{ {
while (segment_pool.IsEmpty()) if (segment_pool.IsEmpty())
{ {
NormBlock* b = GetFreeBlock(objectId, blockId); // First, try to steal (retrievable) buffered source symbol segments
if (b) NormObjectTable::Iterator iterator(rx_table);
block_pool.Put(b); NormObject* obj;
else while ((obj = iterator.GetNextObject()))
break; {
// This takes source segments only from the "oldest" obj/blk
// (TBD) Should these be from the "newest" obj/blk instead?
if (obj->ReclaimSourceSegments(segment_pool))
break;
}
// Second, if necessary, steal an ordinally "newer" block
// (TBD) we might try to keep the block state, and only
// steal the segment needed?
while (segment_pool.IsEmpty())
{
NormBlock* b = GetFreeBlock(objectId, blockId);
if (b)
block_pool.Put(b);
else
break;
}
} }
return segment_pool.Get(); return segment_pool.Get();
} // end NormServerNode::GetFreeSegment() } // end NormServerNode::GetFreeSegment()
@ -869,7 +951,7 @@ void NormServerNode::HandleObjectMessage(const NormObjectMsg& msg)
else else
{ {
// The hacky use of "sync_id" here keeps the debug message from // The hacky use of "sync_id" here keeps the debug message from
// printing too often. // printing too often while "waiting to sync" ...
if (0 == sync_id) if (0 == sync_id)
{ {
DMSG(0, "NormServerNode::HandleObjectMessage() waiting to sync ...\n"); DMSG(0, "NormServerNode::HandleObjectMessage() waiting to sync ...\n");
@ -1278,7 +1360,7 @@ void NormServerNode::RepairCheck(NormObject::CheckLevel checkLevel,
ExponentialRand(grtt_estimate*backoff_factor, gsize_estimate) : ExponentialRand(grtt_estimate*backoff_factor, gsize_estimate) :
0.0; 0.0;
repair_timer.SetInterval(backoffInterval); repair_timer.SetInterval(backoffInterval);
DMSG(4, "NormServerNode::RepairCheck() node>%lu begin NACK back-off: %lf sec)...\n", DMSG(4, "NormServerNode::RepairCheck() node>%lu begin NACK backoff: %lf sec)...\n",
LocalNodeId(), backoffInterval); LocalNodeId(), backoffInterval);
session.ActivateTimer(repair_timer); session.ActivateTimer(repair_timer);
} }
@ -1305,7 +1387,7 @@ void NormServerNode::RepairCheck(NormObject::CheckLevel checkLevel,
if (rewindDetected) if (rewindDetected)
{ {
repair_timer.Deactivate(); repair_timer.Deactivate();
DMSG(4, "NormServerNode::RepairCheck() node>%lu server rewind detected, ending NACK hold-off ...\n", DMSG(4, "NormServerNode::RepairCheck() node>%lu server rewind detected, ending NACK holdoff ...\n",
LocalNodeId()); LocalNodeId());
RepairCheck(checkLevel, objectId, blockId, segmentId); RepairCheck(checkLevel, objectId, blockId, segmentId);
@ -1356,7 +1438,7 @@ bool NormServerNode::OnRepairTimeout(ProtoTimer& /*theTimer*/)
NormNackMsg* nack = (NormNackMsg*)session.GetMessageFromPool(); NormNackMsg* nack = (NormNackMsg*)session.GetMessageFromPool();
if (!nack) if (!nack)
{ {
DMSG(0, "NormServerNode::OnRepairTimeout() node>%lu Warning! " DMSG(3, "NormServerNode::OnRepairTimeout() node>%lu Warning! "
"message pool empty ...\n", LocalNodeId()); "message pool empty ...\n", LocalNodeId());
repair_timer.Deactivate(); repair_timer.Deactivate();
return false; return false;
@ -1446,7 +1528,7 @@ bool NormServerNode::OnRepairTimeout(ProtoTimer& /*theTimer*/)
{ {
if (0 == nack->PackRepairRequest(req)) if (0 == nack->PackRepairRequest(req))
{ {
DMSG(0, "NormServerNode::OnRepairTimeout() warning: full NACK msg\n"); DMSG(3, "NormServerNode::OnRepairTimeout() warning: full NACK msg\n");
break; break;
} }
nackAppended = true; nackAppended = true;
@ -1489,7 +1571,7 @@ bool NormServerNode::OnRepairTimeout(ProtoTimer& /*theTimer*/)
{ {
if (0 == nack->PackRepairRequest(req)) if (0 == nack->PackRepairRequest(req))
{ {
DMSG(0, "NormServerNode::OnRepairTimeout() warning: full NACK msg\n"); DMSG(3, "NormServerNode::OnRepairTimeout() warning: full NACK msg\n");
break; break;
} }
nackAppended = true; nackAppended = true;
@ -1520,7 +1602,7 @@ bool NormServerNode::OnRepairTimeout(ProtoTimer& /*theTimer*/)
if (0 != nack->PackRepairRequest(req)) if (0 != nack->PackRepairRequest(req))
nackAppended = true; nackAppended = true;
else else
DMSG(0, "NormServerNode::OnRepairTimeout() warning: full NACK msg\n"); DMSG(3, "NormServerNode::OnRepairTimeout() warning: full NACK msg\n");
} }
// Queue NACK for transmission // Queue NACK for transmission
nack->SetServerId(GetId()); nack->SetServerId(GetId());
@ -1535,7 +1617,9 @@ bool NormServerNode::OnRepairTimeout(ProtoTimer& /*theTimer*/)
if (nackAppended) if (nackAppended)
{ {
ASSERT(nack->GetRepairContentLength() > 0); ASSERT(nack->GetRepairContentLength() > 0);
session.QueueMessage(nack); //session.QueueMessage(nack);
session.SendMessage(*nack);
session.ReturnMessageToPool(nack);
nack_count++; nack_count++;
} }
else else
@ -1558,7 +1642,8 @@ bool NormServerNode::OnRepairTimeout(ProtoTimer& /*theTimer*/)
double holdoffInterval = double holdoffInterval =
session.Address().IsMulticast() ? grtt_estimate*(backoff_factor + 2.0) : session.Address().IsMulticast() ? grtt_estimate*(backoff_factor + 2.0) :
grtt_estimate; grtt_estimate;
holdoffInterval = (backoff_factor > 0.0) ? holdoffInterval : 1.01*grtt_estimate; // backoff == 0.0 is a special case
//holdoffInterval = (backoff_factor > 0.0) ? holdoffInterval : 1.0*grtt_estimate;
repair_timer.SetInterval(holdoffInterval); repair_timer.SetInterval(holdoffInterval);
DMSG(4, "NormServerNode::OnRepairTimeout() node>%lu begin NACK hold-off: %lf sec ...\n", DMSG(4, "NormServerNode::OnRepairTimeout() node>%lu begin NACK hold-off: %lf sec ...\n",
@ -1742,7 +1827,7 @@ bool NormServerNode::OnCCTimeout(ProtoTimer& /*theTimer*/)
NormAckMsg* ack = (NormAckMsg*)session.GetMessageFromPool(); NormAckMsg* ack = (NormAckMsg*)session.GetMessageFromPool();
if (!ack) if (!ack)
{ {
DMSG(0, "NormServerNode::OnCCTimeout() node>%lu Warning! " DMSG(3, "NormServerNode::OnCCTimeout() node>%lu warning: "
"message pool empty ...\n", LocalNodeId()); "message pool empty ...\n", LocalNodeId());
if (cc_timer.IsActive()) cc_timer.Deactivate(); if (cc_timer.IsActive()) cc_timer.Deactivate();
return false; return false;
@ -1759,16 +1844,16 @@ bool NormServerNode::OnCCTimeout(ProtoTimer& /*theTimer*/)
ack->SetDestination(GetAddress()); ack->SetDestination(GetAddress());
else else
ack->SetDestination(session.Address()); ack->SetDestination(session.Address());
if (is_clr || is_plr) //if (is_clr || is_plr)
{ {
// Don't rate limit clr or plr reps. // Don't rate-limit feedback messages.
session.SendMessage(*ack); session.SendMessage(*ack);
session.ReturnMessageToPool(ack); session.ReturnMessageToPool(ack);
} }
else //else
{ //{
session.QueueMessage(ack); // session.QueueMessage(ack);
} //}
// Begin cc_timer "holdoff" phase // Begin cc_timer "holdoff" phase
cc_timer.SetInterval(grtt_estimate*backoff_factor); cc_timer.SetInterval(grtt_estimate*backoff_factor);
@ -1802,15 +1887,11 @@ bool NormServerNode::OnAckTimeout(ProtoTimer& /*theTimer*/)
else else
ack->SetDestination(session.Address()); ack->SetDestination(session.Address());
if (is_clr || is_plr) // Don't rate limit feedback messages
session.SendMessage(*ack);
session.ReturnMessageToPool(ack);
if (!is_clr && !is_plr)
{ {
// Don't rate limit clr or plr reps.
session.SendMessage(*ack);
session.ReturnMessageToPool(ack);
}
else
{
session.QueueMessage(ack);
// Install cc feedback holdoff // Install cc feedback holdoff
if (cc_timer.IsActive()) cc_timer.Deactivate(); if (cc_timer.IsActive()) cc_timer.Deactivate();
cc_timer.SetInterval(grtt_estimate*backoff_factor); cc_timer.SetInterval(grtt_estimate*backoff_factor);
@ -1820,7 +1901,7 @@ bool NormServerNode::OnAckTimeout(ProtoTimer& /*theTimer*/)
} }
else else
{ {
DMSG(0, "NormServerNode::OnAckTimeout() warning: message pool exhausted!\n"); DMSG(3, "NormServerNode::OnAckTimeout() warning: message pool exhausted!\n");
} }
return true; return true;
} // end NormServerNode::OnAckTimeout() } // end NormServerNode::OnAckTimeout()

24
common/normNode.h
View File

@ -293,6 +293,7 @@ class NormServerNode : public NormNode
block->EmptyToPool(segment_pool); block->EmptyToPool(segment_pool);
block_pool.Put(block); block_pool.Put(block);
} }
bool SegmentPoolIsEmpty() {return segment_pool.IsEmpty();}
char* GetFreeSegment(NormObjectId objectId, NormBlockId blockId); char* GetFreeSegment(NormObjectId objectId, NormBlockId blockId);
void PutFreeSegment(char* segment) void PutFreeSegment(char* segment)
{segment_pool.Put(segment);} {segment_pool.Put(segment);}
@ -303,7 +304,25 @@ class NormServerNode : public NormNode
erasure_loc[index] = value; erasure_loc[index] = value;
} }
UINT16 GetErasureLoc(UINT16 index) UINT16 GetErasureLoc(UINT16 index)
{return erasure_loc[index];} {
return erasure_loc[index];
}
void SetRetrievalLoc(UINT16 index, UINT16 value)
{
ASSERT(index < ndata);
retrieval_loc[index] = value;
}
UINT16 GetRetrievalLoc(UINT16 index)
{
return retrieval_loc[index];
}
char* GetRetrievalSegment()
{
char* s = retrieval_pool[retrieval_index++];
retrieval_index = (retrieval_index >= ndata) ? 0 : retrieval_index;
return s;
}
UINT16 Decode(char** segmentList, UINT16 numData, UINT16 erasureCount) UINT16 Decode(char** segmentList, UINT16 numData, UINT16 erasureCount)
{ {
return decoder.Decode(segmentList, numData, erasureCount, erasure_loc); return decoder.Decode(segmentList, numData, erasureCount, erasure_loc);
@ -373,6 +392,9 @@ class NormServerNode : public NormNode
NormSegmentPool segment_pool; NormSegmentPool segment_pool;
NormDecoder decoder; NormDecoder decoder;
UINT16* erasure_loc; UINT16* erasure_loc;
UINT16* retrieval_loc;
char** retrieval_pool;
UINT16 retrieval_index;
bool server_active; bool server_active;
ProtoTimer activity_timer; ProtoTimer activity_timer;

504
common/normObject.cpp
View File

@ -899,7 +899,7 @@ bool NormObject::AppendRepairRequest(NormNackMsg& nack,
{ {
if (0 == nack.PackRepairRequest(req)) if (0 == nack.PackRepairRequest(req))
{ {
DMSG(0, "NormObject::AppendRepairRequest() warning: full NACK msg\n"); DMSG(3, "NormObject::AppendRepairRequest() warning: full NACK msg\n");
return requestAppended; return requestAppended;
} }
requestAppended = true; requestAppended = true;
@ -958,7 +958,7 @@ bool NormObject::AppendRepairRequest(NormNackMsg& nack,
{ {
if (0 == nack.PackRepairRequest(req)) if (0 == nack.PackRepairRequest(req))
{ {
DMSG(0, "NormObject::AppendRepairRequest() warning: full NACK msg\n"); DMSG(3, "NormObject::AppendRepairRequest() warning: full NACK msg\n");
return requestAppended; return requestAppended;
} }
} }
@ -1004,7 +1004,7 @@ bool NormObject::AppendRepairRequest(NormNackMsg& nack,
{ {
if (0 == nack.PackRepairRequest(req)) if (0 == nack.PackRepairRequest(req))
{ {
DMSG(0, "NormObject::AppendRepairRequest() warning: full NACK msg\n"); DMSG(3, "NormObject::AppendRepairRequest() warning: full NACK msg\n");
return requestAppended; return requestAppended;
} }
requestAppended = true; requestAppended = true;
@ -1020,7 +1020,7 @@ bool NormObject::AppendRepairRequest(NormNackMsg& nack,
req.AppendRepairItem(transport_id, 0, 0, 0); // (TBD) error check req.AppendRepairItem(transport_id, 0, 0, 0); // (TBD) error check
if (0 == nack.PackRepairRequest(req)) if (0 == nack.PackRepairRequest(req))
{ {
DMSG(0, "NormObject::AppendRepairRequest() warning: full NACK msg\n"); DMSG(3, "NormObject::AppendRepairRequest() warning: full NACK msg\n");
return requestAppended; return requestAppended;
} }
requestAppended = true; requestAppended = true;
@ -1115,22 +1115,12 @@ void NormObject::HandleObjectMessage(const NormObjectMsg& msg,
} }
if (block->IsPending(segmentId)) if (block->IsPending(segmentId))
{ {
// 1) Cache segment in block buffer in case its needed for decoding
char* segment = server->GetFreeSegment(transport_id, blockId);
if (!segment)
{
//DMSG(2, "NormObject::HandleObjectMessage() node>%lu server>%lu obj>%hu "
// "Warning! no free segments ...\n", LocalNodeId(), server->GetId(),
// (UINT16)transport_id);
return;
}
UINT16 segmentLength = data.GetPayloadDataLength(); UINT16 segmentLength = data.GetPayloadDataLength();
if (segmentLength > segment_size) if (segmentLength > segment_size)
{ {
DMSG(0, "NormObject::HandleObjectMessage() node>%lu server>%lu obj>%hu " DMSG(0, "NormObject::HandleObjectMessage() node>%lu server>%lu obj>%hu "
"Error! segment too large ...\n", LocalNodeId(), server->GetId(), "Error! segment too large ...\n", LocalNodeId(), server->GetId(),
(UINT16)transport_id); (UINT16)transport_id);
server->PutFreeSegment(segment);
return; return;
} }
UINT16 payloadLength = data.GetPayloadLength(); UINT16 payloadLength = data.GetPayloadLength();
@ -1140,35 +1130,54 @@ void NormObject::HandleObjectMessage(const NormObjectMsg& msg,
payloadMax = MIN(payloadMax, SIM_PAYLOAD_MAX); payloadMax = MIN(payloadMax, SIM_PAYLOAD_MAX);
payloadLength = MIN(payloadLength, SIM_PAYLOAD_MAX); payloadLength = MIN(payloadLength, SIM_PAYLOAD_MAX);
#endif // SIMULATE #endif // SIMULATE
memcpy(segment, data.GetPayload(), payloadLength);
if (payloadLength < payloadMax) // Is this a source symbol or a parity symbol?
memset(segment+payloadLength, 0, payloadMax-payloadLength); bool isSourceSymbol = (segmentId < numData);
if (data.FlagIsSet(NormObjectMsg::FLAG_MSG_START))
// Try to cache segment in block buffer in case it's needed for decoding
char* segment = (!isSourceSymbol || !server->SegmentPoolIsEmpty()) ?
server->GetFreeSegment(transport_id, blockId) : NULL;
if (segment)
{ {
segment[payloadMax] = NormObjectMsg::FLAG_MSG_START; memcpy(segment, data.GetPayload(), payloadLength);
//DMSG(0, "starting segment payload len>%hu msg_len>%hu\n", if (payloadLength < payloadMax)
// payloadLength, ntohs(*((UINT16*)(segment+8)))) ; memset(segment+payloadLength, 0, payloadMax-payloadLength);
block->AttachSegment(segmentId, segment);
} }
else else
segment[payloadMax] = 0; {
block->AttachSegment(segmentId, segment); //DMSG(2, "NormObject::HandleObjectMessage() node>%lu server>%lu obj>%hu "
// "Warning! no free segments ...\n", LocalNodeId(), server->GetId(),
// (UINT16)transport_id);
if (!isSourceSymbol) return;
}
block->UnsetPending(segmentId); block->UnsetPending(segmentId);
bool objectUpdated = false; bool objectUpdated = false;
// 2) Write segment to object (if it's data) // 2) Write segment to object (if it's source symbol (data))
if (segmentId < numData) if (isSourceSymbol)
{ {
block->DecrementErasureCount(); block->DecrementErasureCount();
if (WriteSegment(blockId, segmentId, segment)) if (WriteSegment(blockId, segmentId, data.GetPayload(),
data.FlagIsSet(NormObjectMsg::FLAG_MSG_START)))
{ {
objectUpdated = true; objectUpdated = true;
// For statistics only (TBD) #ifdef NORM_DEBUG // For statistics only (TBD) #ifdef NORM_DEBUG
server->IncrementRecvGoodput(segmentLength); server->IncrementRecvGoodput(segmentLength);
} }
else
{
DMSG(4, "NormObject::HandleObjectMessage() WriteSegment() error\n");
}
} }
else else
{ {
block->IncrementParityCount(); block->IncrementParityCount();
} }
// 3) Decode block if ready and return to pool // 3) Decode block if ready and return to pool
if (block->ErasureCount() <= block->ParityCount()) if (block->ErasureCount() <= block->ParityCount())
{ {
@ -1177,36 +1186,56 @@ void NormObject::HandleObjectMessage(const NormObjectMsg& msg,
DMSG(8, "NormObject::HandleObjectMessage() node>%lu server>%lu obj>%hu blk>%lu " DMSG(8, "NormObject::HandleObjectMessage() node>%lu server>%lu obj>%hu blk>%lu "
"completed block ...\n", LocalNodeId(), server->GetId(), "completed block ...\n", LocalNodeId(), server->GetId(),
(UINT16)transport_id, (UINT32)block->GetId()); (UINT16)transport_id, (UINT32)block->GetId());
UINT16 erasureCount = 0; UINT16 erasureCount = 0;
UINT16 nextErasure = 0; UINT16 nextErasure = 0;
UINT16 retrievalCount = 0;
if (block->GetFirstPending(nextErasure)) if (block->GetFirstPending(nextErasure))
{ {
// Is the block missing _any_ source symbols?
if (nextErasure < numData) if (nextErasure < numData)
{ {
do // Use "NormObject::RetrieveSegment() method to "retrieve"
// source symbol segments already received which weren't cached.
for (UINT16 nextSegment = 0; nextSegment < numData; nextSegment++)
{ {
server->SetErasureLoc(erasureCount++, nextErasure); if (block->IsPending(nextSegment))
if (nextErasure < numData)
{ {
if (!(segment = server->GetFreeSegment(transport_id, blockId))) server->SetErasureLoc(erasureCount++, nextSegment);
{ segment = server->GetRetrievalSegment();
//DMSG(2, "NormObject::HandleObjectMessage() node>%lu server>%lu obj>%hu " ASSERT(segment);
// "Warning! no free segments ...\n", LocalNodeId(), server->GetId(),
// (UINT16)transport_id);
// (TBD) Dump the block ...???
return;
}
UINT16 payloadMax = segment_size + NormDataMsg::GetStreamPayloadHeaderLength(); UINT16 payloadMax = segment_size + NormDataMsg::GetStreamPayloadHeaderLength();
#ifdef SIMULATE #ifdef SIMULATE
payloadMax = MIN(payloadMax, SIM_PAYLOAD_MAX); payloadMax = MIN(payloadMax, SIM_PAYLOAD_MAX);
#endif // SIMULATE #endif // SIMULATE
// Zeroize the missing segment payload in prep for decoding
memset(segment, 0, payloadMax+1); memset(segment, 0, payloadMax+1);
block->SetSegment(nextErasure, segment); block->SetSegment(nextSegment, segment);
} }
nextErasure++; else if (!block->GetSegment(nextSegment))
} while (block->GetNextPending(nextErasure)); {
} if (!(segment = RetrieveSegment(blockId, nextSegment)))
} {
ASSERT(IsStream());
block->SetPending(nextSegment);
block->IncrementErasureCount();
// Clear any erasure/retrieval segments
for (UINT16 i = 0; i < erasureCount; i++)
block->DetachSegment(server->GetErasureLoc(i));
for (UINT16 i = 0; i < retrievalCount; i++)
block->DetachSegment(server->GetRetrievalLoc(i));
return;
}
server->SetRetrievalLoc(retrievalCount++, nextSegment);
block->SetSegment(nextSegment, segment);
}
}
nextErasure = numData;
// Set erasure locs for any missing parity symbol segments
while (block->GetNextPending(nextErasure))
server->SetErasureLoc(erasureCount++, nextErasure++);
} // end if (nextErasure < numData)
} // end (block->GetFirstPending(nextErasure))
if (erasureCount) if (erasureCount)
{ {
server->Decode(block->SegmentList(), numData, erasureCount); server->Decode(block->SegmentList(), numData, erasureCount);
@ -1215,12 +1244,13 @@ void NormObject::HandleObjectMessage(const NormObjectMsg& msg,
NormSegmentId sid = server->GetErasureLoc(i); NormSegmentId sid = server->GetErasureLoc(i);
if (sid < numData) if (sid < numData)
{ {
if (WriteSegment(blockId, sid, block->Segment(sid))) if (WriteSegment(blockId, sid, block->GetSegment(sid), false))
{ {
objectUpdated = true; objectUpdated = true;
// For statistics only (TBD) #ifdef NORM_DEBUG // For statistics only (TBD) #ifdef NORM_DEBUG
server->IncrementRecvGoodput(segmentLength); server->IncrementRecvGoodput(segmentLength);
} }
block->DetachSegment(sid);
} }
else else
{ {
@ -1228,6 +1258,9 @@ void NormObject::HandleObjectMessage(const NormObjectMsg& msg,
} }
} }
} }
// Clear any temporarily retrieved (non-cached) segments for the block
for (UINT16 i = 0; i < retrievalCount; i++)
block->DetachSegment(server->GetRetrievalLoc(i));
// OK, we're done with this block // OK, we're done with this block
pending_mask.Unset(blockId); pending_mask.Unset(blockId);
block_buffer.Remove(block); block_buffer.Remove(block);
@ -1258,6 +1291,28 @@ void NormObject::HandleObjectMessage(const NormObjectMsg& msg,
} // end if/else (NORM_MSG_INFO) } // end if/else (NORM_MSG_INFO)
} // end NormObject::HandleObjectMessage() } // end NormObject::HandleObjectMessage()
// Returns source symbol segments to pool for first block with such resources
bool NormObject::ReclaimSourceSegments(NormSegmentPool& segmentPool)
{
NormBlockBuffer::Iterator iterator(block_buffer);
NormBlock* block;
while ((block = iterator.GetNextBlock()))
{
bool reclaimed = false;
UINT16 numData = GetBlockSize(block->GetId());
for (UINT16 i = 0; i < numData; i++)
{
char* s = block->DetachSegment(i);
if (s)
{
segmentPool.Put(s);
reclaimed = true;
}
}
if (reclaimed) return true;
}
return false;
} // end NormObject::ReclaimSourceSegments()
// Steals non-pending block (oldest first) for server resource management // Steals non-pending block (oldest first) for server resource management
@ -1399,7 +1454,7 @@ bool NormObject::NextServerMsg(NormObjectMsg* msg)
{ {
if (!(block = session.ServerGetFreeBlock(transport_id, blockId))) if (!(block = session.ServerGetFreeBlock(transport_id, blockId)))
{ {
DMSG(2, "NormObject::NextServerMsg() node>%lu Warning! server resource " DMSG(2, "NormObject::NextServerMsg() node>%lu warning: server resource "
"constrained (no free blocks).\n", LocalNodeId()); "constrained (no free blocks).\n", LocalNodeId());
return false; return false;
} }
@ -1419,7 +1474,7 @@ bool NormObject::NextServerMsg(NormObjectMsg* msg)
} }
else else
{ {
DMSG(12, "NormObject::NextServerMsg() node>%lu Warning! server resource " DMSG(2, "NormObject::NextServerMsg() node>%lu warning: server resource "
"constrained (no free segments).\n", LocalNodeId()); "constrained (no free segments).\n", LocalNodeId());
session.ServerPutFreeBlock(block); session.ServerPutFreeBlock(block);
return false; return false;
@ -1461,7 +1516,8 @@ bool NormObject::NextServerMsg(NormObjectMsg* msg)
{ {
// Try to read data segment (Note "ReadSegment" copies in offset/length info also) // Try to read data segment (Note "ReadSegment" copies in offset/length info also)
char* buffer = data->AccessPayload(); char* buffer = data->AccessPayload();
UINT16 payloadLength = ReadSegment(blockId, segmentId, buffer); bool msgStart = false;
UINT16 payloadLength = ReadSegment(blockId, segmentId, buffer, &msgStart);
if (0 == payloadLength) if (0 == payloadLength)
{ {
// (TBD) deal with read error // (TBD) deal with read error
@ -1472,23 +1528,16 @@ bool NormObject::NextServerMsg(NormObjectMsg* msg)
} }
data->SetPayloadLength(payloadLength); data->SetPayloadLength(payloadLength);
if (IsStream()) // Look for FLAG_MSG_START
if (msgStart)
{ {
// Look for FLAG_MSG_START data->SetFlag(NormObjectMsg::FLAG_MSG_START);
UINT16 payloadMax = segment_size+NormDataMsg::GetStreamPayloadHeaderLength(); //DMSG(0, "read start segment len:%hu\n", ntohs(*((UINT16*)(buffer+2))));
#ifdef SIMULATE //DMSG(0, " message len:%hu\n", ntohs(*((UINT16*)(buffer+8))));
payloadMax = MIN(payloadMax, SIM_PAYLOAD_MAX); }
#endif // SIMULATE else
if (buffer[payloadMax]) {
{ //DMSG(0, "non-start segment read ...\n");
data->SetFlag(NormObjectMsg::FLAG_MSG_START);
//DMSG(0, "read start segment len:%hu\n", ntohs(*((UINT16*)(buffer+2))));
//DMSG(0, " message len:%hu\n", ntohs(*((UINT16*)(buffer+8))));
}
else
{
//DMSG(0, "non-start segment read ...\n");
}
} }
// Perform incremental FEC encoding as needed // Perform incremental FEC encoding as needed
@ -1516,7 +1565,7 @@ bool NormObject::NextServerMsg(NormObjectMsg* msg)
CalculateBlockParity(block); CalculateBlockParity(block);
} }
ASSERT(block->ParityReady(numData)); ASSERT(block->ParityReady(numData));
char* segment = block->Segment(segmentId); char* segment = block->GetSegment(segmentId);
ASSERT(segment); ASSERT(segment);
UINT16 payloadLength = segment_size; UINT16 payloadLength = segment_size;
if (IsStream()) payloadLength += NormDataMsg::GetStreamPayloadHeaderLength(); if (IsStream()) payloadLength += NormDataMsg::GetStreamPayloadHeaderLength();
@ -1692,7 +1741,7 @@ bool NormFileObject::Open(const char* thePath,
} }
else else
{ {
if (file.Open(thePath, O_WRONLY | O_CREAT | O_TRUNC)) if (file.Open(thePath, O_RDWR | O_CREAT | O_TRUNC))
{ {
file.Lock(); file.Lock();
} }
@ -1769,7 +1818,8 @@ void NormFileObject::Close()
bool NormFileObject::WriteSegment(NormBlockId blockId, bool NormFileObject::WriteSegment(NormBlockId blockId,
NormSegmentId segmentId, NormSegmentId segmentId,
const char* buffer) const char* buffer,
bool /*msgStart*/)
{ {
UINT16 len; UINT16 len;
if (blockId == final_block_id) if (blockId == final_block_id)
@ -1809,7 +1859,8 @@ bool NormFileObject::WriteSegment(NormBlockId blockId,
UINT16 NormFileObject::ReadSegment(NormBlockId blockId, UINT16 NormFileObject::ReadSegment(NormBlockId blockId,
NormSegmentId segmentId, NormSegmentId segmentId,
char* buffer) char* buffer,
bool* /*msgStart*/)
{ {
// Determine segment length from blockId::segmentId // Determine segment length from blockId::segmentId
UINT16 len; UINT16 len;
@ -1851,6 +1902,33 @@ UINT16 NormFileObject::ReadSegment(NormBlockId blockId,
return (len == nbytes) ? len : 0; return (len == nbytes) ? len : 0;
} // end NormFileObject::ReadSegment() } // end NormFileObject::ReadSegment()
char* NormFileObject::RetrieveSegment(NormBlockId blockId,
NormSegmentId segmentId)
{
if (server)
{
char* segment = server->GetRetrievalSegment();
UINT16 len = ReadSegment(blockId, segmentId, segment);
if (len)
{
// zeroize remainder for proper decodes
if (len < segment_size)
memset(segment+len, 0, segment_size-len);
return segment;
}
else
{
DMSG(0, "NormFileObject::RetrieveSegment() error reading segment\n");
return NULL;
}
}
else
{
DMSG(0, "NormFileObject::RetrieveSegment() error: NULL server!\n");
return NULL;
}
} // end NormFileObject::RetrieveSegment()
///////////////////////////////////////////////////////////////// /////////////////////////////////////////////////////////////////
// //
// NormDataObject Implementation // NormDataObject Implementation
@ -1924,7 +2002,8 @@ void NormDataObject::Close()
bool NormDataObject::WriteSegment(NormBlockId blockId, bool NormDataObject::WriteSegment(NormBlockId blockId,
NormSegmentId segmentId, NormSegmentId segmentId,
const char* buffer) const char* buffer,
bool /*msgStart*/)
{ {
UINT16 len; UINT16 len;
if (blockId == final_block_id) if (blockId == final_block_id)
@ -1964,7 +2043,8 @@ bool NormDataObject::WriteSegment(NormBlockId blockId,
UINT16 NormDataObject::ReadSegment(NormBlockId blockId, UINT16 NormDataObject::ReadSegment(NormBlockId blockId,
NormSegmentId segmentId, NormSegmentId segmentId,
char* buffer) char* buffer,
bool* /*msgStart*/)
{ {
// Determine segment length from blockId::segmentId // Determine segment length from blockId::segmentId
UINT16 len; UINT16 len;
@ -2000,6 +2080,58 @@ UINT16 NormDataObject::ReadSegment(NormBlockId blockId,
return true; return true;
} // end NormDataObject::ReadSegment() } // end NormDataObject::ReadSegment()
char* NormDataObject::RetrieveSegment(NormBlockId blockId,
NormSegmentId segmentId)
{
// Determine segment length from blockId::segmentId
UINT16 len;
if (blockId == final_block_id)
{
if (segmentId == (GetBlockSize(blockId)-1))
len = final_segment_size;
else
len = segment_size;
}
else
{
len = segment_size;
}
if (len < segment_size)
{
if (server)
{
char* segment = server->GetRetrievalSegment();
ReadSegment(blockId, segmentId, segment);
memset(segment+len, 0, segment_size-len);
return segment;
}
else
{
DMSG(0, "NormDataObject::RetrieveSegment() error: NULL server!\n");
return NULL;
}
}
else
{
// Determine segment offset from blockId::segmentId
NormObjectSize segmentOffset;
NormObjectSize segmentSize = NormObjectSize(segment_size);
if ((UINT32)blockId < large_block_count)
{
segmentOffset = large_block_length*(UINT32)blockId + segmentSize*segmentId;
}
else
{
segmentOffset = large_block_length*large_block_count; // (TBD) pre-calc this
UINT32 smallBlockIndex = (UINT32)blockId - large_block_count;
segmentOffset = segmentOffset + small_block_length*smallBlockIndex +
segmentSize*segmentId;
}
ASSERT(0 == segmentOffset.MSB());
ASSERT(data_max >= (segmentOffset.LSB() + len));
return (data_ptr + segmentOffset.LSB());
}
} // end NormDataObject::RetrieveSegment()
///////////////////////////////////////////////////////////////// /////////////////////////////////////////////////////////////////
// //
@ -2012,7 +2144,8 @@ NormStreamObject::NormStreamObject(class NormSession& theSession,
: NormObject(STREAM, theSession, theServer, objectId), : NormObject(STREAM, theSession, theServer, objectId),
stream_sync(false), sync_offset_valid(false), stream_sync(false), sync_offset_valid(false),
flush_pending(false), msg_start(true), flush_pending(false), msg_start(true),
flush_mode(FLUSH_NONE), push_mode(false) flush_mode(FLUSH_NONE), push_mode(false),
block_pool_threshold(0)
{ {
} }
@ -2044,14 +2177,14 @@ bool NormStreamObject::Open(UINT32 bufferSize,
numData = session.ServerBlockSize(); numData = session.ServerBlockSize();
} }
NormObjectSize blockSize((UINT32)segmentSize * (UINT32)numData); UINT32 blockSize = segmentSize * numData;
NormObjectSize numBlocks = NormObjectSize((UINT32)bufferSize) / blockSize; UINT32 numBlocks = bufferSize / blockSize;
ASSERT(0 == numBlocks.MSB());
// Buffering requires at least 2 blocks // Buffering requires at least 2 blocks
numBlocks = MAX(2, numBlocks.LSB()); numBlocks = MAX(2, numBlocks);
UINT32 numSegments = numBlocks.LSB() * numData; UINT32 numSegments = numBlocks * numData;
if (!block_pool.Init(numBlocks.LSB(), numData)) if (!block_pool.Init(numBlocks, numData))
{ {
DMSG(0, "NormStreamObject::Open() block_pool init error\n"); DMSG(0, "NormStreamObject::Open() block_pool init error\n");
Close(); Close();
@ -2065,17 +2198,19 @@ bool NormStreamObject::Open(UINT32 bufferSize,
return false; return false;
} }
if (!stream_buffer.Init(numBlocks.LSB())) if (!stream_buffer.Init(numBlocks))
{ {
DMSG(0, "NormStreamObject::Open() stream_buffer init error\n"); DMSG(0, "NormStreamObject::Open() stream_buffer init error\n");
Close(); Close();
return false; return false;
} }
// (TBD) we really only need one set of indexes // (TBD) we really only need one set of indexes & offset
// since our objects are exclusively read _or_ write // since our objects are exclusively read _or_ write
write_index.block = write_index.segment = 0; read_init = true;
read_index.block = read_index.segment = 0; read_index.block = read_index.segment = 0;
write_index.block = write_index.segment = 0;
write_offset = read_offset = 0; write_offset = read_offset = 0;
if (!server) if (!server)
{ {
if (!NormObject::Open(NormObjectSize((UINT32)bufferSize), if (!NormObject::Open(NormObjectSize((UINT32)bufferSize),
@ -2240,16 +2375,32 @@ bool NormStreamObject::StreamUpdateStatus(NormBlockId blockId)
stream_sync = true; stream_sync = true;
stream_sync_id = blockId; stream_sync_id = blockId;
stream_next_id = blockId + pending_mask.Size(); stream_next_id = blockId + pending_mask.Size();
read_index.block = blockId; read_init = true;
read_index.segment = 0;
return true; return true;
} }
} // end NormStreamObject::StreamUpdateStatus() } // end NormStreamObject::StreamUpdateStatus()
char* NormStreamObject::RetrieveSegment(NormBlockId blockId,
NormSegmentId segmentId)
{
NormBlock* block = stream_buffer.Find(blockId);
if (!block)
{
DMSG(0, "NormStreamObject::RetrieveSegment() segment block unavailable\n");
return NULL;
}
char* segment = block->GetSegment(segmentId);
if (NULL == segment)
DMSG(0, "NormStreamObject::RetrieveSegment() segment unavailable\n");
return segment;
} // end NormStreamObject::RetrieveSegment()
UINT16 NormStreamObject::ReadSegment(NormBlockId blockId, UINT16 NormStreamObject::ReadSegment(NormBlockId blockId,
NormSegmentId segmentId, NormSegmentId segmentId,
char* buffer) char* buffer,
bool* msgStart)
{ {
// (TBD) compare blockId with stream_buffer.RangeLo() and stream_buffer.RangeHi() // (TBD) compare blockId with stream_buffer.RangeLo() and stream_buffer.RangeHi()
NormBlock* block = stream_buffer.Find(blockId); NormBlock* block = stream_buffer.Find(blockId);
@ -2266,7 +2417,7 @@ UINT16 NormStreamObject::ReadSegment(NormBlockId blockId,
return false; return false;
} }
block->UnsetPending(segmentId); block->UnsetPending(segmentId);
char* segment = block->Segment(segmentId); char* segment = block->GetSegment(segmentId);
ASSERT(segment != NULL); ASSERT(segment != NULL);
UINT16 segmentLength = NormDataMsg::ReadStreamPayloadLength(segment); UINT16 segmentLength = NormDataMsg::ReadStreamPayloadLength(segment);
ASSERT(segmentLength <= segment_size); ASSERT(segmentLength <= segment_size);
@ -2279,29 +2430,41 @@ UINT16 NormStreamObject::ReadSegment(NormBlockId blockId,
#else #else
memcpy(buffer, segment, payloadLength); memcpy(buffer, segment, payloadLength);
#endif // SIMULATE #endif // SIMULATE
buffer[payloadMax] = segment[payloadMax]; if (msgStart)
*msgStart = (0 != (NormDataMsg::FLAG_MSG_START & segment[payloadMax]));
return payloadLength; return payloadLength;
} // end NormStreamObject::ReadSegment() } // end NormStreamObject::ReadSegment()
bool NormStreamObject::WriteSegment(NormBlockId blockId, bool NormStreamObject::WriteSegment(NormBlockId blockId,
NormSegmentId segmentId, NormSegmentId segmentId,
const char* segment) const char* segment,
bool msgStart)
{ {
UINT32 segmentOffset = NormDataMsg::ReadStreamPayloadOffset(segment);
if (read_init)
{
read_init = false;
read_index.block = blockId;
read_index.segment = segmentId;
read_offset = segmentOffset;
}
if ((blockId < read_index.block) || if ((blockId < read_index.block) ||
((blockId == read_index.block) && ((blockId == read_index.block) &&
(segmentId < read_index.segment))) (segmentId < read_index.segment)))
{ {
DMSG(4, "NormStreamObject::WriteSegment() block/segment < read_index!?\n"); DMSG(4, "NormStreamObject::WriteSegment() block/segment < read_index!?\n");
return false; return false;
} }
UINT32 segmentOffset = NormDataMsg::ReadStreamPayloadOffset(segment);
// if (segmentOffset < read_offset) // if (segmentOffset < read_offset)
UINT32 diff = segmentOffset - read_offset; UINT32 diff = segmentOffset - read_offset;
if ((diff > 0x80000000) || ((0x80000000 == diff) && (segmentOffset > read_offset))) if ((diff > 0x80000000) || ((0x80000000 == diff) && (segmentOffset > read_offset)))
{ {
DMSG(4, "NormStreamObject::WriteSegment() diff:%lu segmentOffset:%lu < read_offset:%lu \n", DMSG(0, "NormStreamObject::WriteSegment() diff:%lu segmentOffset:%lu < read_offset:%lu \n",
diff, segmentOffset, read_offset); diff, segmentOffset, read_offset);
ASSERT(0);
return false; return false;
} }
@ -2309,22 +2472,25 @@ bool NormStreamObject::WriteSegment(NormBlockId blockId,
if (!block) if (!block)
{ {
bool broken = false; bool broken = false;
// Prune (if necessary) stream_buffer (stream might be broken) bool dataLost = false;
while (!stream_buffer.CanInsert(blockId) || block_pool.IsEmpty()) while (block_pool.IsEmpty() || !stream_buffer.CanInsert(blockId))
{ {
block = stream_buffer.Find(stream_buffer.RangeLo()); block = stream_buffer.Find(stream_buffer.RangeLo());
if (block->IsPending()) broken = true; ASSERT(block);
// This loop feeds any received user data segments to the application if (blockId < block->GetId())
// (if the application doesn't want the data, it's lost!) {
bool dataLost = false; DMSG(4, "NormStreamObject::WriteSegment() blockId too old!?\n");
return false;
}
while (block->IsPending()) while (block->IsPending())
{ {
// Notify app for stream data salvage broken = true;
// Force read_index forward, giving app a chance to read data
read_index.block = block->GetId(); read_index.block = block->GetId();
block->GetFirstPending(read_index.segment); block->GetFirstPending(read_index.segment);
NormBlock* tempBlock = block; NormBlock* tempBlock = block;
UINT32 tempOffset = read_offset; UINT32 tempOffset = read_offset;
session.Notify(NormController::RX_OBJECT_UPDATE, server, this); session.Notify(NormController::RX_OBJECT_UPDATE, server, this);
block = stream_buffer.Find(stream_buffer.RangeLo()); block = stream_buffer.Find(stream_buffer.RangeLo());
if (tempBlock == block) if (tempBlock == block)
{ {
@ -2333,48 +2499,57 @@ bool NormStreamObject::WriteSegment(NormBlockId blockId,
// App didn't want any data here, purge segment // App didn't want any data here, purge segment
//ASSERT(0); //ASSERT(0);
dataLost = true; dataLost = true;
char* s = block->DetachSegment(read_index.segment); block->UnsetPending(read_index.segment++);
segment_pool.Put(s); if (read_index.segment >= ndata)
block->UnsetPending(read_index.segment); {
read_index.block++;
read_index.segment = 0;
stream_buffer.Remove(block);
block->EmptyToPool(segment_pool);
block_pool.Put(block);
block = NULL;
Prune(read_index.block);
break;
}
} }
} }
else else
{ {
// App consumed block via Read() // App consumed block via Read()
block = NULL; block = NULL;
break; break;
} }
} } // end while (block->IsPending())
if (dataLost)
DMSG(0, "NormStreamObject::WriteSegment() broken stream data dropped!\n");
if (block) if (block)
{ {
// if the app didn't consume the block, we must if (block->GetId() == read_index.block)
// return it to the pool {
ASSERT(!block->IsPending()); read_index.block++;
read_index.segment = 0;
Prune(read_index.block);
}
stream_buffer.Remove(block); stream_buffer.Remove(block);
block->EmptyToPool(segment_pool); block->EmptyToPool(segment_pool);
block_pool.Put(block); block_pool.Put(block);
} }
} } // end while (block_pool.IsEmpty() || !stream_buffer.CanInsert(blockId))
block = block_pool.Get();
block->SetId(blockId);
block->ClearPending();
bool success = stream_buffer.Insert(block);
ASSERT(success);
if (broken) if (broken)
{ {
DMSG(4, "NormStreamObject::WriteSegment() node>%lu obj>%hu blk>%lu seg>%hu broken stream ...\n", DMSG(4, "NormStreamObject::WriteSegment() node>%lu obj>%hu blk>%lu seg>%hu broken stream ...\n",
LocalNodeId(), (UINT16)transport_id, (UINT32)blockId, (UINT16)segmentId); LocalNodeId(), (UINT16)transport_id, (UINT32)blockId, (UINT16)segmentId);
NormBlock* first = stream_buffer.Find(stream_buffer.RangeLo()); if (dataLost)
read_index.block = first->GetId(); DMSG(0, "NormStreamObject::WriteSegment() broken stream data not read by app!\n");
read_index.segment = 0; }
} block = block_pool.Get();
} block->SetId(blockId);
block->ClearPending();
ASSERT(blockId >= read_index.block);
bool success = stream_buffer.Insert(block);
ASSERT(success);
} // end if (!block)
// Make sure this segment hasn't already been written // Make sure this segment hasn't already been written
if(!block->Segment(segmentId)) if(!block->GetSegment(segmentId))
{ {
char* s = segment_pool.Get(); char* s = segment_pool.Get();
ASSERT(s != NULL); // for now, this should always succeed ASSERT(s != NULL); // for now, this should always succeed
@ -2385,11 +2560,12 @@ bool NormStreamObject::WriteSegment(NormBlockId blockId,
payloadLength = MIN(payloadMax, payloadLength); payloadLength = MIN(payloadMax, payloadLength);
#endif // SIMULATE #endif // SIMULATE
memcpy(s, segment, payloadLength); memcpy(s, segment, payloadLength);
s[payloadMax] = segment[payloadMax]; // Store "msgStart" info at end of payload space
s[payloadMax] = msgStart ? NormDataMsg::FLAG_MSG_START : 0;
block->AttachSegment(segmentId, s); block->AttachSegment(segmentId, s);
block->SetPending(segmentId); block->SetPending(segmentId);
ASSERT(block->Segment(segmentId) == s); ASSERT(block->GetSegment(segmentId) == s);
} }
if (!sync_offset_valid) if (!sync_offset_valid)
{ {
@ -2428,8 +2604,10 @@ void NormStreamObject::Prune(NormBlockId blockId)
pending_mask.UnsetBits(firstId, count); pending_mask.UnsetBits(firstId, count);
} }
} }
bool result = StreamUpdateStatus(blockId);
ASSERT(result);
if (resync) server->IncrementResyncCount(); if (resync) server->IncrementResyncCount();
StreamUpdateStatus(blockId); result = GetFirstPending(firstId);
} // end NormStreamObject::Prune() } // end NormStreamObject::Prune()
// Sequential (in order) read/write routines (TBD) Add a "Seek()" method // Sequential (in order) read/write routines (TBD) Add a "Seek()" method
@ -2447,20 +2625,62 @@ bool NormStreamObject::Read(char* buffer, unsigned int* buflen, bool findMsgStar
// DMSG(0, "NormStreamObject::Read() stream buffer empty (1) (sbEmpty:%d)\n", stream_buffer.IsEmpty()); // DMSG(0, "NormStreamObject::Read() stream buffer empty (1) (sbEmpty:%d)\n", stream_buffer.IsEmpty());
*buflen = bytesRead; *buflen = bytesRead;
if (bytesRead > 0) if (bytesRead > 0)
{
return true; return true;
}
else else
return findMsgStart ? false : true; {
if (block_pool.GetCount() < block_pool_threshold)
{
// Force read_index forward and try again.
if (++read_index.segment >= ndata)
{
read_index.block++;
read_index.segment = 0;
Prune(read_index.block);
}
continue;
}
else
{
return findMsgStart ? false : true;
}
}
} }
char* segment = block->Segment(read_index.segment); char* segment = block->GetSegment(read_index.segment);
ASSERT(!segment || block->IsPending(read_index.segment));
if (!segment) if (!segment)
{ {
//DMSG(0, "NormStreamObject::Read(%lu:%hu) stream buffer empty (2)\n", //DMSG(0, "NormStreamObject::Read(%lu:%hu) stream buffer empty (2)\n",
// (UINT32)read_index.block, read_index.segment); // (UINT32)read_index.block, read_index.segment);
*buflen = bytesRead; *buflen = bytesRead;
if (bytesRead > 0) if (bytesRead > 0)
{
return true; return true;
}
else else
return findMsgStart ? false : true; {
if (block_pool.GetCount() < block_pool_threshold)
{
// Force read_index forward and try again.
if (++read_index.segment >= ndata)
{
stream_buffer.Remove(block);
block->EmptyToPool(segment_pool);
block_pool.Put(block);
read_index.block++;
read_index.segment = 0;
Prune(read_index.block);
}
continue;
}
else
{
return findMsgStart ? false : true;
}
}
} }
UINT32 segmentOffset = NormDataMsg::ReadStreamPayloadOffset(segment); UINT32 segmentOffset = NormDataMsg::ReadStreamPayloadOffset(segment);
@ -2509,8 +2729,10 @@ bool NormStreamObject::Read(char* buffer, unsigned int* buflen, bool findMsgStar
} }
if (!msgStart) if (!msgStart)
{ {
block->DetachSegment(read_index.segment); // Don't bother managing individual segments since
segment_pool.Put(segment); // stream buffers are exact multiples of block size!
//block->DetachSegment(read_index.segment);
//segment_pool.Put(segment);
block->UnsetPending(read_index.segment++); block->UnsetPending(read_index.segment++);
if (read_index.segment >= ndata) if (read_index.segment >= ndata)
{ {
@ -2519,6 +2741,7 @@ bool NormStreamObject::Read(char* buffer, unsigned int* buflen, bool findMsgStar
block_pool.Put(block); block_pool.Put(block);
read_index.block++; read_index.block++;
read_index.segment = 0; read_index.segment = 0;
Prune(read_index.block);
} }
continue; continue;
} }
@ -2538,8 +2761,10 @@ bool NormStreamObject::Read(char* buffer, unsigned int* buflen, bool findMsgStar
bytesToRead -= count; bytesToRead -= count;
if (index >= length) if (index >= length)
{ {
block->DetachSegment(read_index.segment); // Don't bother managing individual segments since
segment_pool.Put(segment); // stream buffers are multiples of block size!
// block->DetachSegment(read_index.segment);
// segment_pool.Put(segment);
block->UnsetPending(read_index.segment++); block->UnsetPending(read_index.segment++);
if (read_index.segment >= ndata) if (read_index.segment >= ndata)
{ {
@ -2548,6 +2773,7 @@ bool NormStreamObject::Read(char* buffer, unsigned int* buflen, bool findMsgStar
block_pool.Put(block); block_pool.Put(block);
read_index.block++; read_index.block++;
read_index.segment = 0; read_index.segment = 0;
Prune(read_index.block);
} }
} }
} while (bytesToRead > 0); // end while (len > 0) } while (bytesToRead > 0); // end while (len > 0)
@ -2555,6 +2781,7 @@ bool NormStreamObject::Read(char* buffer, unsigned int* buflen, bool findMsgStar
return true; return true;
} // end NormStreamObject::Read() } // end NormStreamObject::Read()
UINT32 NormStreamObject::Write(const char* buffer, UINT32 len, bool eom) UINT32 NormStreamObject::Write(const char* buffer, UINT32 len, bool eom)
{ {
UINT32 nBytes = 0; UINT32 nBytes = 0;
@ -2598,7 +2825,7 @@ UINT32 NormStreamObject::Write(const char* buffer, UINT32 len, bool eom)
ASSERT(success); ASSERT(success);
} }
char* segment = block->Segment(write_index.segment); char* segment = block->GetSegment(write_index.segment);
if (!segment) if (!segment)
{ {
if (!(segment = segment_pool.Get())) if (!(segment = segment_pool.Get()))
@ -2729,7 +2956,8 @@ NormSimObject::~NormSimObject()
UINT16 NormSimObject::ReadSegment(NormBlockId blockId, UINT16 NormSimObject::ReadSegment(NormBlockId blockId,
NormSegmentId segmentId, NormSegmentId segmentId,
char* buffer) char* buffer,
bool* /*msgStart*/)
{ {
// Determine segment length // Determine segment length
UINT16 len; UINT16 len;

63
common/normObject.h
View File

@ -84,10 +84,15 @@ class NormObject
virtual bool WriteSegment(NormBlockId blockId, virtual bool WriteSegment(NormBlockId blockId,
NormSegmentId segmentId, NormSegmentId segmentId,
const char* buffer) = 0; const char* buffer,
bool msgStart) = 0;
virtual UINT16 ReadSegment(NormBlockId blockId, virtual UINT16 ReadSegment(NormBlockId blockId,
NormSegmentId segmentId, NormSegmentId segmentId,
char* buffer) = 0; char* buffer,
bool* msgStart = NULL) = 0;
virtual char* RetrieveSegment(NormBlockId blockId,
NormSegmentId segmentId) = 0;
NackingMode GetNackingMode() const {return nacking_mode;} NackingMode GetNackingMode() const {return nacking_mode;}
void SetNackingMode(NackingMode nackingMode) void SetNackingMode(NackingMode nackingMode)
@ -201,6 +206,7 @@ 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 ReclaimSourceSegments(NormSegmentPool& segmentPool);
bool PassiveRepairCheck(NormBlockId blockId, bool PassiveRepairCheck(NormBlockId blockId,
NormSegmentId segmentId); NormSegmentId segmentId);
bool ClientRepairCheck(CheckLevel level, bool ClientRepairCheck(CheckLevel level,
@ -287,11 +293,15 @@ class NormFileObject : public NormObject
virtual bool WriteSegment(NormBlockId blockId, virtual bool WriteSegment(NormBlockId blockId,
NormSegmentId segmentId, NormSegmentId segmentId,
const char* buffer); const char* buffer,
bool msgStart);
virtual UINT16 ReadSegment(NormBlockId blockId, virtual UINT16 ReadSegment(NormBlockId blockId,
NormSegmentId segmentId, NormSegmentId segmentId,
char* buffer); char* buffer,
bool* msgStart = NULL);
virtual char* RetrieveSegment(NormBlockId blockId,
NormSegmentId segmentId);
private: private:
char path[PATH_MAX]; char path[PATH_MAX];
@ -320,11 +330,16 @@ class NormDataObject : public NormObject
virtual bool WriteSegment(NormBlockId blockId, virtual bool WriteSegment(NormBlockId blockId,
NormSegmentId segmentId, NormSegmentId segmentId,
const char* buffer); const char* buffer,
bool msgStart);
virtual UINT16 ReadSegment(NormBlockId blockId, virtual UINT16 ReadSegment(NormBlockId blockId,
NormSegmentId segmentId, NormSegmentId segmentId,
char* buffer); char* buffer,
bool* msgStart = NULL);
virtual char* RetrieveSegment(NormBlockId blockId,
NormSegmentId segmentId);
private: private:
NormObjectSize large_block_length; NormObjectSize large_block_length;
@ -359,7 +374,7 @@ class NormStreamObject : public NormObject
void Flush(bool eom = false) void Flush(bool eom = false)
{ {
FlushMode oldFlushMode = flush_mode; FlushMode oldFlushMode = flush_mode;
SetFlushMode(FLUSH_ACTIVE); SetFlushMode((FLUSH_ACTIVE == oldFlushMode) ? FLUSH_ACTIVE : FLUSH_PASSIVE);
Write(NULL, 0, eom); Write(NULL, 0, eom);
SetFlushMode(oldFlushMode); SetFlushMode(oldFlushMode);
} }
@ -381,10 +396,15 @@ class NormStreamObject : public NormObject
virtual bool WriteSegment(NormBlockId blockId, virtual bool WriteSegment(NormBlockId blockId,
NormSegmentId segmentId, NormSegmentId segmentId,
const char* buffer); const char* buffer,
bool msgStart);
virtual UINT16 ReadSegment(NormBlockId blockId, virtual UINT16 ReadSegment(NormBlockId blockId,
NormSegmentId segmentId, NormSegmentId segmentId,
char* buffer); char* buffer,
bool* msgStart = NULL);
virtual char* RetrieveSegment(NormBlockId blockId,
NormSegmentId segmentId);
// For receive stream, we can rewind to earliest buffered offset // For receive stream, we can rewind to earliest buffered offset
@ -407,8 +427,15 @@ class NormStreamObject : public NormObject
NormBlockId GetNextBlockId() const NormBlockId GetNextBlockId() const
{return (server ? read_index.block : write_index.block);} {return (server ? read_index.block : write_index.block);}
NormSegmentId GetNextSegmentId() const NormSegmentId GetNextSegmentId() const
{return (server ? read_index.segment : write_index.segment);} {return (server ? read_index.segment : write_index.segment);}
UINT32 GetBlockPoolCount() {return block_pool.GetCount();}
void SetBlockPoolThreshold(UINT32 value)
{block_pool_threshold = value;}
private: private:
class Index class Index
{ {
public: public:
@ -427,12 +454,16 @@ class NormStreamObject : public NormObject
NormBlockBuffer stream_buffer; NormBlockBuffer stream_buffer;
Index write_index; Index write_index;
UINT32 write_offset; UINT32 write_offset;
bool read_init;
Index read_index; Index read_index;
UINT32 read_offset; UINT32 read_offset;
bool flush_pending; bool flush_pending;
bool msg_start; bool msg_start;
FlushMode flush_mode; FlushMode flush_mode;
bool push_mode; bool push_mode;
// For threaded API purposes
UINT32 block_pool_threshold;
}; // end class NormStreamObject }; // end class NormStreamObject
#ifdef SIMULATE #ifdef SIMULATE
@ -457,11 +488,19 @@ class NormSimObject : public NormObject
virtual bool WriteSegment(NormBlockId blockId, virtual bool WriteSegment(NormBlockId blockId,
NormSegmentId segmentId, NormSegmentId segmentId,
const char* buffer) {return true;} const char* buffer,
bool msgStart) {return true;}
virtual UINT16 ReadSegment(NormBlockId blockId, virtual UINT16 ReadSegment(NormBlockId blockId,
NormSegmentId segmentId, NormSegmentId segmentId,
char* buffer); char* buffer,
bool* msgStart = NULL);
virtual char* RetrieveSegment(NormBlockId blockId,
NormSegmentId segmentId)
{
return server ? server->GetRetrievalSegment() : NULL;
}
}; // end class NormSimObject }; // end class NormSimObject
#endif // SIMULATE #endif // SIMULATE

12
common/normSegment.cpp
View File

@ -541,7 +541,7 @@ bool NormBlock::AppendRepairRequest(NormNackMsg& nack,
{ {
if (0 == nack.PackRepairRequest(req)) if (0 == nack.PackRepairRequest(req))
{ {
DMSG(0, "NormBlock::AppendRepairRequest() warning: full NACK msg\n"); DMSG(3, "NormBlock::AppendRepairRequest() warning: full NACK msg\n");
break; break;
} }
} }
@ -573,13 +573,13 @@ bool NormBlock::AppendRepairRequest(NormNackMsg& nack,
if (NormRepairRequest::INVALID != prevForm) if (NormRepairRequest::INVALID != prevForm)
{ {
if (0 == nack.PackRepairRequest(req)) if (0 == nack.PackRepairRequest(req))
DMSG(0, "NormBlock::AppendRepairRequest() warning: full NACK msg\n"); DMSG(3, "NormBlock::AppendRepairRequest() warning: full NACK msg\n");
} }
return true; return true;
} // end NormBlock::AppendRepairRequest() } // end NormBlock::AppendRepairRequest()
NormBlockPool::NormBlockPool() NormBlockPool::NormBlockPool()
: head((NormBlock*)NULL), overruns(0), overrun_flag(false) : head((NormBlock*)NULL), count(0), overruns(0), overrun_flag(false)
{ {
} }
@ -588,7 +588,7 @@ NormBlockPool::~NormBlockPool()
Destroy(); Destroy();
} }
bool NormBlockPool::Init(UINT32 numBlocks, UINT16 totalSize) bool NormBlockPool::Init(UINT32 numBlocks, UINT16 segsPerBlock)
{ {
if (head) Destroy(); if (head) Destroy();
for (UINT32 i = 0; i < numBlocks; i++) for (UINT32 i = 0; i < numBlocks; i++)
@ -596,7 +596,7 @@ bool NormBlockPool::Init(UINT32 numBlocks, UINT16 totalSize)
NormBlock* b = new NormBlock(); NormBlock* b = new NormBlock();
if (b) if (b)
{ {
if (!b->Init(totalSize)) if (!b->Init(segsPerBlock))
{ {
DMSG(0, "NormBlockPool::Init() block init error\n"); DMSG(0, "NormBlockPool::Init() block init error\n");
delete b; delete b;
@ -613,6 +613,7 @@ bool NormBlockPool::Init(UINT32 numBlocks, UINT16 totalSize)
return false; return false;
} }
} }
count = numBlocks;
return true; return true;
} // end NormBlockPool::Init() } // end NormBlockPool::Init()
@ -624,6 +625,7 @@ void NormBlockPool::Destroy()
head = next->next; head = next->next;
delete next; delete next;
} }
count = 0;
} // end NormBlockPool::Destroy() } // end NormBlockPool::Destroy()
NormBlockBuffer::NormBlockBuffer() NormBlockBuffer::NormBlockBuffer()

7
common/normSegment.h
View File

@ -69,7 +69,7 @@ class NormBlock
void SetParityReadiness(UINT16 ndata) {erasure_count = ndata;} void SetParityReadiness(UINT16 ndata) {erasure_count = ndata;}
char** SegmentList(UINT16 index = 0) {return &segment_table[index];} char** SegmentList(UINT16 index = 0) {return &segment_table[index];}
char* Segment(NormSegmentId sid) char* GetSegment(NormSegmentId sid)
{ {
ASSERT(sid < size); ASSERT(sid < size);
return segment_table[sid]; return segment_table[sid];
@ -152,6 +152,7 @@ class NormBlock
// Note: This invalidates the repair_mask state. // Note: This invalidates the repair_mask state.
bool IsRepairPending(UINT16 ndata, UINT16 nparity); bool IsRepairPending(UINT16 ndata, UINT16 nparity);
void DecrementErasureCount() {erasure_count--;} void DecrementErasureCount() {erasure_count--;}
void IncrementErasureCount() {erasure_count++;}
UINT16 ErasureCount() const {return erasure_count;} UINT16 ErasureCount() const {return erasure_count;}
void IncrementParityCount() {parity_count++;} void IncrementParityCount() {parity_count++;}
UINT16 ParityCount() const {return parity_count;} UINT16 ParityCount() const {return parity_count;}
@ -257,6 +258,7 @@ class NormBlockPool
head = b ? b->next : NULL; head = b ? b->next : NULL;
if (b) if (b)
{ {
count--;
overrun_flag = false; overrun_flag = false;
} }
else if (!overrun_flag) else if (!overrun_flag)
@ -271,11 +273,14 @@ class NormBlockPool
{ {
b->next = head; b->next = head;
head = b; head = b;
count++;
} }
unsigned long OverrunCount() const {return overruns;} unsigned long OverrunCount() const {return overruns;}
UINT32 GetCount() {return count;}
private: private:
NormBlock* head; NormBlock* head;
UINT32 count;
unsigned long overruns; unsigned long overruns;
bool overrun_flag; bool overrun_flag;
}; // end class NormBlockPool }; // end class NormBlockPool

20
common/normSession.cpp
View File

@ -43,6 +43,7 @@ NormSession::NormSession(NormSessionMgr& sessionMgr, NormNodeId localNodeId)
trace(false), tx_loss_rate(0.0), rx_loss_rate(0.0), trace(false), tx_loss_rate(0.0), rx_loss_rate(0.0),
next(NULL) next(NULL)
{ {
interface_name[0] = '\0';
tx_socket.SetNotifier(&sessionMgr.GetSocketNotifier()); tx_socket.SetNotifier(&sessionMgr.GetSocketNotifier());
tx_socket.SetListener(this, &NormSession::TxSocketRecvHandler); tx_socket.SetListener(this, &NormSession::TxSocketRecvHandler);
@ -133,6 +134,8 @@ bool NormSession::Open(const char* interfaceName)
{ {
rx_socket.SetMulticastInterface(interfaceName); rx_socket.SetMulticastInterface(interfaceName);
tx_socket.SetMulticastInterface(interfaceName); tx_socket.SetMulticastInterface(interfaceName);
strncpy(interface_name, interfaceName, 31);
interface_name[31] = '\0';
} }
} }
for (unsigned int i = 0; i < DEFAULT_MESSAGE_POOL_DEPTH; i++) for (unsigned int i = 0; i < DEFAULT_MESSAGE_POOL_DEPTH; i++)
@ -165,7 +168,12 @@ void NormSession::Close()
if (tx_socket.IsOpen()) tx_socket.Close(); if (tx_socket.IsOpen()) tx_socket.Close();
if (rx_socket.IsOpen()) if (rx_socket.IsOpen())
{ {
if (address.IsMulticast()) rx_socket.LeaveGroup(address); if (address.IsMulticast())
{
const char* interfaceName = ('\0' != interface_name[0]) ?
interface_name : NULL;
rx_socket.LeaveGroup(address, interfaceName);
}
rx_socket.Close(); rx_socket.Close();
} }
} // end NormSession::Close() } // end NormSession::Close()
@ -654,7 +662,7 @@ void NormSession::ServerQueueFlush()
else else
{ {
// Why did I do this? - Brian // Why did I do this? - Brian
// (TBD) send NORM_CMD(EOT) instead? // (TBD) send NORM_CMD(EOT) instead? - no
if (ServerQueueSquelch(next_tx_object_id)) if (ServerQueueSquelch(next_tx_object_id))
{ {
flush_count++; flush_count++;
@ -1395,7 +1403,6 @@ void NormSession::ServerUpdateGrttEstimate(double clientRtt)
// Calculate grtt_advertised since quantization rounds upward // Calculate grtt_advertised since quantization rounds upward
grtt_advertised = NormUnquantizeRtt(grtt_quantized); grtt_advertised = NormUnquantizeRtt(grtt_quantized);
double clrRtt = cc_node_list.Head() ? ((NormCCNode*)cc_node_list.Head())->GetRtt() : -1;
if (grttQuantizedOld != grtt_quantized) if (grttQuantizedOld != grtt_quantized)
DMSG(4, "NormSession::ServerUpdateGrttEstimate() node>%lu new grtt>%lf sec\n", DMSG(4, "NormSession::ServerUpdateGrttEstimate() node>%lu new grtt>%lf sec\n",
LocalNodeId(), grtt_advertised); LocalNodeId(), grtt_advertised);
@ -2122,9 +2129,10 @@ void NormSession::ServerHandleNackMessage(const struct timeval& currentTime, Nor
if (startTimer && !repair_timer.IsActive()) if (startTimer && !repair_timer.IsActive())
{ {
// BACKOFF related code // BACKOFF related code
double aggregateInterval = address.IsMulticast() ? grtt_advertised * (backoff_factor + 1.0) : double aggregateInterval = address.IsMulticast() ?
0.0; grtt_advertised * (backoff_factor + 1.0) : 0.0;
aggregateInterval = (backoff_factor > 0.0) ? aggregateInterval : 0.0;//grtt_advertised / 100.0; // backoff == 0.0 is a special case
//aggregateInterval = (backoff_factor > 0.0) ? aggregateInterval : 0.0;
if (tx_timer.IsActive()) if (tx_timer.IsActive())
{ {

3
common/normSession.h
View File

@ -339,7 +339,8 @@ class NormSession
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
bool loopback; // to receive own traffic bool loopback; // to receive own traffic
char interface_name[32];
double tx_rate; // bytes per second double tx_rate; // bytes per second
double backoff_factor; double backoff_factor;

46
common/normTest.cpp
View File

@ -50,7 +50,7 @@ int main(int argc, char* argv[])
NormStartReceiver(session, 1024*1024); NormStartReceiver(session, 1024*1024);
// Uncomment the following line to start sender // Uncomment the following line to start sender
NormStartSender(session, 1024*1024, 1024, 64, 0); //NormStartSender(session, 1024*1024, 1024, 64, 0);
NormAddAckingNode(session, NormGetLocalNodeId(session)); NormAddAckingNode(session, NormGetLocalNodeId(session));
@ -59,8 +59,12 @@ int main(int argc, char* argv[])
const char* fileName = "ferrari.jpg"; const char* fileName = "ferrari.jpg";
// Uncomment this line to send a stream instead of the file // Uncomment this line to send a stream instead of the file
stream = NormOpenStream(session, 1024*1024); stream = NormStreamOpen(session, 1024*1024);
NormSetStreamFlushMode(stream, NORM_FLUSH_PASSIVE);
// NORM_FLUSH_PASSIVE automatically flushes full writes to
// the stream.
NormStreamSetFlushMode(stream, NORM_FLUSH_PASSIVE);
// Some variable for stream input/output // Some variable for stream input/output
@ -91,10 +95,16 @@ int main(int argc, char* argv[])
sprintf(txBuffer+1037, "normTest says hello %d ...\n", sendCount); sprintf(txBuffer+1037, "normTest says hello %d ...\n", sendCount);
txLen = strlen(txBuffer); txLen = strlen(txBuffer);
} }
txIndex += NormWriteStream(stream, txBuffer+txIndex, (txLen - txIndex)); txIndex += NormStreamWrite(stream, txBuffer+txIndex, (txLen - txIndex));
if (txIndex == txLen) if (txIndex == txLen)
{ {
NormMarkStreamEom(stream); // Instead of "NormStreamSetFlushMode(stream, NORM_FLUSH_PASSIVE)" above
// and "NormStreamMarkEom()" here, I could have used
// "NormStreamFlush(stream, true)" here to perform explicit flushing
// and EOM marking in one fell swoop. That would be a better approach
// for apps where big stream messages need to be written with
// multiple calls to "NormStreamWrite()"
NormStreamMarkEom(stream);
txLen = txIndex = 0; txLen = txIndex = 0;
sendCount++; sendCount++;
} }
@ -102,10 +112,10 @@ int main(int argc, char* argv[])
else else
{ {
NormObjectHandle txFile = NormObjectHandle txFile =
NormQueueFile(session, NormFileEnqueue(session,
filePath, filePath,
fileName, fileName,
strlen(fileName)); strlen(fileName));
// Repeatedly queue our file for sending // Repeatedly queue our file for sending
if (NORM_OBJECT_INVALID == txFile) if (NORM_OBJECT_INVALID == txFile)
{ {
@ -131,12 +141,12 @@ int main(int argc, char* argv[])
case NORM_RX_OBJECT_INFO: case NORM_RX_OBJECT_INFO:
// Assume info contains '/' delimited <path/fileName> string // Assume info contains '/' delimited <path/fileName> string
if (NORM_OBJECT_FILE == NormGetObjectType(theEvent.object)) if (NORM_OBJECT_FILE == NormObjectGetType(theEvent.object))
{ {
NormObjectTransportId id = NormGetObjectTransportId(theEvent.object); NormObjectTransportId id = NormObjectGetTransportId(theEvent.object);
if (0 != (id & 0x01)) if (0 != (id & 0x01))
{ {
//NormCancelObject(theEvent.object); //NormObjectCancel(theEvent.object);
//break; //break;
} }
@ -144,7 +154,7 @@ int main(int argc, char* argv[])
strcpy(fileName, cachePath); strcpy(fileName, cachePath);
int pathLen = strlen(fileName); int pathLen = strlen(fileName);
unsigned short nameLen = PATH_MAX - pathLen; unsigned short nameLen = PATH_MAX - pathLen;
NormGetObjectInfo(theEvent.object, fileName+pathLen, &nameLen); NormObjectGetInfo(theEvent.object, fileName+pathLen, &nameLen);
fileName[nameLen + pathLen] = '\0'; fileName[nameLen + pathLen] = '\0';
char* ptr = fileName + 5; char* ptr = fileName + 5;
while ('\0' != *ptr) while ('\0' != *ptr)
@ -152,7 +162,7 @@ int main(int argc, char* argv[])
if ('/' == *ptr) *ptr = PROTO_PATH_DELIMITER; if ('/' == *ptr) *ptr = PROTO_PATH_DELIMITER;
ptr++; ptr++;
} }
if (!NormSetFileName(theEvent.object, fileName)) if (!NormFileRename(theEvent.object, fileName))
TRACE("normTest: NormSetFileName(%s) error\n", fileName); TRACE("normTest: NormSetFileName(%s) error\n", fileName);
DMSG(3, "normTest: recv'd info for file: %s\n", fileName); DMSG(3, "normTest: recv'd info for file: %s\n", fileName);
@ -163,16 +173,16 @@ int main(int argc, char* argv[])
case NORM_RX_OBJECT_UPDATE: case NORM_RX_OBJECT_UPDATE:
{ {
//TRACE("normTest: NORM_RX_OBJECT_UPDATE event ...\n"); //TRACE("normTest: NORM_RX_OBJECT_UPDATE event ...\n");
if (NORM_OBJECT_STREAM != NormGetObjectType(theEvent.object)) if (NORM_OBJECT_STREAM != NormObjectGetType(theEvent.object))
break; break;
unsigned int len; unsigned int len;
do do
{ {
if (!msgSync) if (!msgSync)
msgSync = NormFindStreamMsgStart(theEvent.object); msgSync = NormStreamSeekMsgStart(theEvent.object);
if (!msgSync) break; if (!msgSync) break;
len = 8191 - rxIndex; len = 8191 - rxIndex;
if (NormReadStream(theEvent.object, rxBuffer+rxIndex, &len)) if (NormStreamRead(theEvent.object, rxBuffer+rxIndex, &len))
{ {
rxIndex += len; rxIndex += len;
rxBuffer[rxIndex] = '\0'; rxBuffer[rxIndex] = '\0';
@ -266,7 +276,7 @@ int main(int argc, char* argv[])
sleep(30); // allows time for cleanup if we're sending to someone else sleep(30); // allows time for cleanup if we're sending to someone else
#endif // if/else WIN32/UNIX #endif // if/else WIN32/UNIX
NormCloseStream(stream); NormStreamClose(stream);
NormStopReceiver(session); NormStopReceiver(session);
NormStopSender(session); NormStopSender(session);
NormDestroySession(session); NormDestroySession(session);

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