diff --git a/NormUserGuide.doc b/NormUserGuide.doc new file mode 100644 index 0000000..bd59f08 Binary files /dev/null and b/NormUserGuide.doc differ diff --git a/NormUserGuide.pdf b/NormUserGuide.pdf new file mode 100644 index 0000000..c534c29 Binary files /dev/null and b/NormUserGuide.pdf differ diff --git a/TODO.TXT b/TODO.TXT index 4b9f48d..bc4d224 100644 --- a/TODO.TXT +++ b/TODO.TXT @@ -41,7 +41,7 @@ results in a NormNode attempting to be assigned NORM_NODE_NONE as a node id. -14) IMPORANT: double check ACKing in reponse to watermark polling and +14) IMPORTANT: double check ACKing in reponse to watermark polling and relation to cc feedback suppression, etc ... watermark acks should be unicast and not effect cc feedback? diff --git a/VERSION.TXT b/VERSION.TXT index 0f9718d..203c073 100644 --- a/VERSION.TXT +++ b/VERSION.TXT @@ -1,5 +1,30 @@ NORM Version History +Version 1.4b3 +============= + - Fixed many (hopefully all) cases where NORM_DATA message FLAG_REPAIR + was not always being set when it should resulting in receivers + sometimes "syncing" (or "re-syncing") to sender repair transmissions + when they shouldn't ... sometimes resulting in duplicative stream + data being delivered to the receiver app. + - Fixed case when repair request for old NORM_OBJECT_STREAM segment could + lockup advancement of stream when stream "push" is enabled. + (Thanks to Greg Lauer, Clifton Lin, BAE Systems CBMANET team) + - Added command-line options to "norm" demo app for "txrobustfactor" + and "rxrobustfactor" + - Added "NormNodeFreeBuffers()" API call and changed NORM behavior + to require app to free buffers allocate for remote sender unless + indefinite persistence is desired. + - Added "rxpersist" command-line option to "norm" demo app to make + "norm" demo app receivers keep full state on remote senders' + indefinitely. + - Fixed issue with NormSession::QueueTxObject() where ASSERT() failure + occurred depending upon previous tx object enqueue/dequeue history + (Thanks to Jeff Bush for finding this one) + - Changed repeated scheduling of receiver remote sender inactivity + timer to ensure that busy CPU doesn't result in premature + inactivity indication. + Version 1.4b2 ============= - Fixed bug where NormObjectCancel() usage could result in extra @@ -17,6 +42,7 @@ Version 1.4b2 (Thanks to Peter Lu) - Fixed case when repair request for old NORM_OBJECT_STREAM segment could lockup advancement of stream. + (Also thanks to Peter Lu) Version 1.4b1 ============= diff --git a/common/normApi.cpp b/common/normApi.cpp index 4c668d2..4a53a92 100644 --- a/common/normApi.cpp +++ b/common/normApi.cpp @@ -46,12 +46,12 @@ class NormInstance : public NormController bool Startup(bool priorityBoost = false); void Shutdown(); - void Pause() // pause NORM protocol engine + void Stop() // pause NORM protocol engine { dispatcher.Stop(); Notify(NormController::EVENT_INVALID, &session_mgr, NULL, NULL, NULL); } - bool Resume() + bool Start() { if (dispatcher.StartThread(priority_boost)) { @@ -73,7 +73,7 @@ class NormInstance : public NormController void PurgeObjectNotifications(NormObjectHandle objectHandle); - unsigned long CountCompletedObjects(NormSession* theSession); + UINT32 CountCompletedObjects(NormSession* theSession); ProtoDispatcher::Descriptor GetDescriptor() const { @@ -641,9 +641,9 @@ void NormInstance::Shutdown() notify_pool.Destroy(); } // end NormInstance::Shutdown() -unsigned long NormInstance::CountCompletedObjects(NormSession* session) +UINT32 NormInstance::CountCompletedObjects(NormSession* session) { - unsigned long result = 0UL; + UINT32 result = 0UL; Notification* n = notify_queue.GetHead(); while (NULL != n) { @@ -683,6 +683,13 @@ void NormDestroyInstance(NormInstanceHandle instanceHandle) delete ((NormInstance*)instanceHandle); } // end NormDestroyInstance() +NORM_API_LINKAGE +void NormStopInstance(NormInstanceHandle instanceHandle) +{ + NormInstance* instance = (NormInstance*)instanceHandle; + if (instance) instance->Stop(); // stops NORM protocol thread +} // end NormStopInstance() + NORM_API_LINKAGE bool NormRestartInstance(NormInstanceHandle instanceHandle) { @@ -690,19 +697,29 @@ bool NormRestartInstance(NormInstanceHandle instanceHandle) if (instance) { if (instance->dispatcher.IsThreaded()) - instance->Pause(); - return instance->Resume(); + instance->Stop(); + return instance->Start(); } return false; // invalid instanceHandle } // end NormStartInstance() NORM_API_LINKAGE -void NormStopInstance(NormInstanceHandle instanceHandle) +bool NormSuspendInstance(NormInstanceHandle instanceHandle) { NormInstance* instance = (NormInstance*)instanceHandle; - if (instance) instance->Pause(); // stops NORM protocol thread -} // end NormStopInstance() + if (instance) + return instance->dispatcher.SuspendThread(); // stops NORM protocol thread + else + return false; +} // end NormSuspendInstance() + +NORM_API_LINKAGE +void NormResumeInstance(NormInstanceHandle instanceHandle) +{ + NormInstance* instance = (NormInstance*)instanceHandle; + if (instance) instance->dispatcher.ResumeThread(); +} // end NormResumeInstance() NORM_API_LINKAGE @@ -756,7 +773,7 @@ bool NormGetNextEvent(NormInstanceHandle instanceHandle, NormEvent* theEvent) NORM_API_LINKAGE NormSessionHandle NormCreateSession(NormInstanceHandle instanceHandle, const char* sessionAddr, - unsigned short sessionPort, + UINT16 sessionPort, NormNodeId localNodeId) { // (TBD) wrap this with SuspendThread/ResumeThread ??? @@ -834,7 +851,7 @@ NormNodeId NormGetLocalNodeId(NormSessionHandle sessionHandle) NORM_API_LINKAGE void NormSetTxPort(NormSessionHandle sessionHandle, - unsigned short txPort) + UINT16 txPort) { NormInstance* instance = NormInstance::GetInstanceFromSession(sessionHandle); if (NULL != instance) @@ -961,8 +978,8 @@ void NormSetRxLoss(NormSessionHandle sessionHandle, double percent) NORM_API_LINKAGE bool NormStartSender(NormSessionHandle sessionHandle, NormSessionId sessionId, - unsigned long bufferSpace, - unsigned short segmentSize, + UINT32 bufferSpace, + UINT16 segmentSize, unsigned char numData, unsigned char numParity) { @@ -1060,8 +1077,8 @@ void NormSetTransmitRateBounds(NormSessionHandle sessionHandle, NORM_API_LINKAGE void NormSetTransmitCacheBounds(NormSessionHandle sessionHandle, NormSize sizeMax, - unsigned long countMin, - unsigned long countMax) + UINT32 countMin, + UINT32 countMax) { NormInstance* instance = NormInstance::GetInstanceFromSession(sessionHandle); if (instance && instance->dispatcher.SuspendThread()) @@ -1220,7 +1237,7 @@ NormObjectHandle NormFileEnqueue(NormSessionHandle sessionHandle, NORM_API_LINKAGE NormObjectHandle NormDataEnqueue(NormSessionHandle sessionHandle, const char* dataPtr, - unsigned long dataLen, + UINT32 dataLen, const char* infoPtr, unsigned int infoLen) { @@ -1261,7 +1278,7 @@ bool NormRequeueObject(NormSessionHandle sessionHandle, NormObjectHandle objectH NORM_API_LINKAGE NormObjectHandle NormStreamOpen(NormSessionHandle sessionHandle, - unsigned long bufferSize) + UINT32 bufferSize) { NormObjectHandle objectHandle = NORM_OBJECT_INVALID; NormInstance* instance = NormInstance::GetInstanceFromSession(sessionHandle); @@ -1483,7 +1500,7 @@ NormAckingStatus NormGetAckingStatus(NormSessionHandle sessionHandle, NORM_API_LINKAGE bool NormStartReceiver(NormSessionHandle sessionHandle, - unsigned long bufferSpace) + UINT32 bufferSpace) { bool result = false; NormInstance* instance = NormInstance::GetInstanceFromSession(sessionHandle); @@ -1662,7 +1679,7 @@ bool NormStreamSeekMsgStart(NormObjectHandle streamHandle) NORM_API_LINKAGE -unsigned long NormStreamGetReadOffset(NormObjectHandle streamHandle) +UINT32 NormStreamGetReadOffset(NormObjectHandle streamHandle) { NormStreamObject* stream = static_cast((NormObject*)streamHandle); if (stream) @@ -1690,17 +1707,17 @@ bool NormObjectHasInfo(NormObjectHandle objectHandle) } // end NormObjectHasInfo() NORM_API_LINKAGE -unsigned short NormObjectGetInfoLength(NormObjectHandle objectHandle) +UINT16 NormObjectGetInfoLength(NormObjectHandle objectHandle) { return ((NormObject*)objectHandle)->GetInfoLength(); } // end NormObjectGetInfoLength() NORM_API_LINKAGE -unsigned short NormObjectGetInfo(NormObjectHandle objectHandle, +UINT16 NormObjectGetInfo(NormObjectHandle objectHandle, char* buffer, - unsigned short bufferLen) + UINT16 bufferLen) { - unsigned short result = 0; + UINT16 result = 0; if (NORM_OBJECT_INVALID != objectHandle) { NormObject* object = (NormObject*)objectHandle; @@ -1863,7 +1880,7 @@ NORM_API_LINKAGE bool NormNodeGetAddress(NormNodeHandle nodeHandle, char* addrBuffer, unsigned int* bufferLen, - unsigned short* port) + UINT16* port) { bool result = false; if (NORM_NODE_INVALID != nodeHandle) @@ -1901,6 +1918,21 @@ double NormNodeGetGrtt(NormNodeHandle nodeHandle) return -1.0; } // end NormNodeGetGrtt() +NORM_API_LINKAGE +void NormNodeFreeBuffers(NormNodeHandle nodeHandle) +{ + if (NORM_NODE_INVALID != nodeHandle) + { + NormInstance* instance = NormInstance::GetInstanceFromNode(nodeHandle); + if (instance && instance->dispatcher.SuspendThread()) + { + NormServerNode* node = (NormServerNode*)nodeHandle; + node->FreeBuffers(); + instance->dispatcher.ResumeThread(); + } + } +} // end NormNodeFreeBuffers() + NORM_API_LINKAGE void NormNodeRetain(NormNodeHandle nodeHandle) { @@ -1947,9 +1979,9 @@ void NormNodeRelease(NormNodeHandle nodeHandle) // NORM_API_LINKAGE -unsigned long NormCountCompletedObjects(NormSessionHandle sessionHandle) +UINT32 NormCountCompletedObjects(NormSessionHandle sessionHandle) { - unsigned long result = 0; + UINT32 result = 0; NormSession* session = (NormSession*)sessionHandle; NormInstance* instance = NormInstance::GetInstanceFromSession(sessionHandle); if (instance && instance->dispatcher.SuspendThread()) @@ -1958,4 +1990,4 @@ unsigned long NormCountCompletedObjects(NormSessionHandle sessionHandle) instance->dispatcher.ResumeThread(); } return result; -} // end long NormCountCompletedObjects() +} // end NormCountCompletedObjects() diff --git a/common/normApi.h b/common/normApi.h index a9ca6f5..fff5cec 100644 --- a/common/normApi.h +++ b/common/normApi.h @@ -10,6 +10,7 @@ #ifdef WIN32 #include #include +#include // for UINT32/INT32, etc types #ifdef NORM_USE_DLL #ifdef _NORM_API_BUILD #define NORM_API_LINKAGE __declspec(dllexport) // to support building of "Norm.dll" @@ -21,6 +22,17 @@ #endif // if/else NORM_USE_DLL #else #include // for "off_t" +#include // for proper uint32_t, etc definitions +typedef int8_t INT8; +typedef int16_t INT16; +#ifdef _USING_X11 +typedef long int INT32; +#else +typedef int32_t INT32; +#endif // if/else _USING_X11 +typedef uint8_t UINT8; +typedef uint16_t UINT16; +typedef uint32_t UINT32; #define NORM_API_LINKAGE #endif // if/else WIN32/UNIX @@ -43,12 +55,12 @@ const NormInstanceHandle NORM_INSTANCE_INVALID; typedef const void* NormSessionHandle; extern NORM_API_LINKAGE const NormSessionHandle NORM_SESSION_INVALID; -typedef unsigned short NormSessionId; +typedef UINT16 NormSessionId; typedef const void* NormNodeHandle; extern NORM_API_LINKAGE const NormNodeHandle NORM_NODE_INVALID; -typedef unsigned long NormNodeId; +typedef UINT32 NormNodeId; extern NORM_API_LINKAGE const NormNodeId NORM_NODE_NONE; extern NORM_API_LINKAGE @@ -57,7 +69,7 @@ const NormNodeId NORM_NODE_ANY; typedef const void* NormObjectHandle; extern NORM_API_LINKAGE const NormObjectHandle NORM_OBJECT_INVALID; -typedef unsigned short NormObjectTransportId; +typedef UINT16 NormObjectTransportId; #ifdef WIN32 typedef __int64 NormSize; @@ -161,6 +173,14 @@ void NormStopInstance(NormInstanceHandle instanceHandle); NORM_API_LINKAGE bool NormRestartInstance(NormInstanceHandle instanceHandle); +NORM_API_LINKAGE +bool NormSuspendInstance(NormInstanceHandle instanceHandle); + +NORM_API_LINKAGE +void NormResumeInstance(NormInstanceHandle instanceHandle); + + + // This MUST be set to enable NORM_OBJECT_FILE reception! // (otherwise received files are ignored) NORM_API_LINKAGE @@ -195,7 +215,7 @@ NormDescriptor NormGetDescriptor(NormInstanceHandle instanceHandle); NORM_API_LINKAGE NormSessionHandle NormCreateSession(NormInstanceHandle instanceHandle, const char* sessionAddress, - unsigned short sessionPort, + UINT16 sessionPort, NormNodeId localNodeId); NORM_API_LINKAGE @@ -212,7 +232,7 @@ NormNodeId NormGetLocalNodeId(NormSessionHandle sessionHandle); NORM_API_LINKAGE void NormSetTxPort(NormSessionHandle sessionHandle, - unsigned short txPortNumber); + UINT16 txPortNumber); NORM_API_LINKAGE void NormSetRxPortReuse(NormSessionHandle sessionHandle, @@ -248,8 +268,8 @@ void NormSetRxLoss(NormSessionHandle sessionHandle, double percent); NORM_API_LINKAGE bool NormStartSender(NormSessionHandle sessionHandle, NormSessionId sessionId, - unsigned long bufferSpace, - unsigned short segmentSize, + UINT32 bufferSpace, + UINT16 segmentSize, unsigned char numData, unsigned char numParity); @@ -278,9 +298,9 @@ void NormSetTransmitRateBounds(NormSessionHandle sessionHandle, NORM_API_LINKAGE void NormSetTransmitCacheBounds(NormSessionHandle sessionHandle, - NormSize sizeMax, - unsigned long countMin, - unsigned long countMax); + NormSize sizeMax, + UINT32 countMin, + UINT32 countMax); NORM_API_LINKAGE void NormSetAutoParity(NormSessionHandle sessionHandle, @@ -327,7 +347,7 @@ NormObjectHandle NormFileEnqueue(NormSessionHandle sessionHandle, NORM_API_LINKAGE NormObjectHandle NormDataEnqueue(NormSessionHandle sessionHandle, const char* dataPtr, - unsigned long dataLen, + UINT32 dataLen, const char* infoPtr = (const char*)0, unsigned int infoLen = 0); @@ -336,7 +356,7 @@ bool NormRequeueObject(NormSessionHandle sessionHandle, NormObjectHandle objectH NORM_API_LINKAGE NormObjectHandle NormStreamOpen(NormSessionHandle sessionHandle, - unsigned long bufferSize); + UINT32 bufferSize); NORM_API_LINKAGE void NormStreamClose(NormObjectHandle streamHandle, bool graceful = false); @@ -388,7 +408,7 @@ NormAckingStatus NormGetAckingStatus(NormSessionHandle sessionHandle, NORM_API_LINKAGE bool NormStartReceiver(NormSessionHandle sessionHandle, - unsigned long bufferSpace); + UINT32 bufferSpace); NORM_API_LINKAGE void NormStopReceiver(NormSessionHandle sessionHandle); @@ -447,7 +467,7 @@ NORM_API_LINKAGE bool NormStreamSeekMsgStart(NormObjectHandle streamHandle); NORM_API_LINKAGE -unsigned long NormStreamGetReadOffset(NormObjectHandle streamHandle); +UINT32 NormStreamGetReadOffset(NormObjectHandle streamHandle); /** NORM Object Functions */ @@ -459,12 +479,12 @@ NORM_API_LINKAGE bool NormObjectHasInfo(NormObjectHandle objectHandle); NORM_API_LINKAGE -unsigned short NormObjectGetInfoLength(NormObjectHandle objectHandle); +UINT16 NormObjectGetInfoLength(NormObjectHandle objectHandle); NORM_API_LINKAGE -unsigned short NormObjectGetInfo(NormObjectHandle objectHandle, +UINT16 NormObjectGetInfo(NormObjectHandle objectHandle, char* buffer, - unsigned short bufferLen); + UINT16 bufferLen); NORM_API_LINKAGE NormSize NormObjectGetSize(NormObjectHandle objectHandle); @@ -499,9 +519,6 @@ char* NormDataDetachData(NormObjectHandle objectHandle); NORM_API_LINKAGE NormNodeHandle NormObjectGetSender(NormObjectHandle objectHandle); -NORM_API_LINKAGE -char* NormGetData(NormObjectHandle dataHandle); - /** NORM Node Functions */ NORM_API_LINKAGE @@ -511,11 +528,13 @@ NORM_API_LINKAGE bool NormNodeGetAddress(NormNodeHandle nodeHandle, char* addrBuffer, unsigned int* bufferLen, - unsigned short* port = (unsigned short*)0); + UINT16* port = (UINT16*)0); NORM_API_LINKAGE double NormNodeGetGrtt(NormNodeHandle nodeHandle); +NORM_API_LINKAGE +void NormNodeFreeBuffers(NormNodeHandle nodeHandle); // The next 4 functions have not yet been implemented // (work in progress) @@ -542,6 +561,6 @@ void NormNodeRelease(NormNodeHandle nodeHandle); /** Some experimental functions */ NORM_API_LINKAGE -unsigned long NormCountCompletedObjects(NormSessionHandle sessionHandle); +UINT32 NormCountCompletedObjects(NormSessionHandle sessionHandle); #endif // _NORM_API diff --git a/common/normApp.cpp b/common/normApp.cpp index d8ab881..bae6363 100644 --- a/common/normApp.cpp +++ b/common/normApp.cpp @@ -97,7 +97,7 @@ class NormApp : public NormController, public ProtoApp double tx_rate_max; bool cc_enable; - // NormSession server-only parameters + // NormSession sender-only parameters UINT16 segment_size; UINT8 ndata; UINT8 nparity; @@ -111,6 +111,7 @@ class NormApp : public NormController, public ProtoApp bool tx_one_shot; bool tx_ack_shot; bool tx_file_queued; + int tx_robust_factor; // save last obj/block/seg id for later in case needed NormObjectId tx_last_object_id; @@ -124,7 +125,7 @@ class NormApp : public NormController, public ProtoApp char* acking_node_list; // comma-delimited string bool watermark_pending; - // NormSession client-only parameters + // NormSession receiver-only parameters unsigned long rx_buffer_size; // bytes unsigned int rx_sock_buffer_size; NormFileList rx_file_cache; @@ -133,6 +134,8 @@ class NormApp : public NormController, public ProtoApp bool unicast_nacks; bool silent_client; bool low_delay; + int rx_robust_factor; + bool rx_persistent; bool process_aborted_files; // Debug parameters @@ -153,15 +156,14 @@ NormApp::NormApp() address(NULL), port(0), ttl(32), loopback(false), interface_name(NULL), tx_rate(64000.0), tx_rate_min(-1.0), tx_rate_max(-1.0), cc_enable(false), segment_size(1024), ndata(32), nparity(16), auto_parity(0), extra_parity(0), - backoff_factor(NormSession::DEFAULT_BACKOFF_FACTOR), - grtt_estimate(NormSession::DEFAULT_GRTT_ESTIMATE), + backoff_factor(NormSession::DEFAULT_BACKOFF_FACTOR), grtt_estimate(NormSession::DEFAULT_GRTT_ESTIMATE), group_size(NormSession::DEFAULT_GSIZE_ESTIMATE), tx_buffer_size(1024*1024), tx_one_shot(false), tx_ack_shot(false), tx_file_queued(false), - tx_object_interval(0.0), tx_repeat_count(0), tx_repeat_interval(2.0), tx_repeat_clear(true), - acking_node_list(NULL), watermark_pending(false), + tx_robust_factor(NormSession::DEFAULT_ROBUST_FACTOR), tx_object_interval(0.0), tx_repeat_count(0), + tx_repeat_interval(2.0), tx_repeat_clear(true), acking_node_list(NULL), watermark_pending(false), rx_buffer_size(1024*1024), rx_sock_buffer_size(0), rx_cache_path(NULL), post_processor(NULL), unicast_nacks(false), silent_client(false), - low_delay(false), process_aborted_files(false), + low_delay(false), rx_robust_factor(NormSession::DEFAULT_ROBUST_FACTOR), process_aborted_files(false), tracing(false), tx_loss(0.0), rx_loss(0.0) { @@ -834,6 +836,11 @@ bool NormApp::OnCommand(const char* cmd, const char* val) return false; } } + else if (!strncmp("txrobustfactor", cmd, len)) + { + tx_robust_factor = atoi(val); + if (session) session->SetTxRobustFactor(tx_robust_factor); + } else if (!strncmp("rxbuffer", cmd, len)) { if (1 != sscanf(val, "%lu", &rx_buffer_size)) @@ -871,6 +878,15 @@ bool NormApp::OnCommand(const char* cmd, const char* val) low_delay = true; if (session) session->RcvrSetMaxDelay(1); } + else if (!strncmp("rxrobustfactor", cmd, len)) + { + rx_robust_factor = atoi(val); + if (session) session->SetRxRobustFactor(rx_robust_factor); + } + else if (!strncmp("rxpersist", cmd, len)) + { + rx_persistent = true; + } else if (!strncmp("saveAborts", cmd, len)) { process_aborted_files = true; @@ -1290,6 +1306,16 @@ void NormApp::Notify(NormController::Event event, } break; } + + case REMOTE_SENDER_ACTIVE: + DMSG(4, "NormApp::Notify(REMOTE_SENDER_ACTIVE) ...\n"); + break; + + case REMOTE_SENDER_INACTIVE: + DMSG(4, "NormApp::Notify(REMOTE_SENDER_INACTIVE) ...\n"); + if (!rx_persistent) server->FreeBuffers(); + break; + case RX_OBJECT_NEW: { @@ -1777,8 +1803,12 @@ bool NormApp::OnIntervalTimeout(ProtoTimer& /*theTimer*/) return true; } // end NormApp::OnIntervalTimeout() +#include "normSimAgent.h" + bool NormApp::OnStartup(int argc, const char*const* argv) { + TRACE("sizeof(NormSimAgent) = %d\n", sizeof(NormSimAgent)); + if (argc < 3) { ShowHelp(); @@ -1796,6 +1826,8 @@ bool NormApp::OnStartup(int argc, const char*const* argv) DMSG(0, "NormApp::OnStartup() error processing command-line commands\n"); return false; } + + if (control_remote) return false; @@ -1838,6 +1870,7 @@ bool NormApp::OnStartup(int argc, const char*const* argv) session->SetBackoffFactor(backoff_factor); session->ServerSetGrtt(grtt_estimate); session->ServerSetGroupSize(group_size); + session->SetTxRobustFactor(tx_robust_factor); if (!AddAckingNodes(acking_node_list)) { DMSG(0, "NormApp::OnStartup() error: bad acking node list\n"); @@ -1879,7 +1912,8 @@ bool NormApp::OnStartup(int argc, const char*const* argv) if (low_delay) session->RcvrSetMaxDelay(1); else - session->RcvrSetMaxDelay(-1); + session->RcvrSetMaxDelay(-1); + session->SetRxRobustFactor(rx_robust_factor); if (!session->StartClient(rx_buffer_size, interface_name)) { DMSG(0, "NormApp::OnStartup() start client error!\n"); diff --git a/common/normMessage.cpp b/common/normMessage.cpp index 2fb408d..1545d24 100644 --- a/common/normMessage.cpp +++ b/common/normMessage.cpp @@ -419,16 +419,16 @@ NormMsg* NormMessageQueue::RemoveTail() */ // valid for rtt = 1.0e-06 to 1.0e+03 -unsigned char NormQuantizeRtt(double rtt) +UINT8 NormQuantizeRtt(double rtt) { if (rtt > NORM_RTT_MAX) rtt = NORM_RTT_MAX; else if (rtt < NORM_RTT_MIN) rtt = NORM_RTT_MIN; if (rtt < 3.3e-05) - return ((unsigned char)((rtt/NORM_RTT_MIN)) - 1); + return ((UINT8)((rtt/NORM_RTT_MIN)) - 1); else - return ((unsigned char)(ceil(255.0 - (13.0*log(NORM_RTT_MAX/rtt))))); + return ((UINT8)(ceil(255.0 - (13.0*log(NORM_RTT_MAX/rtt))))); } // end NormQuantizeRtt() @@ -443,7 +443,7 @@ unsigned char NormQuantizeRtt(double rtt) // following routine was used to generate // this table: // -// inline double NormUnquantizeRtt(unsigned char qrtt) +// inline double NormUnquantizeRtt(UINT8 qrtt) // { // return ((qrtt < 31) ? // (((double)(qrtt+1))*(double)NORM_RTT_MIN) : @@ -528,7 +528,7 @@ const double NORM_RTT[256] = // The following routine was used to generate // this table // -// inline double NormUnquantizeGroupSize(unsigned char gsize) +// inline double NormUnquantizeGroupSize(UINT8 gsize) // { // double exponent = (double)((gsize & 0x07) + 1); // double mantissa = (0 != (gsize & 0x08)) ? 5.0 : 1.0; @@ -543,6 +543,39 @@ const double NORM_GSIZE[16] = 5.000e+01, 5.000e+02, 5.000e+03, 5.000e+04, 5.000e+05, 5.000e+06, 5.000e+07, 5.000e+08 }; + +// This function rounds up the "gsize" to be conservative +UINT8 NormQuantizeGroupSize(double gsize) +{ + UINT8 exponent = (int)log10(gsize); + if (exponent > 8) + { + return 0x0f; // = 0x08 + 0x07 + } + else if (exponent >= 1) + { + UINT8 mantissa = (int)ceil(gsize / pow(10.0, exponent)); + if (mantissa > 5) + { + if (exponent > 7) + return 0x0f; // = 0x08 + 0x07 + else + return exponent; // = 0x00 + exponent; + } + else if (mantissa > 1) + { + return (exponent + 0x07); // = 0x08 + exponent - 1 + } + else + { + return (exponent - 1); // = 0x00 + exponent - 1; + } + } + else + { + return 0x00; // "gsize" < 10 = 0x00 + 0x00 + } +} // end NormQuantizeGroupSize() diff --git a/common/normMessage.h b/common/normMessage.h index 73de2b3..29f1b31 100644 --- a/common/normMessage.h +++ b/common/normMessage.h @@ -50,51 +50,41 @@ const double NORM_GRTT_MAX = 15.0; // 15 sec const double NORM_RTT_MIN = 1.0e-06; const double NORM_RTT_MAX = 1000.0; extern const double NORM_RTT[]; -inline double NormUnquantizeRtt(unsigned char qrtt) +inline double NormUnquantizeRtt(UINT8 qrtt) { return NORM_RTT[qrtt]; } -unsigned char NormQuantizeRtt(double rtt); +UINT8 NormQuantizeRtt(double rtt); extern const double NORM_GSIZE[]; -inline double NormUnquantizeGroupSize(unsigned char gsize) +inline double NormUnquantizeGroupSize(UINT8 gsize) { return NORM_GSIZE[gsize]; } -inline unsigned char NormQuantizeGroupSize(double gsize) -{ - unsigned char ebits = (unsigned char)log10(gsize); - int mantissa = (int)((gsize/pow(10.0, (double)ebits)) + 0.5); - // round up quantized group size - unsigned char mbit = (mantissa > 1) ? ((mantissa > 5) ? 0x00 : 0x08) : 0x00; - ebits = (mantissa > 5) ? ebits : ebits-1; - mbit = (ebits > 0x07) ? 0x08 : mbit; - ebits = (ebits > 0x07) ? 0x07 : ebits; - return (mbit | ebits); -} +UINT8 NormQuantizeGroupSize(double gsize); -inline unsigned short NormQuantizeLoss(double lossFraction) +inline UINT16 NormQuantizeLoss(double lossFraction) { lossFraction = MAX(lossFraction, 0.0); lossFraction = lossFraction*65535.0 + 0.5; lossFraction = MIN(lossFraction, 65535.0); - return (unsigned short)lossFraction; + return (UINT16)lossFraction; } // end NormQuantizeLossFraction() -inline double NormUnquantizeLoss(unsigned short lossQuantized) +inline double NormUnquantizeLoss(UINT16 lossQuantized) { return (((double)lossQuantized) / 65535.0); } // end NormUnquantizeLossFraction() -inline unsigned short NormQuantizeRate(double rate) +inline UINT16 NormQuantizeRate(double rate) { if (rate <= 0.0) return 0x01; // rate = 0.0 - unsigned short exponent = (unsigned short)log10(rate); - unsigned short mantissa = (unsigned short)((4096.0/10.0) * (rate / pow(10.0, (double)exponent)) + 0.5); + UINT16 exponent = (UINT16)log10(rate); + UINT16 mantissa = (UINT16)((4096.0/10.0) * (rate / pow(10.0, (double)exponent)) + 0.5); return ((mantissa << 4) | exponent); } -inline double NormUnquantizeRate(unsigned short rate) +inline double NormUnquantizeRate(UINT16 rate) { double mantissa = ((double)(rate >> 4)) * (10.0/4096.0); double exponent = (double)(rate & 0x000f); @@ -164,7 +154,7 @@ class NormObjectSize }; // end class NormObjectSize #ifndef _NORM_API -typedef unsigned long NormNodeId; +typedef UINT32 NormNodeId; const NormNodeId NORM_NODE_NONE = 0x00000000; const NormNodeId NORM_NODE_ANY = 0xffffffff; #endif // !_NORM_API diff --git a/common/normNode.cpp b/common/normNode.cpp index 40d52ba..8e02f37 100644 --- a/common/normNode.cpp +++ b/common/normNode.cpp @@ -1864,7 +1864,7 @@ bool NormServerNode::OnActivityTimeout(ProtoTimer& /*theTimer*/) // Serve completely inactive? DMSG(0, "NormServerNode::OnActivityTimeout() node>%lu server>%lu gone inactive?\n", LocalNodeId(), GetId()); - FreeBuffers(); + //FreeBuffers(); This now needs to be done by the app as of norm version 1.4b3 session.Notify(NormController::REMOTE_SENDER_INACTIVE, this, NULL); } else @@ -1915,6 +1915,16 @@ bool NormServerNode::OnActivityTimeout(ProtoTimer& /*theTimer*/) // max_pending_object, 0, 0); } } + // We manually managed the "repeat_count" here to avoid the + // case where "bursty" receiver scheduling may lead to false + // inactivity indication + int repeatCount = activity_timer.GetRepeatCount(); + if (repeatCount > 0) repeatCount--; + activity_timer.Deactivate(); + session.ActivateTimer(activity_timer); + activity_timer.SetRepeatCount(repeatCount); + server_active = false; + return false; // since we manually deactivated/reactivated the timer } server_active = false; return true; diff --git a/common/normNode.h b/common/normNode.h index 2dff03d..6ca5e9e 100644 --- a/common/normNode.h +++ b/common/normNode.h @@ -364,7 +364,7 @@ class NormServerNode : public NormNode unsigned long NackCount() const {return nack_count;} unsigned long SuppressCount() const {return suppress_count;} unsigned long CompletionCount() const {return completion_count;} - unsigned long PendingCount() const {return rx_table.Count();} + unsigned long PendingCount() const {return rx_table.GetCount();} unsigned long FailureCount() const {return failure_count;} diff --git a/common/normObject.cpp b/common/normObject.cpp index 5bd642a..ba970e0 100644 --- a/common/normObject.cpp +++ b/common/normObject.cpp @@ -325,7 +325,11 @@ bool NormObject::TxReset(NormBlockId firstBlock, bool requeue) if (requeue) block->ClearFlag(NormBlock::IN_REPAIR); // since we're requeuing } } - if (requeue) first_pass = true; + if (requeue) + { + first_pass = true; + max_pending_block = 0; + } return increasedRepair; } // end NormObject::TxReset() @@ -376,8 +380,10 @@ bool NormObject::ActivateRepairs() // (TBD) This check can be eventually eliminated if everything else is done right if (!pending_mask.Set(nextId)) { - if (block) block->ClearPending(); DMSG(0, "NormObject::ActivateRepairs() pending_mask.Set(%lu) error 1!\n", (UINT32)nextId); + if (block) block->ClearPending(); + if (IsStream()) + static_cast(this)->UnlockBlock(nextId); } else { @@ -397,11 +403,12 @@ bool NormObject::ActivateRepairs() { DMSG(6, "NormObject::ActivateRepairs() node>%lu obj>%hu activated blk>%lu segment repairs ...\n", LocalNodeId(), (UINT16)transport_id, (UINT32)block->GetId()); - // (TBD) This check can be eventually eliminated if everything else is done right if (!pending_mask.Set(block->GetId())) { - block->ClearPending(); DMSG(0, "NormObject::ActivateRepairs() pending_mask.Set(%lu) error 2!\n", (UINT32)block->GetId()); + block->ClearPending(); + if (IsStream()) + static_cast(this)->UnlockBlock(block->GetId()); } else { @@ -618,6 +625,7 @@ bool NormObject::IsPending(bool flush) const } else { + ASSERT(NULL != server); NormBlockId firstId; if (GetFirstPending(firstId)) { @@ -1590,6 +1598,9 @@ bool NormObject::NextServerMsg(NormObjectMsg* msg) } block->TxInit(blockId, numData, session.ServerAutoParity()); + + if (blockId < max_pending_block) + block->SetFlag(NormBlock::IN_REPAIR); while (!block_buffer.Insert(block)) { ASSERT(STREAM == type); @@ -1610,6 +1621,8 @@ bool NormObject::NextServerMsg(NormObjectMsg* msg) { // Prune old non-pending block (or even pending if "push" enabled stream) block_buffer.Remove(b); + repair_mask.Unset(blockId); // just in case + pending_mask.Unset(blockId); if (IsStream()) static_cast(this)->UnlockBlock(b->GetId()); session.ServerPutFreeBlock(b); @@ -1725,6 +1738,10 @@ bool NormObject::NextServerMsg(NormObjectMsg* msg) // for EMCON sending, mark NORM_INFO for re-transmission, if applicable if (session.SndrEmcon() && HaveInfo()) pending_info = true; + // Advance sender use of "max_pending_block" so we always + // know when a block should be flagged as IN_REPAIR + if (blockId == max_pending_block) + max_pending_block++; } if (!pending_mask.IsSet()) diff --git a/common/normObject.h b/common/normObject.h index 24ff445..4f7cb0d 100644 --- a/common/normObject.h +++ b/common/normObject.h @@ -565,7 +565,7 @@ class NormObjectTable NormObjectId RangeLo() const {return range_lo;} NormObjectId RangeHi() const {return range_hi;} bool IsEmpty() const {return (0 == range);} - UINT32 Count() const {return count;} + UINT32 GetCount() const {return count;} const NormObjectSize& GetSize() const {return size;} class Iterator diff --git a/common/normSession.cpp b/common/normSession.cpp index e95decb..3fac087 100644 --- a/common/normSession.cpp +++ b/common/normSession.cpp @@ -564,7 +564,7 @@ void NormSession::Serve() /* if (command_pending && !command_timer.IsActive()) { - SenderQueueAppCommand() xxx + SenderQueueAppCommand() } */ @@ -738,7 +738,7 @@ void NormSession::Serve() } } } - //ASSERT(stream->IsPending() || stream->IsRepairPending() || stream->IsClosing()); + ASSERT(stream->IsPending() || stream->IsRepairPending() || stream->IsClosing()); if (!posted_tx_queue_empty && !stream->IsClosing() && stream->IsPending()) // post if pending || !repair_timer.IsActive() || (repair_timer.GetRepeatCount() == 0) ??? { @@ -1253,34 +1253,41 @@ bool NormSession::QueueTxObject(NormObject* obj) { if (!IsServer()) { - DMSG(0, "NormSession::QueueTxObject() non-server session error!\n"); + DMSG(0, "NormSession::QueueTxObject() non-sender session error!?\n"); return false; } // Manage tx_table min/max count and max size bounds - if (tx_table.Count() >= tx_cache_count_min) + // Depending on tx cache bounds _and_ what has been + // enqueued/dequeued, we may need to prune the + // "tx_table" a little + // The cases when pruning is needed include: + // + // 1) When the cache bounds dictate: + // i.e., ((count >= count_min) && ((count > count_max) || (size > size_max))), or + // 2) When the "tx_table" state (from insert/remove history) doesn't allow + // i.e., !tx_table.CanInsert(obj) + unsigned long newCount = tx_table.GetCount() + 1; + while (!tx_table.CanInsert(obj->GetId()) || + ((newCount >= tx_cache_count_min) && + ((newCount >= tx_cache_count_max) || + ((tx_table.GetSize() + obj->GetSize()) > tx_cache_size_max)))) { - unsigned long count = tx_table.Count(); - while ((count >= tx_cache_count_min) && - ((count >= tx_cache_count_max) || - ((tx_table.GetSize() + obj->GetSize()) > tx_cache_size_max))) + // Remove oldest non-pending + NormObject* oldest = tx_table.Find(tx_table.RangeLo()); + if (oldest->IsRepairPending() || oldest->IsPending()) { - // Remove oldest non-pending - NormObject* oldest = tx_table.Find(tx_table.RangeLo()); - if (oldest->IsRepairPending() || oldest->IsPending()) - { - DMSG(0, "NormSession::QueueTxObject() all held objects repair pending\n"); - posted_tx_queue_empty = false; - return false; - } - else - { - Notify(NormController::TX_OBJECT_PURGED, (NormServerNode*)NULL, oldest); - DeleteTxObject(oldest); - } - count = tx_table.Count(); - } - } + DMSG(0, "NormSession::QueueTxObject() all held objects repair pending\n"); + posted_tx_queue_empty = false; + return false; + } + else + { + Notify(NormController::TX_OBJECT_PURGED, (NormServerNode*)NULL, oldest); + DeleteTxObject(oldest); + } + newCount = tx_table.GetCount() + 1; + } // Attempt to queue the object (note it gets "retained" by the tx_table) if (!tx_table.Insert(obj)) { @@ -1345,14 +1352,16 @@ bool NormSession::SetTxCacheBounds(NormObjectSize sizeMax, bool result = true; tx_cache_size_max = sizeMax; tx_cache_count_min = (countMin < countMax) ? countMin : countMax; + if (tx_cache_count_min < 1) tx_cache_count_min = 1; tx_cache_count_max = (countMax > countMin) ? countMax : countMin; + if (tx_cache_count_max < 1) tx_cache_count_max = 1; if (IsServer()) { // Trim/resize the tx_table and tx masks as needed - unsigned long count = tx_table.Count(); + unsigned long count = tx_table.GetCount(); while ((count >= tx_cache_count_min) && - ((count >= tx_cache_count_max) || + ((count > tx_cache_count_max) || (tx_table.GetSize() > tx_cache_size_max))) { // Remove oldest (hopefully non-pending ) object @@ -1360,7 +1369,7 @@ bool NormSession::SetTxCacheBounds(NormObjectSize sizeMax, ASSERT(oldest); Notify(NormController::TX_OBJECT_PURGED, (NormServerNode*)NULL, oldest); DeleteTxObject(oldest); - count = tx_table.Count(); + count = tx_table.GetCount(); } if (tx_cache_count_max < DEFAULT_TX_CACHE_MAX) countMax = DEFAULT_TX_CACHE_MAX; diff --git a/common/normTest.cpp b/common/normTest.cpp index b0662a5..97cec6a 100644 --- a/common/normTest.cpp +++ b/common/normTest.cpp @@ -16,7 +16,51 @@ int main(int argc, char* argv[]) { - printf("normTest starting (sizeof(NormSize) = %d)...\n", (int)sizeof(NormSize)); + bool send = false; + bool recv = false; + bool cc = false; + bool trace = false; + int debugLevel = 2; + double loss = 0.0; + + // 1) Parse command-line options + for (int i = 1; i < argc; i++) + { + if (!strcmp(argv[i], "send")) + { + send = true; + } + else if (!strcmp(argv[i], "recv")) + { + recv = true; + } + else if (!strcmp(argv[i], "cc")) + { + cc = true; + } + else if (!strcmp(argv[i], "trace")) + { + trace = true; + } + else if (!strcmp(argv[i], "debug")) + { + debugLevel = atoi(argv[++i]); + } + + else if (!strcmp(argv[i], "loss")) + { + loss = (double)atoi(argv[++i]); + } + } + + if (!send && !recv) + { + TRACE("normTest: neither 'send' or 'recv' operation was specified?!\n"); + TRACE("Usage: normTest {send, recv} [cc]\n"); + return -1; + } + + bool loopback = (send && recv); NormInstanceHandle instance = NormCreateInstance(); @@ -71,11 +115,12 @@ int main(int argc, char* argv[]) // (IMPORTANT NOTE: On Win32 builds with Norm.dll, this "SetDebugLevel()" has no // effect on the Protolib instance in the NORM DLL !!! // (TBD - provide a "NormSetDebugLevel()" function for this purpose!) - SetDebugLevel(2); - // Uncomment to turn on debug NORM message tracing - //NormSetMessageTrace(session, true); + SetDebugLevel(debugLevel); + // Option to turn on debug NORM message tracing + NormSetMessageTrace(session, trace); + // Uncomment to turn on some random packet loss - //NormSetTxLoss(session, 10.0); // 10% packet loss + NormSetTxLoss(session, loss); //NormSetRxLoss(session, 20.0); struct timeval currentTime; ProtoSystemTime(currentTime); @@ -85,14 +130,13 @@ int main(int argc, char* argv[]) NormSetGrttEstimate(session, 0.001); // 1 msec initial grtt - NormSetTransmitRate(session, 1.0e+07); // in bits/second + NormSetTransmitRate(session, 1.0e+04); // in bits/second - // Uncomment to enable TCP-friendly congestion control (overrides NormSetTransmitRate()) - //NormSetCongestionControl(session, true); + // Option to enable TCP-friendly congestion control (overrides NormSetTransmitRate()) + if (cc) NormSetCongestionControl(session, true); + NormSetTransmitRateBounds(session, 5.0e+06, 10.0e+06); - //NormSetTransmitRateBounds(session, 1000.0, -1.0); - - NormSetDefaultRepairBoundary(session, NORM_BOUNDARY_BLOCK); + //NormSetDefaultRepairBoundary(session, NORM_BOUNDARY_BLOCK); // Uncomment to use a _specific_ transmit port number // (Can be the same as session port (rx port), but this @@ -106,25 +150,25 @@ int main(int argc, char* argv[]) // (This allows multiple "normTest" instances on the same machine // for the same NormSession - note those instances must use // unique local NormNodeIds (see NormCreateSession() above). - NormSetRxPortReuse(session, true); + //NormSetRxPortReuse(session, true); // Uncomment to receive your own traffic - NormSetLoopback(session, true); + if (loopback) + NormSetLoopback(session, true); //NormSetSilentReceiver(session, true); // Uncomment this line to participate as a receiver - //NormStartReceiver(session, 1024*1024); - - // Uncomment to set large rx socket buffer size - // (might be needed for high rate sessions) - NormSetRxSocketBuffer(session, 512000); + if (recv) NormStartReceiver(session, 1024*1024); // We use a random "sessionId" NormSessionId sessionId = (NormSessionId)rand(); - // Uncomment the following line to start sender - NormStartSender(session, sessionId, 1024*1024, 1400, 64, 8); + if (send) NormStartSender(session, sessionId, 1024*1024, 1400, 64, 8); + + // Uncomment to set large rx socket buffer size + // (might be needed for high rate sessions) + //NormSetRxSocketBuffer(session, 512000); //NormSetAutoParity(session, 6); @@ -166,25 +210,43 @@ int main(int argc, char* argv[]) int msgCount = 0; int recvCount = -1; // used to monitor reliable stream reception int sendCount = 0; - int sendMax = -1;//300;//-1; // -1 means unlimited + int sendMax = 3000;//-1; // -1 means unlimited int fileMax = 1; NormObjectHandle txFile = NORM_OBJECT_INVALID; NormEvent theEvent; + bool rxActive = false; + while (NormGetNextEvent(instance, &theEvent)) { switch (theEvent.type) { + case NORM_CC_ACTIVE: + DMSG(2, "normTest: NORM_CC_ACTIVE event ...\n"); + if (rxActive) + { + break; + } + else + { + rxActive = true; + // assume there is vacancy + } + case NORM_TX_QUEUE_VACANCY: case NORM_TX_QUEUE_EMPTY: - /*if (NORM_TX_QUEUE_VACANCY == theEvent.type) + if (!rxActive) break; + if (NORM_TX_QUEUE_VACANCY == theEvent.type) TRACE("NORM_TX_QUEUE_VACANCY ...\n"); + else if (NORM_TX_QUEUE_EMPTY == theEvent.type) + TRACE("NORM_TX_QUEUE_EMPTY ...\n"); else - TRACE("NORM_TX_QUEUE_EMPTY ...\n");*/ + TRACE("writing to stream after CC_ACTIVE\n"); if ((sendMax > 0) && (sendCount >= sendMax)) break; - if (NORM_OBJECT_INVALID != theEvent.object) + //if (NORM_OBJECT_INVALID != theEvent.object) + if (true) { // We loop here to keep stream buffer full .... bool keepWriting = true; @@ -212,6 +274,8 @@ int main(int argc, char* argv[]) NormStreamMarkEom(stream); txLen = txIndex = 0; sendCount++; + if (0 == (sendCount % 1000)) + TRACE("normTest: sender sent %d messages ...\n", sendCount); if (sendCount == 15) { //NormSetWatermark(session, stream); @@ -220,13 +284,14 @@ int main(int argc, char* argv[]) { // Uncomment to gracefully shut down the stream // after "sendMax" messages + TRACE("normTest: sender closing stream ...\n"); NormStreamClose(stream, true); keepWriting = false; } } } // end while(keepWriting) } - else if (NORM_OBJECT_INVALID == stream) // we weren't sending a stream that finished + else if (NORM_OBJECT_INVALID == stream) // we aren't sending a stream { if ((fileMax < 0) || (sendCount < fileMax)) { @@ -270,9 +335,24 @@ int main(int argc, char* argv[]) case NORM_TX_OBJECT_PURGED: DMSG(2, "normTest: NORM_TX_OBJECT_PURGED event ...\n"); break; + + case NORM_CC_INACTIVE: + DMSG(2, "normTest: NORM_CC_INACTIVE event ...\n"); + rxActive = false; + // (TBD) add APIs to delete remote sender's state entirely + // (TBD) add APIs to automate buffer freeing, remote sender managment somewhat + break; + + case NORM_REMOTE_SENDER_ACTIVE: + break; + + case NORM_REMOTE_SENDER_INACTIVE: + NormNodeFreeBuffers(theEvent.sender); // frees up some memory allocated for this remote sender's state + break; + case NORM_RX_OBJECT_NEW: - DMSG(3, "normTest: NORM_RX_OBJECT_NEW event ...\n"); + DMSG(2, "normTest: NORM_RX_OBJECT_NEW event ...\n"); break; case NORM_RX_OBJECT_INFO: @@ -347,9 +427,8 @@ int main(int argc, char* argv[]) recvCount = value+1; if (0 == msgCount % 1000) { - TRACE("normTest: status> msgCount:%d of total:%d (%lf)\n", + TRACE("normTest: recv status> msgCount:%d of total:%d (%lf)\n", msgCount, recvCount, 100.0*((double)msgCount)/((double)recvCount)); - } } else @@ -363,7 +442,6 @@ int main(int argc, char* argv[]) TRACE("couldn't find \"hello\"!? len:%d in %s\n", len, rxBuffer); ASSERT(0); } - } else { @@ -396,8 +474,8 @@ int main(int argc, char* argv[]) else { TRACE("normTest: error reading stream\n"); - TRACE("status: msgCount:%d of total:%d (%lf)\n", - msgCount, recvCount, 100.0*((double)msgCount)/((double)recvCount)); + TRACE("normTest: recv status> msgCount:%d of total:%d (%lf)\n", + msgCount, recvCount, 100.0*((double)msgCount)/((double)recvCount)); msgSync = false; rxIndex = 0; break; @@ -408,6 +486,8 @@ int main(int argc, char* argv[]) case NORM_RX_OBJECT_COMPLETED: TRACE("normTest: NORM_RX_OBJECT_COMPLETED event ...\n"); + TRACE("normTest: recv status> msgCount:%d of total:%d (%lf)\n", + msgCount, recvCount, 100.0*((double)msgCount)/((double)recvCount)); //NormStopReceiver(session); break; @@ -424,9 +504,10 @@ int main(int argc, char* argv[]) } // end switch(theEvent.type) // Uncomment to exit program after sending "sendMax" messages or files - if ((sendMax > 0) && (sendCount >= sendMax)) break; + //if ((sendMax > 0) && (sendCount >= sendMax)) break; } // end while (NormGetNextEvent()) + #ifdef WIN32 Sleep(30000); diff --git a/common/normThreadTest.cpp b/common/normThreadTest.cpp index a3ef652..6011d02 100644 --- a/common/normThreadTest.cpp +++ b/common/normThreadTest.cpp @@ -302,6 +302,10 @@ void NormThreadApp::OnNormEvent() case NORM_REMOTE_SENDER_NEW: case NORM_REMOTE_SENDER_ACTIVE: break; + + case NORM_REMOTE_SENDER_INACTIVE: + NormNodeFreeBuffers(theEvent.sender); + break; case NORM_RX_OBJECT_NEW: TRACE("NORM_RX_OBJECT_NEW ...\n"); diff --git a/common/normVersion.h b/common/normVersion.h index b54a3fc..908a3c9 100644 --- a/common/normVersion.h +++ b/common/normVersion.h @@ -32,6 +32,6 @@ #ifndef _NORM_VERSION #define _NORM_VERSION -#define VERSION "1.4b2" +#define VERSION "1.4b3" #endif // _NORM_VERSION diff --git a/npcUsage.pdf b/npcUsage.pdf new file mode 100644 index 0000000..9c84645 Binary files /dev/null and b/npcUsage.pdf differ diff --git a/win32/Norm.opt b/win32/Norm.opt index 0c04171..3ef6944 100644 Binary files a/win32/Norm.opt and b/win32/Norm.opt differ diff --git a/win32/Norm.suo b/win32/Norm.suo index c5294eb..a374c82 100644 Binary files a/win32/Norm.suo and b/win32/Norm.suo differ