From f451a006f648e13dd937f3ec663633491965dfcc Mon Sep 17 00:00:00 2001 From: Jeff Weston Date: Wed, 11 Sep 2019 11:42:03 -0400 Subject: [PATCH] v1.2b4 --- README.md | 21 +- VERSION.TXT | 13 ++ common/normApi.cpp | 169 +++++++------- common/normApi.h | 88 +++---- common/normApp.cpp | 65 +++++- common/normBitmask.cpp | 2 - common/normEncoder.cpp | 2 +- common/normNode.cpp | 171 ++++++++++---- common/normNode.h | 24 +- common/normObject.cpp | 504 ++++++++++++++++++++++++++++++----------- common/normObject.h | 63 +++++- common/normSegment.cpp | 12 +- common/normSegment.h | 7 +- common/normSession.cpp | 20 +- common/normSession.h | 3 +- common/normTest.cpp | 46 ++-- common/normVersion.h | 2 +- 17 files changed, 853 insertions(+), 359 deletions(-) diff --git a/README.md b/README.md index f6a2ff6..3697528 100644 --- a/README.md +++ b/README.md @@ -4,10 +4,21 @@ The NORM code here is still preliminary but a functional "norm" 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- -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 -code could be assembled as Win32 project under VC++. +A "normTest.cpp" program is now included which is a simple demonstration +(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. -See for that code. +The Win32 code is not yet as well-tested as the Unix version, but +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. diff --git a/VERSION.TXT b/VERSION.TXT index e4a20c8..6e62914 100644 --- a/VERSION.TXT +++ b/VERSION.TXT @@ -1,5 +1,18 @@ 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 ============= - There have been a number of bug fixes since 1.1b3. This release diff --git a/common/normApi.cpp b/common/normApi.cpp index 6591d8d..d1311ec 100644 --- a/common/normApi.cpp +++ b/common/normApi.cpp @@ -217,12 +217,19 @@ void NormInstance::Notify(NormController::Event event, NormStreamObject* stream = static_cast(object); // (TBD) implement silent_client accept differently 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"); notify_pool.Append(n); 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; } case NormObject::FILE: @@ -688,33 +695,6 @@ void NormSetRxLoss(NormSessionHandle sessionHandle, double percent) } // 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((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((NormNode*)nodeHandle); - if (node) - node->SetRepairBoundary((NormServerNode::RepairBoundary)repairBoundary); -} // end NormSetNodeRepairBoundary() - void NormSetDefaultRepairBoundary(NormSessionHandle sessionHandle, NormRepairBoundary repairBoundary) { @@ -724,15 +704,44 @@ void NormSetDefaultRepairBoundary(NormSessionHandle sessionHandle, } // end NormSetDefaultRepairBoundary() -NormObjectType NormGetObjectType(NormObjectHandle objectHandle) +NormRepairBoundary NormNodeGetRepairBoundary(NormNodeHandle nodeHandle) +{ + NormServerNode* node = static_cast((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((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) return ((NormObjectType)(((NormObject*)objectHandle)->GetType())); else 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 // to provide an error code result @@ -740,9 +749,9 @@ NormObjectTransportId NormGetObjectTransportId(NormObjectHandle objectHandle) return ((NormObjectTransportId)(((NormObject*)objectHandle)->GetId())); else return 0; -} // end NormGetFileName() +} // end NormObjectGetTransportId() -bool NormGetObjectInfo(NormObjectHandle objectHandle, +bool NormObjectGetInfo(NormObjectHandle objectHandle, char* infoBuffer, unsigned short* infoLen) { @@ -768,9 +777,9 @@ bool NormGetObjectInfo(NormObjectHandle objectHandle, return result; } return false; -} // end NormGetObjectInfo() +} // end NormObjectGetInfo() -void NormCancelObject(NormObjectHandle objectHandle) +void NormObjectCancel(NormObjectHandle objectHandle) { if (NORM_OBJECT_INVALID != objectHandle) { @@ -780,16 +789,16 @@ void NormCancelObject(NormObjectHandle objectHandle) NormObject* obj = (NormObject*)objectHandle; NormServerNode* server = obj->GetServer(); if (server) - server->DeleteObject(obj, 8); + server->DeleteObject(obj); else obj->GetSession().DeleteTxObject(obj); instance->PurgeObjectNotifications(objectHandle); instance->dispatcher.ResumeThread(); } } -} // end NormCancelObject() +} // end NormObjectCancel() -void NormRetainObject(NormObjectHandle objectHandle) +void NormObjectRetain(NormObjectHandle objectHandle) { if (NORM_OBJECT_INVALID != objectHandle) { @@ -803,7 +812,7 @@ void NormRetainObject(NormObjectHandle objectHandle) } } // end NormRetainObject() -void NormReleaseObject(NormObjectHandle objectHandle) +void NormObjectRelease(NormObjectHandle objectHandle) { if (NORM_OBJECT_INVALID != objectHandle) { @@ -812,20 +821,20 @@ void NormReleaseObject(NormObjectHandle objectHandle) { NormObject* obj = (NormObject*)objectHandle; obj->Release(); - //instance->PurgeObjectNotifications(objectHandle); + instance->PurgeObjectNotifications(objectHandle); instance->dispatcher.ResumeThread(); } } -} // end NormReleaseObject() +} // end NormObjectRelease() -NormNackingMode NormGetObjectNackingMode(NormObjectHandle objectHandle) +NormNackingMode NormObjectGetNackingMode(NormObjectHandle objectHandle) { NormObject* object = (NormObject*)objectHandle; if (object) return ((NormNackingMode)object->GetNackingMode()); else return NORM_NACK_NONE; -} // end NormGetObjectNackingMode() +} // end NormObjectGetNackingMode() void NormSetObjectNackingMode(NormObjectHandle objectHandle, NormNackingMode nackingMode) @@ -834,12 +843,12 @@ void NormSetObjectNackingMode(NormObjectHandle objectHandle, if (object) object->SetNackingMode((NormObject::NackingMode)nackingMode); } // end NormSetObjectNackingMode() -void NormSetNodeNackingMode(NormNodeHandle nodeHandle, +void NormObjectSetNackingMode(NormNodeHandle nodeHandle, NormNackingMode nackingMode) { NormServerNode* node = (NormServerNode*)nodeHandle; if (node) node->SetDefaultNackingMode((NormObject::NackingMode)nackingMode); -} // end NormSetNodeNackingMode() +} // end NormObjectSetNackingMode() void NormSetDefaultNackingMode(NormSessionHandle sessionHandle, NormNackingMode nackingMode) @@ -910,7 +919,7 @@ void NormRemoveAckingNode(NormSessionHandle sessionHandle, } } // end NormAddAckingNode() -NormObjectHandle NormOpenStream(NormSessionHandle sessionHandle, +NormObjectHandle NormStreamOpen(NormSessionHandle sessionHandle, unsigned long bufferSize) { NormObjectHandle objectHandle = NORM_OBJECT_INVALID; @@ -927,12 +936,12 @@ NormObjectHandle NormOpenStream(NormSessionHandle sessionHandle, instance->dispatcher.ResumeThread(); } return objectHandle; -} // end NormOpenStream() +} // end NormStreamOpen() -void NormCloseStream(NormObjectHandle streamHandle) +void NormStreamClose(NormObjectHandle streamHandle) { - NormCancelObject(streamHandle); -} // end NormCloseStream() + NormObjectCancel(streamHandle); +} // end NormStreamClose() // (TBD) Some stream i/o performance improvement can be realized @@ -950,35 +959,35 @@ void NormCloseStream(NormObjectHandle streamHandle) // events with a "NormDispatchEvents()" call ... // -void NormSetStreamFlushMode(NormObjectHandle streamHandle, +void NormStreamSetFlushMode(NormObjectHandle streamHandle, NormFlushMode flushMode) { NormStreamObject* stream = static_cast((NormObject*)streamHandle); if (stream) stream->SetFlushMode((NormStreamObject::FlushMode)flushMode); -} // end NormSetStreamFlushMode() +} // end NormStreamSetFlushMode() -void NormSetStreamPushMode(NormObjectHandle streamHandle, bool state) +void NormStreamSetPushMode(NormObjectHandle streamHandle, bool state) { NormStreamObject* stream = static_cast((NormObject*)streamHandle); if (stream) stream->SetPushMode(state); -} // end NormSetStreamPushMode() +} // end NormStreamSetPushMode() -void NormFlushStream(NormObjectHandle streamHandle) +void NormStreamFlush(NormObjectHandle streamHandle, bool eom) { NormInstance* instance = NormInstance::GetInstanceFromObject(streamHandle); if (instance && instance->dispatcher.SuspendThread()) { NormStreamObject* stream = static_cast((NormObject*)streamHandle); - if (stream) stream->Flush(); + if (stream) stream->Flush(eom); instance->dispatcher.ResumeThread(); } -} // end NormFlushStream() +} // end NormStreamFlush() -void NormMarkStreamEom(NormObjectHandle streamHandle) +void NormStreamMarkEom(NormObjectHandle streamHandle) { NormInstance* instance = NormInstance::GetInstanceFromObject(streamHandle); if (instance && instance->dispatcher.SuspendThread()) @@ -990,7 +999,7 @@ void NormMarkStreamEom(NormObjectHandle streamHandle) } } // end NormMarkStreamEom() -unsigned int NormWriteStream(NormObjectHandle streamHandle, +unsigned int NormStreamWrite(NormObjectHandle streamHandle, const char* buffer, unsigned int numBytes) { @@ -1005,9 +1014,9 @@ unsigned int NormWriteStream(NormObjectHandle streamHandle, instance->dispatcher.ResumeThread(); } return result; -} // end NormWriteStream() +} // end NormStreamWrite() -bool NormReadStream(NormObjectHandle streamHandle, +bool NormStreamRead(NormObjectHandle streamHandle, char* buffer, unsigned int* numBytes) { @@ -1022,9 +1031,9 @@ bool NormReadStream(NormObjectHandle streamHandle, instance->dispatcher.ResumeThread(); } return result; -} // end NormReadStream() +} // end NormStreamRead() -bool NormFindStreamMsgStart(NormObjectHandle streamHandle) +bool NormStreamSeekMsgStart(NormObjectHandle streamHandle) { bool result = false; NormInstance* instance = NormInstance::GetInstanceFromObject(streamHandle); @@ -1038,12 +1047,12 @@ bool NormFindStreamMsgStart(NormObjectHandle streamHandle) instance->dispatcher.ResumeThread(); } return result; -} // end NormFindStreamMsgStart() +} // end NormStreamSeekMsgStart() -NormObjectHandle NormQueueFile(NormSessionHandle sessionHandle, - const char* fileName, - const char* infoPtr, - unsigned int infoLen) +NormObjectHandle NormFileEnqueue(NormSessionHandle sessionHandle, + const char* fileName, + const char* infoPtr, + unsigned int infoLen) { NormObjectHandle objectHandle = NORM_OBJECT_INVALID; NormInstance* instance = NormInstance::GetInstanceFromSession(sessionHandle); @@ -1058,7 +1067,7 @@ NormObjectHandle NormQueueFile(NormSessionHandle sessionHandle, instance->dispatcher.ResumeThread(); } return objectHandle; -} // end NormQueueFile() +} // end NormFileEnqueue() bool NormSetCacheDirectory(NormInstanceHandle instanceHandle, const char* cachePath) @@ -1070,8 +1079,8 @@ bool NormSetCacheDirectory(NormInstanceHandle instanceHandle, return false; } // end NormSetCacheDirectory() -bool NormSetFileName(NormObjectHandle fileHandle, - const char* fileName) +bool NormFileRename(NormObjectHandle fileHandle, + const char* fileName) { bool result = false; NormInstance* instance = NormInstance::GetInstanceFromObject(fileHandle); @@ -1084,9 +1093,9 @@ bool NormSetFileName(NormObjectHandle fileHandle, instance->dispatcher.ResumeThread(); } return result; -} // end NormGetFileName() +} // end NormFileRename() -bool NormGetFileName(NormObjectHandle fileHandle, +bool NormFileGetName(NormObjectHandle fileHandle, char* nameBuffer, unsigned int bufferLen) { @@ -1102,13 +1111,13 @@ bool NormGetFileName(NormObjectHandle fileHandle, result = true; } return result; -} // end NormGetFileName() +} // end NormFileGetName() -NormObjectHandle NormQueueData(NormSessionHandle sessionHandle, - const char* dataPtr, - unsigned long dataLen, - const char* infoPtr = NULL, - unsigned int infoLen = 0) +NormObjectHandle NormDataEnqueue(NormSessionHandle sessionHandle, + const char* dataPtr, + unsigned long dataLen, + const char* infoPtr = NULL, + unsigned int infoLen = 0) { NormObjectHandle objectHandle = NORM_OBJECT_INVALID; NormInstance* instance = NormInstance::GetInstanceFromSession(sessionHandle); @@ -1123,7 +1132,7 @@ NormObjectHandle NormQueueData(NormSessionHandle sessionHandle, instance->dispatcher.ResumeThread(); } return objectHandle; -} // end NormQueueData() +} // end NormDataEnqueue() bool NormGetNextEvent(NormInstanceHandle instanceHandle, NormEvent* theEvent) diff --git a/common/normApi.h b/common/normApi.h index 028039e..a8eff0c 100644 --- a/common/normApi.h +++ b/common/normApi.h @@ -93,22 +93,28 @@ enum NormNackingMode NORM_NACK_NORMAL }; +void NormSetDefaultNackingMode(NormSessionHandle sessionHandle, + NormNackingMode nackingMode); + enum NormRepairBoundary { NORM_BOUNDARY_BLOCK, NORM_BOUNDARY_OBJECT }; - -NormNodeId NormGetNodeId(NormNodeHandle nodeHandle); - -NormRepairBoundary NormGetNodeRepairBoundary(NormNodeHandle nodeHandle); - -void NormSetNodeRepairBoundary(NormNodeHandle nodeHandle, - NormRepairBoundary repairBoundary); - void NormSetDefaultRepairBoundary(NormSessionHandle sessionHandle, NormRepairBoundary repairBoundary); + +NormRepairBoundary NormNodeGetRepairBoundary(NormNodeHandle nodeHandle); + +void NormNodeSetRepairBoundary(NormNodeHandle nodeHandle, + NormRepairBoundary repairBoundary); + +NormNodeId NormNodeGetId(NormNodeHandle nodeHandle); + +void NormNodeSetNackingMode(NormNodeHandle nodeHandle, + NormNackingMode nackingMode); + // 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, unsigned short* infoLen); -NormObjectTransportId NormGetObjectTransportId(NormObjectHandle objectHandle); +NormObjectTransportId NormObjectGetTransportId(NormObjectHandle objectHandle); -void NormCancelObject(NormObjectHandle objectHandle); +void NormObjectCancel(NormObjectHandle objectHandle); -void NormRetainObject(NormObjectHandle objectHandle); -void NormReleaseObject(NormObjectHandle objectHandle); +void NormObjectReNormObjectRetaintain(NormObjectHandle objectHandle); +void NormObjectRelease(NormObjectHandle objectHandle); // Receiver-only NormObject functions -NormNackingMode NormGetObjectNackingMode(NormObjectHandle objectHandle); +NormNackingMode NormObjectGetNackingMode(NormObjectHandle objectHandle); -void NormSetObjectNackingMode(NormObjectHandle objectHandle, +void NormObjectSetNackingMode(NormObjectHandle objectHandle, NormNackingMode nackingMode); -void NormSetNodeNackingMode(NormNodeHandle nodeHandle, - NormNackingMode nackingMode); -void NormSetDefaultNackingMode(NormSessionHandle sessionHandle, - NormNackingMode nackingMode); + // Sender-only NormObject functions bool NormSetWatermark(NormSessionHandle sessionHandle, @@ -156,10 +159,10 @@ bool NormSetWatermark(NormSessionHandle sessionHandle, // NormStreamObject functions -NormObjectHandle NormOpenStream(NormSessionHandle sessionHandle, +NormObjectHandle NormStreamOpen(NormSessionHandle sessionHandle, unsigned long bufferSize); -void NormCloseStream(NormObjectHandle streamHandle); +void NormStreamClose(NormObjectHandle streamHandle); enum NormFlushMode { @@ -168,47 +171,47 @@ enum NormFlushMode NORM_FLUSH_ACTIVE }; -void NormSetStreamFlushMode(NormObjectHandle streamHandle, +void NormStreamSetFlushMode(NormObjectHandle streamHandle, NormFlushMode flushMode); -void NormSetStreamPushMode(NormObjectHandle streamHandle, +void NormStreamSetPushMode(NormObjectHandle streamHandle, bool state); -unsigned int NormWriteStream(NormObjectHandle streamHandle, +unsigned int NormStreamWrite(NormObjectHandle streamHandle, const char* buffer, 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, unsigned int* numBytes); -bool NormFindStreamMsgStart(NormObjectHandle streamHandle); +bool NormStreamSeekMsgStart(NormObjectHandle streamHandle); // NormFileObject Functions -NormObjectHandle NormQueueFile(NormSessionHandle sessionHandle, - const char* fileName, - const char* infoPtr = (const char*)0, - unsigned int infoLen = 0); +NormObjectHandle NormFileEnqueue(NormSessionHandle sessionHandle, + const char* fileName, + const char* infoPtr = (const char*)0, + unsigned int infoLen = 0); bool NormSetCacheDirectory(NormInstanceHandle instanceHandle, const char* cachePath); -bool NormGetFileName(NormObjectHandle fileHandle, +bool NormFileGetName(NormObjectHandle fileHandle, char* nameBuffer, unsigned int bufferLen); -bool NormSetFileName(NormObjectHandle fileHandle, - const char* fileName); +bool NormFileRename(NormObjectHandle fileHandle, + const char* fileName); // NormDataObject Functions -bool NormQueueData(const char* dataPtr, - unsigned long dataLen, - const char* infoPtr = (const char*)0, - unsigned int infoLen = 0); +bool NormDataEnqueue(const char* dataPtr, + unsigned long dataLen, + const char* infoPtr = (const char*)0, + unsigned int infoLen = 0); // NORM Event Notification Routines @@ -252,9 +255,10 @@ bool NormGetNextEvent(NormInstanceHandle instanceHandle, NormEvent* theEvent); // not block. #ifdef WIN32 -HANDLE NormGetDescriptor(NormInstanceHandle instanceHandle); +typedef HANDLE NormDescriptor; #else -int NormGetDescriptor(NormInstanceHandle instanceHandle); +typedef int NormDescriptor; #endif // if/else WIN32/UNIX +NormDescriptor NormGetDescriptor(NormInstanceHandle instanceHandle); #endif // _NORM_API diff --git a/common/normApp.cpp b/common/normApp.cpp index c3ecf37..9807d5c 100644 --- a/common/normApp.cpp +++ b/common/normApp.cpp @@ -35,6 +35,7 @@ class NormApp : public NormController, public ProtoApp const void* userData); private: + void ShowHelp(); void OnInputReady(); enum CmdType {CMD_INVALID, CMD_NOARG, CMD_ARG}; @@ -184,6 +185,7 @@ NormApp::~NormApp() const char* const NormApp::cmd_list[] = { + "-help", // show this help" "+debug", // debug level "+log", // log file name "+trace", // message tracing on @@ -225,6 +227,56 @@ const char* const NormApp::cmd_list[] = 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) { switch (theEvent) @@ -275,6 +327,12 @@ bool NormApp::OnCommand(const char* cmd, const char* val) return false; } + if (!strncmp("help", cmd, len)) + { + ShowHelp(); + return false; + } + if (control_remote) { // 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) { - + if (argc < 3) + { + ShowHelp(); + return false; + } + if (!(post_processor = NormPostProcessor::Create())) { DMSG(0, "NormApp::OnStartup() error creating post processor\n"); diff --git a/common/normBitmask.cpp b/common/normBitmask.cpp index a143ad3..74a33dd 100644 --- a/common/normBitmask.cpp +++ b/common/normBitmask.cpp @@ -201,8 +201,6 @@ void NormBitmask::Destroy() 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 < first_set) return GetFirstSet(index); UINT32 maskIndex = index >> 3; diff --git a/common/normEncoder.cpp b/common/normEncoder.cpp index 5103e5f..757aa88 100644 --- a/common/normEncoder.cpp +++ b/common/normEncoder.cpp @@ -409,7 +409,7 @@ int NormDecoder::Decode(char** dVec, int ndata, UINT16 erasureCount, UINT16* era int j; for (j = 1; j < degree; j += 2) 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); // Now evaluate Omega at alpha^(-i) (numerator) diff --git a/common/normNode.cpp b/common/normNode.cpp index 99c0063..6f3d3b0 100644 --- a/common/normNode.cpp +++ b/common/normNode.cpp @@ -48,6 +48,7 @@ NormServerNode::NormServerNode(class NormSession& theSession, NormNodeId nodeId) : 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), repair_boundary(BLOCK_BOUNDARY), erasure_loc(NULL), + retrieval_loc(NULL), retrieval_pool(NULL), cc_sequence(0), cc_enable(false), cc_rate(0.0), rtt_confirmed(false), is_clr(false), is_plr(false), slow_start(true), send_rate(0.0), recv_rate(0.0), recv_accumulator(0), @@ -140,22 +141,42 @@ void NormServerNode::Close() is_open = false; } // end NormServerNode::Close() -bool NormServerNode::AllocateBuffers(UINT16 segmentSize, UINT16 numData, UINT16 numParity) +bool NormServerNode::AllocateBuffers(UINT16 segmentSize, + UINT16 numData, + UINT16 numParity) { ASSERT(IsOpen()); // Calculate how much memory each buffered block will require UINT16 blockSize = numData + numParity; unsigned long maskSize = blockSize >> 3; if (0 != (blockSize & 0x07)) maskSize++; - unsigned long blockSpace = sizeof(NormBlock) + - blockSize * sizeof(char*) + - 2*maskSize + - numData * (segmentSize + NormDataMsg::GetStreamPayloadHeaderLength()); + unsigned long blockStateSpace = sizeof(NormBlock) + blockSize * sizeof(char*) + 2*maskSize; 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; if (bufferSpace > (numBlocks*blockSpace)) numBlocks++; if (numBlocks < 2) numBlocks = 2; - unsigned long numSegments = numBlocks * numData; + unsigned long numSegments = numBlocks * segPerBlock; + if (!block_pool.Init(numBlocks, blockSize)) { @@ -164,7 +185,7 @@ bool NormServerNode::AllocateBuffers(UINT16 segmentSize, UINT16 numData, UINT16 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 // a NormStreamObject if (!segment_pool.Init(numSegments, segmentSize+NormDataMsg::GetStreamPayloadHeaderLength()+1)) @@ -174,17 +195,44 @@ bool NormServerNode::AllocateBuffers(UINT16 segmentSize, UINT16 numData, UINT16 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())) { - DMSG(0, "NormServerNode::Open() decoder init error: %s\n", - strerror(errno)); + DMSG(0, "NormServerNode::Open() decoder init error\n"); Close(); return false; } if (!(erasure_loc = new UINT16[numParity])) { - DMSG(0, "NormServerNode::Open() erasure_loc allocation error: %s\n", - strerror(errno)); + DMSG(0, "NormServerNode::Open() erasure_loc allocation error: %s\n", GetErrorString()); Close(); return false; } @@ -197,12 +245,31 @@ bool NormServerNode::AllocateBuffers(UINT16 segmentSize, UINT16 numData, UINT16 void NormServerNode::FreeBuffers() { - decoder.Destroy(); if (erasure_loc) { - delete []erasure_loc; + delete[] erasure_loc; 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; while ((obj = rx_table.Find(rx_table.RangeLo()))) { @@ -701,7 +768,6 @@ NormBlock* NormServerNode::GetFreeBlock(NormObjectId objectId, NormBlockId block if (!b) { if (session.ClientIsSilent()) - //if (1) { // forward iteration to find oldest older object with resources NormObjectTable::Iterator iterator(rx_table); @@ -757,13 +823,29 @@ NormBlock* NormServerNode::GetFreeBlock(NormObjectId objectId, NormBlockId block char* NormServerNode::GetFreeSegment(NormObjectId objectId, NormBlockId blockId) { - while (segment_pool.IsEmpty()) + if (segment_pool.IsEmpty()) { - NormBlock* b = GetFreeBlock(objectId, blockId); - if (b) - block_pool.Put(b); - else - break; + // First, try to steal (retrievable) buffered source symbol segments + NormObjectTable::Iterator iterator(rx_table); + NormObject* obj; + while ((obj = iterator.GetNextObject())) + { + // 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(); } // end NormServerNode::GetFreeSegment() @@ -869,7 +951,7 @@ void NormServerNode::HandleObjectMessage(const NormObjectMsg& msg) else { // 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) { 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) : 0.0; 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); session.ActivateTimer(repair_timer); } @@ -1305,7 +1387,7 @@ void NormServerNode::RepairCheck(NormObject::CheckLevel checkLevel, if (rewindDetected) { 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()); RepairCheck(checkLevel, objectId, blockId, segmentId); @@ -1356,7 +1438,7 @@ bool NormServerNode::OnRepairTimeout(ProtoTimer& /*theTimer*/) NormNackMsg* nack = (NormNackMsg*)session.GetMessageFromPool(); if (!nack) { - DMSG(0, "NormServerNode::OnRepairTimeout() node>%lu Warning! " + DMSG(3, "NormServerNode::OnRepairTimeout() node>%lu Warning! " "message pool empty ...\n", LocalNodeId()); repair_timer.Deactivate(); return false; @@ -1446,7 +1528,7 @@ bool NormServerNode::OnRepairTimeout(ProtoTimer& /*theTimer*/) { if (0 == nack->PackRepairRequest(req)) { - DMSG(0, "NormServerNode::OnRepairTimeout() warning: full NACK msg\n"); + DMSG(3, "NormServerNode::OnRepairTimeout() warning: full NACK msg\n"); break; } nackAppended = true; @@ -1489,7 +1571,7 @@ bool NormServerNode::OnRepairTimeout(ProtoTimer& /*theTimer*/) { if (0 == nack->PackRepairRequest(req)) { - DMSG(0, "NormServerNode::OnRepairTimeout() warning: full NACK msg\n"); + DMSG(3, "NormServerNode::OnRepairTimeout() warning: full NACK msg\n"); break; } nackAppended = true; @@ -1520,7 +1602,7 @@ bool NormServerNode::OnRepairTimeout(ProtoTimer& /*theTimer*/) if (0 != nack->PackRepairRequest(req)) nackAppended = true; else - DMSG(0, "NormServerNode::OnRepairTimeout() warning: full NACK msg\n"); + DMSG(3, "NormServerNode::OnRepairTimeout() warning: full NACK msg\n"); } // Queue NACK for transmission nack->SetServerId(GetId()); @@ -1535,7 +1617,9 @@ bool NormServerNode::OnRepairTimeout(ProtoTimer& /*theTimer*/) if (nackAppended) { ASSERT(nack->GetRepairContentLength() > 0); - session.QueueMessage(nack); + //session.QueueMessage(nack); + session.SendMessage(*nack); + session.ReturnMessageToPool(nack); nack_count++; } else @@ -1558,7 +1642,8 @@ bool NormServerNode::OnRepairTimeout(ProtoTimer& /*theTimer*/) double holdoffInterval = session.Address().IsMulticast() ? grtt_estimate*(backoff_factor + 2.0) : 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); 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(); if (!ack) { - DMSG(0, "NormServerNode::OnCCTimeout() node>%lu Warning! " + DMSG(3, "NormServerNode::OnCCTimeout() node>%lu warning: " "message pool empty ...\n", LocalNodeId()); if (cc_timer.IsActive()) cc_timer.Deactivate(); return false; @@ -1759,16 +1844,16 @@ bool NormServerNode::OnCCTimeout(ProtoTimer& /*theTimer*/) ack->SetDestination(GetAddress()); else 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.ReturnMessageToPool(ack); } - else - { - session.QueueMessage(ack); - } + //else + //{ + // session.QueueMessage(ack); + //} // Begin cc_timer "holdoff" phase cc_timer.SetInterval(grtt_estimate*backoff_factor); @@ -1802,15 +1887,11 @@ bool NormServerNode::OnAckTimeout(ProtoTimer& /*theTimer*/) else 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 if (cc_timer.IsActive()) cc_timer.Deactivate(); cc_timer.SetInterval(grtt_estimate*backoff_factor); @@ -1820,7 +1901,7 @@ bool NormServerNode::OnAckTimeout(ProtoTimer& /*theTimer*/) } else { - DMSG(0, "NormServerNode::OnAckTimeout() warning: message pool exhausted!\n"); + DMSG(3, "NormServerNode::OnAckTimeout() warning: message pool exhausted!\n"); } return true; } // end NormServerNode::OnAckTimeout() diff --git a/common/normNode.h b/common/normNode.h index cdd8678..6c85051 100644 --- a/common/normNode.h +++ b/common/normNode.h @@ -293,6 +293,7 @@ class NormServerNode : public NormNode block->EmptyToPool(segment_pool); block_pool.Put(block); } + bool SegmentPoolIsEmpty() {return segment_pool.IsEmpty();} char* GetFreeSegment(NormObjectId objectId, NormBlockId blockId); void PutFreeSegment(char* segment) {segment_pool.Put(segment);} @@ -303,7 +304,25 @@ class NormServerNode : public NormNode erasure_loc[index] = value; } 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) { return decoder.Decode(segmentList, numData, erasureCount, erasure_loc); @@ -373,6 +392,9 @@ class NormServerNode : public NormNode NormSegmentPool segment_pool; NormDecoder decoder; UINT16* erasure_loc; + UINT16* retrieval_loc; + char** retrieval_pool; + UINT16 retrieval_index; bool server_active; ProtoTimer activity_timer; diff --git a/common/normObject.cpp b/common/normObject.cpp index 1857e18..10520a2 100644 --- a/common/normObject.cpp +++ b/common/normObject.cpp @@ -899,7 +899,7 @@ bool NormObject::AppendRepairRequest(NormNackMsg& nack, { 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; } requestAppended = true; @@ -958,7 +958,7 @@ bool NormObject::AppendRepairRequest(NormNackMsg& nack, { 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; } } @@ -1004,7 +1004,7 @@ bool NormObject::AppendRepairRequest(NormNackMsg& nack, { 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; } requestAppended = true; @@ -1020,7 +1020,7 @@ bool NormObject::AppendRepairRequest(NormNackMsg& nack, req.AppendRepairItem(transport_id, 0, 0, 0); // (TBD) error check 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; } requestAppended = true; @@ -1115,22 +1115,12 @@ void NormObject::HandleObjectMessage(const NormObjectMsg& msg, } 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(); if (segmentLength > segment_size) { DMSG(0, "NormObject::HandleObjectMessage() node>%lu server>%lu obj>%hu " "Error! segment too large ...\n", LocalNodeId(), server->GetId(), (UINT16)transport_id); - server->PutFreeSegment(segment); return; } UINT16 payloadLength = data.GetPayloadLength(); @@ -1140,35 +1130,54 @@ void NormObject::HandleObjectMessage(const NormObjectMsg& msg, payloadMax = MIN(payloadMax, SIM_PAYLOAD_MAX); payloadLength = MIN(payloadLength, SIM_PAYLOAD_MAX); #endif // SIMULATE - memcpy(segment, data.GetPayload(), payloadLength); - if (payloadLength < payloadMax) - memset(segment+payloadLength, 0, payloadMax-payloadLength); - if (data.FlagIsSet(NormObjectMsg::FLAG_MSG_START)) + + // Is this a source symbol or a parity symbol? + bool isSourceSymbol = (segmentId < numData); + + // 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; - //DMSG(0, "starting segment payload len>%hu msg_len>%hu\n", - // payloadLength, ntohs(*((UINT16*)(segment+8)))) ; + memcpy(segment, data.GetPayload(), payloadLength); + if (payloadLength < payloadMax) + memset(segment+payloadLength, 0, payloadMax-payloadLength); + block->AttachSegment(segmentId, segment); } 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); + bool objectUpdated = false; - // 2) Write segment to object (if it's data) - if (segmentId < numData) + // 2) Write segment to object (if it's source symbol (data)) + if (isSourceSymbol) { block->DecrementErasureCount(); - if (WriteSegment(blockId, segmentId, segment)) + if (WriteSegment(blockId, segmentId, data.GetPayload(), + data.FlagIsSet(NormObjectMsg::FLAG_MSG_START))) { objectUpdated = true; // For statistics only (TBD) #ifdef NORM_DEBUG server->IncrementRecvGoodput(segmentLength); } + else + { + DMSG(4, "NormObject::HandleObjectMessage() WriteSegment() error\n"); + } } else { block->IncrementParityCount(); } + + + // 3) Decode block if ready and return to pool 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 " "completed block ...\n", LocalNodeId(), server->GetId(), (UINT16)transport_id, (UINT32)block->GetId()); - UINT16 erasureCount = 0; + UINT16 erasureCount = 0; UINT16 nextErasure = 0; + UINT16 retrievalCount = 0; if (block->GetFirstPending(nextErasure)) { + // Is the block missing _any_ source symbols? 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 (nextErasure < numData) + if (block->IsPending(nextSegment)) { - if (!(segment = server->GetFreeSegment(transport_id, blockId))) - { - //DMSG(2, "NormObject::HandleObjectMessage() node>%lu server>%lu obj>%hu " - // "Warning! no free segments ...\n", LocalNodeId(), server->GetId(), - // (UINT16)transport_id); - // (TBD) Dump the block ...??? - return; - } + server->SetErasureLoc(erasureCount++, nextSegment); + segment = server->GetRetrievalSegment(); + ASSERT(segment); UINT16 payloadMax = segment_size + NormDataMsg::GetStreamPayloadHeaderLength(); #ifdef SIMULATE payloadMax = MIN(payloadMax, SIM_PAYLOAD_MAX); #endif // SIMULATE + // Zeroize the missing segment payload in prep for decoding memset(segment, 0, payloadMax+1); - block->SetSegment(nextErasure, segment); + block->SetSegment(nextSegment, segment); } - nextErasure++; - } while (block->GetNextPending(nextErasure)); - } - } + else if (!block->GetSegment(nextSegment)) + { + 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) { server->Decode(block->SegmentList(), numData, erasureCount); @@ -1215,12 +1244,13 @@ void NormObject::HandleObjectMessage(const NormObjectMsg& msg, NormSegmentId sid = server->GetErasureLoc(i); if (sid < numData) { - if (WriteSegment(blockId, sid, block->Segment(sid))) + if (WriteSegment(blockId, sid, block->GetSegment(sid), false)) { objectUpdated = true; // For statistics only (TBD) #ifdef NORM_DEBUG server->IncrementRecvGoodput(segmentLength); } + block->DetachSegment(sid); } 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 pending_mask.Unset(blockId); block_buffer.Remove(block); @@ -1258,6 +1291,28 @@ void NormObject::HandleObjectMessage(const NormObjectMsg& msg, } // end if/else (NORM_MSG_INFO) } // 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 @@ -1399,7 +1454,7 @@ bool NormObject::NextServerMsg(NormObjectMsg* msg) { 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()); return false; } @@ -1419,7 +1474,7 @@ bool NormObject::NextServerMsg(NormObjectMsg* msg) } 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()); session.ServerPutFreeBlock(block); return false; @@ -1461,7 +1516,8 @@ bool NormObject::NextServerMsg(NormObjectMsg* msg) { // Try to read data segment (Note "ReadSegment" copies in offset/length info also) char* buffer = data->AccessPayload(); - UINT16 payloadLength = ReadSegment(blockId, segmentId, buffer); + bool msgStart = false; + UINT16 payloadLength = ReadSegment(blockId, segmentId, buffer, &msgStart); if (0 == payloadLength) { // (TBD) deal with read error @@ -1472,23 +1528,16 @@ bool NormObject::NextServerMsg(NormObjectMsg* msg) } data->SetPayloadLength(payloadLength); - if (IsStream()) + // Look for FLAG_MSG_START + if (msgStart) { - // Look for FLAG_MSG_START - UINT16 payloadMax = segment_size+NormDataMsg::GetStreamPayloadHeaderLength(); -#ifdef SIMULATE - payloadMax = MIN(payloadMax, SIM_PAYLOAD_MAX); -#endif // SIMULATE - if (buffer[payloadMax]) - { - 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"); - } + 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 @@ -1516,7 +1565,7 @@ bool NormObject::NextServerMsg(NormObjectMsg* msg) CalculateBlockParity(block); } ASSERT(block->ParityReady(numData)); - char* segment = block->Segment(segmentId); + char* segment = block->GetSegment(segmentId); ASSERT(segment); UINT16 payloadLength = segment_size; if (IsStream()) payloadLength += NormDataMsg::GetStreamPayloadHeaderLength(); @@ -1692,7 +1741,7 @@ bool NormFileObject::Open(const char* thePath, } else { - if (file.Open(thePath, O_WRONLY | O_CREAT | O_TRUNC)) + if (file.Open(thePath, O_RDWR | O_CREAT | O_TRUNC)) { file.Lock(); } @@ -1769,7 +1818,8 @@ void NormFileObject::Close() bool NormFileObject::WriteSegment(NormBlockId blockId, NormSegmentId segmentId, - const char* buffer) + const char* buffer, + bool /*msgStart*/) { UINT16 len; if (blockId == final_block_id) @@ -1809,7 +1859,8 @@ bool NormFileObject::WriteSegment(NormBlockId blockId, UINT16 NormFileObject::ReadSegment(NormBlockId blockId, NormSegmentId segmentId, - char* buffer) + char* buffer, + bool* /*msgStart*/) { // Determine segment length from blockId::segmentId UINT16 len; @@ -1851,6 +1902,33 @@ UINT16 NormFileObject::ReadSegment(NormBlockId blockId, return (len == nbytes) ? len : 0; } // 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 @@ -1924,7 +2002,8 @@ void NormDataObject::Close() bool NormDataObject::WriteSegment(NormBlockId blockId, NormSegmentId segmentId, - const char* buffer) + const char* buffer, + bool /*msgStart*/) { UINT16 len; if (blockId == final_block_id) @@ -1964,7 +2043,8 @@ bool NormDataObject::WriteSegment(NormBlockId blockId, UINT16 NormDataObject::ReadSegment(NormBlockId blockId, NormSegmentId segmentId, - char* buffer) + char* buffer, + bool* /*msgStart*/) { // Determine segment length from blockId::segmentId UINT16 len; @@ -2000,6 +2080,58 @@ UINT16 NormDataObject::ReadSegment(NormBlockId blockId, return true; } // 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), stream_sync(false), sync_offset_valid(false), 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(); } - NormObjectSize blockSize((UINT32)segmentSize * (UINT32)numData); - NormObjectSize numBlocks = NormObjectSize((UINT32)bufferSize) / blockSize; - ASSERT(0 == numBlocks.MSB()); + UINT32 blockSize = segmentSize * numData; + UINT32 numBlocks = bufferSize / blockSize; + // Buffering requires at least 2 blocks - numBlocks = MAX(2, numBlocks.LSB()); - UINT32 numSegments = numBlocks.LSB() * numData; + numBlocks = MAX(2, numBlocks); + 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"); Close(); @@ -2065,17 +2198,19 @@ bool NormStreamObject::Open(UINT32 bufferSize, return false; } - if (!stream_buffer.Init(numBlocks.LSB())) + if (!stream_buffer.Init(numBlocks)) { DMSG(0, "NormStreamObject::Open() stream_buffer init error\n"); Close(); 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 - write_index.block = write_index.segment = 0; + read_init = true; read_index.block = read_index.segment = 0; + write_index.block = write_index.segment = 0; write_offset = read_offset = 0; + if (!server) { if (!NormObject::Open(NormObjectSize((UINT32)bufferSize), @@ -2240,16 +2375,32 @@ bool NormStreamObject::StreamUpdateStatus(NormBlockId blockId) stream_sync = true; stream_sync_id = blockId; stream_next_id = blockId + pending_mask.Size(); - read_index.block = blockId; - read_index.segment = 0; + read_init = true; return true; } } // 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, NormSegmentId segmentId, - char* buffer) + char* buffer, + bool* msgStart) { // (TBD) compare blockId with stream_buffer.RangeLo() and stream_buffer.RangeHi() NormBlock* block = stream_buffer.Find(blockId); @@ -2266,7 +2417,7 @@ UINT16 NormStreamObject::ReadSegment(NormBlockId blockId, return false; } block->UnsetPending(segmentId); - char* segment = block->Segment(segmentId); + char* segment = block->GetSegment(segmentId); ASSERT(segment != NULL); UINT16 segmentLength = NormDataMsg::ReadStreamPayloadLength(segment); ASSERT(segmentLength <= segment_size); @@ -2279,29 +2430,41 @@ UINT16 NormStreamObject::ReadSegment(NormBlockId blockId, #else memcpy(buffer, segment, payloadLength); #endif // SIMULATE - buffer[payloadMax] = segment[payloadMax]; + if (msgStart) + *msgStart = (0 != (NormDataMsg::FLAG_MSG_START & segment[payloadMax])); return payloadLength; } // end NormStreamObject::ReadSegment() bool NormStreamObject::WriteSegment(NormBlockId blockId, 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) || ((blockId == read_index.block) && (segmentId < read_index.segment))) { DMSG(4, "NormStreamObject::WriteSegment() block/segment < read_index!?\n"); return false; - } - - UINT32 segmentOffset = NormDataMsg::ReadStreamPayloadOffset(segment); + } + // if (segmentOffset < read_offset) UINT32 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); + ASSERT(0); return false; } @@ -2309,22 +2472,25 @@ bool NormStreamObject::WriteSegment(NormBlockId blockId, if (!block) { bool broken = false; - // Prune (if necessary) stream_buffer (stream might be broken) - while (!stream_buffer.CanInsert(blockId) || block_pool.IsEmpty()) + bool dataLost = false; + while (block_pool.IsEmpty() || !stream_buffer.CanInsert(blockId)) { block = stream_buffer.Find(stream_buffer.RangeLo()); - if (block->IsPending()) broken = true; - // This loop feeds any received user data segments to the application - // (if the application doesn't want the data, it's lost!) - bool dataLost = false; + ASSERT(block); + if (blockId < block->GetId()) + { + DMSG(4, "NormStreamObject::WriteSegment() blockId too old!?\n"); + return false; + } 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(); block->GetFirstPending(read_index.segment); NormBlock* tempBlock = block; - UINT32 tempOffset = read_offset; - session.Notify(NormController::RX_OBJECT_UPDATE, server, this); + UINT32 tempOffset = read_offset; + session.Notify(NormController::RX_OBJECT_UPDATE, server, this); block = stream_buffer.Find(stream_buffer.RangeLo()); if (tempBlock == block) { @@ -2333,48 +2499,57 @@ bool NormStreamObject::WriteSegment(NormBlockId blockId, // App didn't want any data here, purge segment //ASSERT(0); dataLost = true; - char* s = block->DetachSegment(read_index.segment); - segment_pool.Put(s); - block->UnsetPending(read_index.segment); + block->UnsetPending(read_index.segment++); + if (read_index.segment >= ndata) + { + 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 { - // App consumed block via Read() - block = NULL; - break; - } - } - if (dataLost) - DMSG(0, "NormStreamObject::WriteSegment() broken stream data dropped!\n"); - + // App consumed block via Read() + block = NULL; + break; + } + } // end while (block->IsPending()) if (block) { - // if the app didn't consume the block, we must - // return it to the pool - ASSERT(!block->IsPending()); + if (block->GetId() == read_index.block) + { + read_index.block++; + read_index.segment = 0; + Prune(read_index.block); + } stream_buffer.Remove(block); block->EmptyToPool(segment_pool); block_pool.Put(block); - } - } - block = block_pool.Get(); - block->SetId(blockId); - block->ClearPending(); - bool success = stream_buffer.Insert(block); - ASSERT(success); + } + } // end while (block_pool.IsEmpty() || !stream_buffer.CanInsert(blockId)) if (broken) { DMSG(4, "NormStreamObject::WriteSegment() node>%lu obj>%hu blk>%lu seg>%hu broken stream ...\n", LocalNodeId(), (UINT16)transport_id, (UINT32)blockId, (UINT16)segmentId); - NormBlock* first = stream_buffer.Find(stream_buffer.RangeLo()); - read_index.block = first->GetId(); - read_index.segment = 0; - } - } + if (dataLost) + DMSG(0, "NormStreamObject::WriteSegment() broken stream data not read by app!\n"); + } + 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 - if(!block->Segment(segmentId)) + if(!block->GetSegment(segmentId)) { char* s = segment_pool.Get(); ASSERT(s != NULL); // for now, this should always succeed @@ -2385,11 +2560,12 @@ bool NormStreamObject::WriteSegment(NormBlockId blockId, payloadLength = MIN(payloadMax, payloadLength); #endif // SIMULATE 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->SetPending(segmentId); - ASSERT(block->Segment(segmentId) == s); + ASSERT(block->GetSegment(segmentId) == s); } if (!sync_offset_valid) { @@ -2428,8 +2604,10 @@ void NormStreamObject::Prune(NormBlockId blockId) pending_mask.UnsetBits(firstId, count); } } + bool result = StreamUpdateStatus(blockId); + ASSERT(result); if (resync) server->IncrementResyncCount(); - StreamUpdateStatus(blockId); + result = GetFirstPending(firstId); } // end NormStreamObject::Prune() // 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()); *buflen = bytesRead; if (bytesRead > 0) + { return true; + } 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) { //DMSG(0, "NormStreamObject::Read(%lu:%hu) stream buffer empty (2)\n", // (UINT32)read_index.block, read_index.segment); *buflen = bytesRead; if (bytesRead > 0) + { return true; + } 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); @@ -2509,8 +2729,10 @@ bool NormStreamObject::Read(char* buffer, unsigned int* buflen, bool findMsgStar } if (!msgStart) { - block->DetachSegment(read_index.segment); - segment_pool.Put(segment); + // Don't bother managing individual segments since + // stream buffers are exact multiples of block size! + //block->DetachSegment(read_index.segment); + //segment_pool.Put(segment); block->UnsetPending(read_index.segment++); if (read_index.segment >= ndata) { @@ -2519,6 +2741,7 @@ bool NormStreamObject::Read(char* buffer, unsigned int* buflen, bool findMsgStar block_pool.Put(block); read_index.block++; read_index.segment = 0; + Prune(read_index.block); } continue; } @@ -2538,8 +2761,10 @@ bool NormStreamObject::Read(char* buffer, unsigned int* buflen, bool findMsgStar bytesToRead -= count; if (index >= length) { - block->DetachSegment(read_index.segment); - segment_pool.Put(segment); + // Don't bother managing individual segments since + // stream buffers are multiples of block size! + // block->DetachSegment(read_index.segment); + // segment_pool.Put(segment); block->UnsetPending(read_index.segment++); if (read_index.segment >= ndata) { @@ -2548,6 +2773,7 @@ bool NormStreamObject::Read(char* buffer, unsigned int* buflen, bool findMsgStar block_pool.Put(block); read_index.block++; read_index.segment = 0; + Prune(read_index.block); } } } while (bytesToRead > 0); // end while (len > 0) @@ -2555,6 +2781,7 @@ bool NormStreamObject::Read(char* buffer, unsigned int* buflen, bool findMsgStar return true; } // end NormStreamObject::Read() + UINT32 NormStreamObject::Write(const char* buffer, UINT32 len, bool eom) { UINT32 nBytes = 0; @@ -2598,7 +2825,7 @@ UINT32 NormStreamObject::Write(const char* buffer, UINT32 len, bool eom) ASSERT(success); } - char* segment = block->Segment(write_index.segment); + char* segment = block->GetSegment(write_index.segment); if (!segment) { if (!(segment = segment_pool.Get())) @@ -2729,7 +2956,8 @@ NormSimObject::~NormSimObject() UINT16 NormSimObject::ReadSegment(NormBlockId blockId, NormSegmentId segmentId, - char* buffer) + char* buffer, + bool* /*msgStart*/) { // Determine segment length UINT16 len; diff --git a/common/normObject.h b/common/normObject.h index c58475e..5c4cd6a 100644 --- a/common/normObject.h +++ b/common/normObject.h @@ -84,10 +84,15 @@ class NormObject virtual bool WriteSegment(NormBlockId blockId, NormSegmentId segmentId, - const char* buffer) = 0; + const char* buffer, + bool msgStart) = 0; virtual UINT16 ReadSegment(NormBlockId blockId, 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;} void SetNackingMode(NackingMode nackingMode) @@ -201,6 +206,7 @@ class NormObject // Used by receiver for resource management scheme NormBlock* StealNewestBlock(bool excludeBlock, NormBlockId excludeId = 0); NormBlock* StealOldestBlock(bool excludeBlock, NormBlockId excludeId = 0); + bool ReclaimSourceSegments(NormSegmentPool& segmentPool); bool PassiveRepairCheck(NormBlockId blockId, NormSegmentId segmentId); bool ClientRepairCheck(CheckLevel level, @@ -287,11 +293,15 @@ class NormFileObject : public NormObject virtual bool WriteSegment(NormBlockId blockId, NormSegmentId segmentId, - const char* buffer); + const char* buffer, + bool msgStart); virtual UINT16 ReadSegment(NormBlockId blockId, NormSegmentId segmentId, - char* buffer); + char* buffer, + bool* msgStart = NULL); + virtual char* RetrieveSegment(NormBlockId blockId, + NormSegmentId segmentId); private: char path[PATH_MAX]; @@ -320,11 +330,16 @@ class NormDataObject : public NormObject virtual bool WriteSegment(NormBlockId blockId, NormSegmentId segmentId, - const char* buffer); + const char* buffer, + bool msgStart); virtual UINT16 ReadSegment(NormBlockId blockId, NormSegmentId segmentId, - char* buffer); + char* buffer, + bool* msgStart = NULL); + + virtual char* RetrieveSegment(NormBlockId blockId, + NormSegmentId segmentId); private: NormObjectSize large_block_length; @@ -359,7 +374,7 @@ class NormStreamObject : public NormObject void Flush(bool eom = false) { FlushMode oldFlushMode = flush_mode; - SetFlushMode(FLUSH_ACTIVE); + SetFlushMode((FLUSH_ACTIVE == oldFlushMode) ? FLUSH_ACTIVE : FLUSH_PASSIVE); Write(NULL, 0, eom); SetFlushMode(oldFlushMode); } @@ -381,10 +396,15 @@ class NormStreamObject : public NormObject virtual bool WriteSegment(NormBlockId blockId, NormSegmentId segmentId, - const char* buffer); + const char* buffer, + bool msgStart); virtual UINT16 ReadSegment(NormBlockId blockId, 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 @@ -407,8 +427,15 @@ class NormStreamObject : public NormObject NormBlockId GetNextBlockId() const {return (server ? read_index.block : write_index.block);} 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: + + class Index { public: @@ -427,12 +454,16 @@ class NormStreamObject : public NormObject NormBlockBuffer stream_buffer; Index write_index; UINT32 write_offset; + bool read_init; Index read_index; UINT32 read_offset; bool flush_pending; bool msg_start; FlushMode flush_mode; bool push_mode; + + // For threaded API purposes + UINT32 block_pool_threshold; }; // end class NormStreamObject #ifdef SIMULATE @@ -457,11 +488,19 @@ class NormSimObject : public NormObject virtual bool WriteSegment(NormBlockId blockId, NormSegmentId segmentId, - const char* buffer) {return true;} + const char* buffer, + bool msgStart) {return true;} virtual UINT16 ReadSegment(NormBlockId blockId, NormSegmentId segmentId, - char* buffer); + char* buffer, + bool* msgStart = NULL); + + virtual char* RetrieveSegment(NormBlockId blockId, + NormSegmentId segmentId) + { + return server ? server->GetRetrievalSegment() : NULL; + } }; // end class NormSimObject #endif // SIMULATE diff --git a/common/normSegment.cpp b/common/normSegment.cpp index 1d49149..6ad7f51 100644 --- a/common/normSegment.cpp +++ b/common/normSegment.cpp @@ -541,7 +541,7 @@ bool NormBlock::AppendRepairRequest(NormNackMsg& nack, { if (0 == nack.PackRepairRequest(req)) { - DMSG(0, "NormBlock::AppendRepairRequest() warning: full NACK msg\n"); + DMSG(3, "NormBlock::AppendRepairRequest() warning: full NACK msg\n"); break; } } @@ -573,13 +573,13 @@ bool NormBlock::AppendRepairRequest(NormNackMsg& nack, if (NormRepairRequest::INVALID != prevForm) { 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; } // end NormBlock::AppendRepairRequest() 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(); } -bool NormBlockPool::Init(UINT32 numBlocks, UINT16 totalSize) +bool NormBlockPool::Init(UINT32 numBlocks, UINT16 segsPerBlock) { if (head) Destroy(); for (UINT32 i = 0; i < numBlocks; i++) @@ -596,7 +596,7 @@ bool NormBlockPool::Init(UINT32 numBlocks, UINT16 totalSize) NormBlock* b = new NormBlock(); if (b) { - if (!b->Init(totalSize)) + if (!b->Init(segsPerBlock)) { DMSG(0, "NormBlockPool::Init() block init error\n"); delete b; @@ -613,6 +613,7 @@ bool NormBlockPool::Init(UINT32 numBlocks, UINT16 totalSize) return false; } } + count = numBlocks; return true; } // end NormBlockPool::Init() @@ -624,6 +625,7 @@ void NormBlockPool::Destroy() head = next->next; delete next; } + count = 0; } // end NormBlockPool::Destroy() NormBlockBuffer::NormBlockBuffer() diff --git a/common/normSegment.h b/common/normSegment.h index b765610..377fab8 100644 --- a/common/normSegment.h +++ b/common/normSegment.h @@ -69,7 +69,7 @@ class NormBlock void SetParityReadiness(UINT16 ndata) {erasure_count = ndata;} char** SegmentList(UINT16 index = 0) {return &segment_table[index];} - char* Segment(NormSegmentId sid) + char* GetSegment(NormSegmentId sid) { ASSERT(sid < size); return segment_table[sid]; @@ -152,6 +152,7 @@ class NormBlock // Note: This invalidates the repair_mask state. bool IsRepairPending(UINT16 ndata, UINT16 nparity); void DecrementErasureCount() {erasure_count--;} + void IncrementErasureCount() {erasure_count++;} UINT16 ErasureCount() const {return erasure_count;} void IncrementParityCount() {parity_count++;} UINT16 ParityCount() const {return parity_count;} @@ -257,6 +258,7 @@ class NormBlockPool head = b ? b->next : NULL; if (b) { + count--; overrun_flag = false; } else if (!overrun_flag) @@ -271,11 +273,14 @@ class NormBlockPool { b->next = head; head = b; + count++; } unsigned long OverrunCount() const {return overruns;} + UINT32 GetCount() {return count;} private: NormBlock* head; + UINT32 count; unsigned long overruns; bool overrun_flag; }; // end class NormBlockPool diff --git a/common/normSession.cpp b/common/normSession.cpp index 563b6cb..b437dc1 100644 --- a/common/normSession.cpp +++ b/common/normSession.cpp @@ -43,6 +43,7 @@ NormSession::NormSession(NormSessionMgr& sessionMgr, NormNodeId localNodeId) trace(false), tx_loss_rate(0.0), rx_loss_rate(0.0), next(NULL) { + interface_name[0] = '\0'; tx_socket.SetNotifier(&sessionMgr.GetSocketNotifier()); tx_socket.SetListener(this, &NormSession::TxSocketRecvHandler); @@ -133,6 +134,8 @@ bool NormSession::Open(const char* interfaceName) { rx_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++) @@ -165,7 +168,12 @@ void NormSession::Close() if (tx_socket.IsOpen()) tx_socket.Close(); 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(); } } // end NormSession::Close() @@ -654,7 +662,7 @@ void NormSession::ServerQueueFlush() else { // 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)) { flush_count++; @@ -1395,7 +1403,6 @@ void NormSession::ServerUpdateGrttEstimate(double clientRtt) // Calculate grtt_advertised since quantization rounds upward grtt_advertised = NormUnquantizeRtt(grtt_quantized); - double clrRtt = cc_node_list.Head() ? ((NormCCNode*)cc_node_list.Head())->GetRtt() : -1; if (grttQuantizedOld != grtt_quantized) DMSG(4, "NormSession::ServerUpdateGrttEstimate() node>%lu new grtt>%lf sec\n", LocalNodeId(), grtt_advertised); @@ -2122,9 +2129,10 @@ void NormSession::ServerHandleNackMessage(const struct timeval& currentTime, Nor if (startTimer && !repair_timer.IsActive()) { // BACKOFF related code - double aggregateInterval = address.IsMulticast() ? grtt_advertised * (backoff_factor + 1.0) : - 0.0; - aggregateInterval = (backoff_factor > 0.0) ? aggregateInterval : 0.0;//grtt_advertised / 100.0; + double aggregateInterval = address.IsMulticast() ? + grtt_advertised * (backoff_factor + 1.0) : 0.0; + // backoff == 0.0 is a special case + //aggregateInterval = (backoff_factor > 0.0) ? aggregateInterval : 0.0; if (tx_timer.IsActive()) { diff --git a/common/normSession.h b/common/normSession.h index 76562ff..032579c 100644 --- a/common/normSession.h +++ b/common/normSession.h @@ -339,7 +339,8 @@ class NormSession NormNodeId local_node_id; ProtoAddress address; // session destination address & port 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 backoff_factor; diff --git a/common/normTest.cpp b/common/normTest.cpp index 0929bf6..6d9f506 100644 --- a/common/normTest.cpp +++ b/common/normTest.cpp @@ -50,7 +50,7 @@ int main(int argc, char* argv[]) NormStartReceiver(session, 1024*1024); // 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)); @@ -59,8 +59,12 @@ int main(int argc, char* argv[]) const char* fileName = "ferrari.jpg"; // Uncomment this line to send a stream instead of the file - stream = NormOpenStream(session, 1024*1024); - NormSetStreamFlushMode(stream, NORM_FLUSH_PASSIVE); + stream = NormStreamOpen(session, 1024*1024); + + + // NORM_FLUSH_PASSIVE automatically flushes full writes to + // the stream. + NormStreamSetFlushMode(stream, NORM_FLUSH_PASSIVE); // 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); txLen = strlen(txBuffer); } - txIndex += NormWriteStream(stream, txBuffer+txIndex, (txLen - txIndex)); + txIndex += NormStreamWrite(stream, txBuffer+txIndex, (txLen - txIndex)); 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; sendCount++; } @@ -102,10 +112,10 @@ int main(int argc, char* argv[]) else { NormObjectHandle txFile = - NormQueueFile(session, - filePath, - fileName, - strlen(fileName)); + NormFileEnqueue(session, + filePath, + fileName, + strlen(fileName)); // Repeatedly queue our file for sending if (NORM_OBJECT_INVALID == txFile) { @@ -131,12 +141,12 @@ int main(int argc, char* argv[]) case NORM_RX_OBJECT_INFO: // Assume info contains '/' delimited 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)) { - //NormCancelObject(theEvent.object); + //NormObjectCancel(theEvent.object); //break; } @@ -144,7 +154,7 @@ int main(int argc, char* argv[]) strcpy(fileName, cachePath); int pathLen = strlen(fileName); unsigned short nameLen = PATH_MAX - pathLen; - NormGetObjectInfo(theEvent.object, fileName+pathLen, &nameLen); + NormObjectGetInfo(theEvent.object, fileName+pathLen, &nameLen); fileName[nameLen + pathLen] = '\0'; char* ptr = fileName + 5; while ('\0' != *ptr) @@ -152,7 +162,7 @@ int main(int argc, char* argv[]) if ('/' == *ptr) *ptr = PROTO_PATH_DELIMITER; ptr++; } - if (!NormSetFileName(theEvent.object, fileName)) + if (!NormFileRename(theEvent.object, fileName)) TRACE("normTest: NormSetFileName(%s) error\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: { //TRACE("normTest: NORM_RX_OBJECT_UPDATE event ...\n"); - if (NORM_OBJECT_STREAM != NormGetObjectType(theEvent.object)) + if (NORM_OBJECT_STREAM != NormObjectGetType(theEvent.object)) break; unsigned int len; do { if (!msgSync) - msgSync = NormFindStreamMsgStart(theEvent.object); + msgSync = NormStreamSeekMsgStart(theEvent.object); if (!msgSync) break; len = 8191 - rxIndex; - if (NormReadStream(theEvent.object, rxBuffer+rxIndex, &len)) + if (NormStreamRead(theEvent.object, rxBuffer+rxIndex, &len)) { rxIndex += len; 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 #endif // if/else WIN32/UNIX - NormCloseStream(stream); + NormStreamClose(stream); NormStopReceiver(session); NormStopSender(session); NormDestroySession(session); diff --git a/common/normVersion.h b/common/normVersion.h index 2b7cf13..0be29a5 100644 --- a/common/normVersion.h +++ b/common/normVersion.h @@ -32,6 +32,6 @@ #ifndef _NORM_VERSION #define _NORM_VERSION -#define VERSION "1.2b3" +#define VERSION "1.2b4" #endif // _NORM_VERSION