diff --git a/CMakeLists.txt b/CMakeLists.txt index d646cfb..250e85c 100755 --- a/CMakeLists.txt +++ b/CMakeLists.txt @@ -1,7 +1,7 @@ cmake_minimum_required(VERSION 3.14) cmake_policy(SET CMP0077 NEW) # set the project name -project(norm VERSION 1.5.8) +project(norm VERSION 1.5.10) set(CMAKE_C_FLAGS "${CMAKE_C_FLAGS} -fvisibility=hidden") set(CMAKE_CXX_FLAGS "${CMAKE_CXX_FLAGS} -fvisibility=hidden") @@ -126,7 +126,7 @@ target_include_directories(norm PRIVATE ${CMAKE_CURRENT_BINARY_DIR}) target_include_directories(norm PUBLIC $ $) if(BUILD_SHARED_LIBS) - set_target_properties(norm PROPERTIES SOVERSION 1) + set_target_properties(norm PROPERTIES SOVERSION 1.5.10) if(WIN32) target_compile_definitions(norm PUBLIC NORM_USE_DLL) endif() diff --git a/examples/normSocket.cpp b/examples/normSocket.cpp index 5d57b8c..829ed08 100755 --- a/examples/normSocket.cpp +++ b/examples/normSocket.cpp @@ -519,11 +519,13 @@ bool NormSocket::Listen(UINT16 serverPort, const char* groupAddr, const char* se NormSetSilentReceiver(norm_session, true); // So that the listener can construct (unsent) ACKs without failure + // (I.e., these ACKs are never sent by the listener but this call is + // needed as a "work around" existing NORM "connectionless" behavior) NormSetDefaultUnicastNack(norm_session, true); - // Note we use a small buffer size here since a "listening" socket isn't + // Note we use a _small_ buffer size here since a "listening" socket isn't // going to be receiving data (TBD - implement a mechanism to handoff remote - // sender (i.e. "client") from parent + // sender (i.e. "client") from parent directly?) if (!NormStartReceiver(norm_session, 2048)) { fprintf(stderr, "NormSocket::Listen() error: NormStartReceiver() failure (perhaps port already in use)\n"); @@ -580,7 +582,7 @@ NormSocket* NormSocket::Accept(NormNodeHandle client, NormInstanceHandle instanc // However, note that even though we've "connected" this sender, // there is a chance that additional packets in the "serverSession" // rx socket buffer may look like a new sender if deleted now, so - // we wait for NORM_REMOTE_SENDER_INACTIVE to delete + // we wait for NORM_REMOTE_SENDER_INACTIVE to delete the remote sender from the listener sesssion #ifndef WIN32 // Enable rx port reuse since it's the server port, and connect diff --git a/include/normApi.h b/include/normApi.h index 10ba966..0caa5c3 100755 --- a/include/normApi.h +++ b/include/normApi.h @@ -64,7 +64,7 @@ extern "C" { #define NORM_VERSION_MAJOR 1 #define NORM_VERSION_MINOR 5 -#define NORM_VERSION_PATCH 7 +#define NORM_VERSION_PATCH 10 /** NORM API Types */ typedef const void* NormInstanceHandle; diff --git a/include/normMessage.h b/include/normMessage.h index 1cc70b6..5aee81a 100755 --- a/include/normMessage.h +++ b/include/normMessage.h @@ -195,13 +195,11 @@ class NormObjectId UINT16 diff = value - id.value; return ((diff > 0x8000) || ((0x8000 == diff) && (value > id.value))); } - bool operator>(const NormObjectId& id) const { UINT16 diff = id.value - value; return ((diff > 0x8000) || ((0x8000 == diff) && (id.value > value))); } - bool operator<=(const NormObjectId& id) const {return ((value == id.value) || (*this < id));} bool operator>=(const NormObjectId& id) const diff --git a/src/common/normApi.cpp b/src/common/normApi.cpp index ef172b7..9764a3f 100755 --- a/src/common/normApi.cpp +++ b/src/common/normApi.cpp @@ -21,7 +21,6 @@ extern NORM_API_LINKAGE const NormNodeId NORM_NODE_ANY = ((NormNodeId)0xffffffff); extern NORM_API_LINKAGE const NormObjectHandle NORM_OBJECT_INVALID = ((NormObjectHandle)0); - extern NORM_API_LINKAGE const NormDescriptor NORM_DESCRIPTOR_INVALID = ProtoDispatcher::INVALID_DESCRIPTOR; @@ -763,6 +762,7 @@ UINT32 NormInstance::CountCompletedObjects(NormSession* session) NORM_API_LINKAGE int NormGetVersion(int* major, int* minor, int* patch) { + // TBD - use normVersion.h info instead if (NULL != major) *major = NORM_VERSION_MAJOR; if (NULL != minor) *minor = NORM_VERSION_MINOR; if (NULL != patch) *patch = NORM_VERSION_PATCH; @@ -1257,6 +1257,10 @@ void NormSetServerListener(NormSessionHandle sessionHandle, bool state) NORM_API_LINKAGE bool NormTransferSender(NormSessionHandle sessionHandle, NormNodeHandle senderHandle) { + // The NormSession::InsertRemoteSender() used below synthesizes a NORM_CMD(CC) that + // is injected into the "sessionHandle" NormSession which results in a new remote + // sender being established. I.e., it basically "clones" the sender from another + // NormSession into the target NormSession bool result = false; NormInstance* dstInstance = NormInstance::GetInstanceFromSession(sessionHandle); if (NULL != dstInstance) diff --git a/src/common/normNode.cpp b/src/common/normNode.cpp index 0c24295..c176143 100755 --- a/src/common/normNode.cpp +++ b/src/common/normNode.cpp @@ -1190,6 +1190,8 @@ void NormSenderNode::HandleRepairContent(const UINT32* buffer, UINT16 bufferLen) void NormSenderNode::CalculateGrttResponse(const struct timeval& currentTime, struct timeval& grttResponse) const { + // The returned "grttResponse" is the remote sender's cached "grtt_send_time" with + // any processing delay (currentTime - grtt_recv_time) removed. grttResponse.tv_sec = grttResponse.tv_usec = 0; if (grtt_send_time.tv_sec || grtt_send_time.tv_usec) { @@ -3357,7 +3359,7 @@ void NormNodeTree::Destroy() while ((n = root)) { DetachNode(n); - n->Release(); + n->Release(); // note will delete the node if no other reference } } // end NormNodeTree::Destroy() diff --git a/src/common/normObject.cpp b/src/common/normObject.cpp index 600d168..6aa30f0 100755 --- a/src/common/normObject.cpp +++ b/src/common/normObject.cpp @@ -59,7 +59,7 @@ void NormObject::Release() { PLOG(PL_ERROR, "NormObject::Release() releasing non-retained object?!\n"); } - if (0 == reference_count) delete this; + if (0 == reference_count) delete this; } // end NormObject::Release() // This is mainly used for debug messages @@ -4393,7 +4393,8 @@ void NormObjectTable::Destroy() count = range = range_max = 0; } // end NormObjectTable::Destroy() -#else +#else // !USE_PROTO_TREE + NormObject* NormObjectTable::Find(const NormObjectId& objectId) const { if (0 != range) diff --git a/src/common/normSession.cpp b/src/common/normSession.cpp index 8290d31..d02008e 100755 --- a/src/common/normSession.cpp +++ b/src/common/normSession.cpp @@ -1008,7 +1008,7 @@ void NormSession::StopSender() obj->Release(); } // Then destroy table - tx_table.Destroy(); + tx_table.Destroy(); // I think this is redundant with above tx_table iteration block_pool.Destroy(); segment_pool.Destroy(); tx_repair_mask.Destroy(); @@ -1892,7 +1892,7 @@ NormFileObject *NormSession::QueueTxFile(const char *path, PLOG(PL_FATAL, "NormSession::QueueTxFile() Error: sender is closed\n"); return NULL; } - NormFileObject *file = new NormFileObject(*this, (NormSenderNode *)NULL, next_tx_object_id); + NormFileObject* file = new NormFileObject(*this, (NormSenderNode*)NULL, next_tx_object_id); if (NULL == file) { PLOG(PL_FATAL, "NormSession::QueueTxFile() new file object error: %s\n", @@ -1917,10 +1917,10 @@ NormFileObject *NormSession::QueueTxFile(const char *path, } } // end NormSession::QueueTxFile() -NormDataObject *NormSession::QueueTxData(const char *dataPtr, - UINT32 dataLen, - const char *infoPtr, - UINT16 infoLen) +NormDataObject *NormSession::QueueTxData(const char* dataPtr, + UINT32 dataLen, + const char* infoPtr, + UINT16 infoLen) { if (!IsSender()) { @@ -1952,10 +1952,10 @@ NormDataObject *NormSession::QueueTxData(const char *dataPtr, } } // end NormSession::QueueTxData() -NormStreamObject *NormSession::QueueTxStream(UINT32 bufferSize, - bool doubleBuffer, - const char *infoPtr, - UINT16 infoLen) +NormStreamObject *NormSession::QueueTxStream(UINT32 bufferSize, + bool doubleBuffer, + const char* infoPtr, + UINT16 infoLen) { if (!IsSender()) { @@ -2132,9 +2132,12 @@ void NormSession::DeleteTxObject(NormObject *obj, bool notify) ASSERT(NULL != obj); if (tx_table.Remove(obj)) { - if (NormObject::FILE == obj->GetType()) - static_cast(obj)->CloseFile(); - Notify(NormController::TX_OBJECT_PURGED, (NormSenderNode *)NULL, obj); + if (notify) + { + if (NormObject::FILE == obj->GetType()) + static_cast(obj)->CloseFile(); + Notify(NormController::TX_OBJECT_PURGED, (NormSenderNode*)NULL, obj); + } NormObjectId objectId = obj->GetId(); tx_pending_mask.Unset(objectId); tx_repair_mask.Unset(objectId); @@ -3021,6 +3024,7 @@ void NormSession::ReceiverHandleObjectMessage(const struct timeval ¤tTime, PLOG(PL_FATAL, "NormSession::ReceiverHandleObjectMessage() node>%lu error opening NormSenderNode\n", (unsigned long)LocalNodeId()); // (TBD) notify application of error + theSender->Release(); // will delete theSender if no other reference return; } Notify(NormController::REMOTE_SENDER_NEW, theSender, NULL); @@ -3112,6 +3116,7 @@ void NormSession::ReceiverHandleCommand(const struct timeval ¤tTime, { PLOG(PL_ERROR, "NormSession::ReceiverHandleCommand() node>%lu error opening NormSenderNode\n"); // (TBD) notify application of error + theSender->Release(); // will delete the sender if no other reference return; } Notify(NormController::REMOTE_SENDER_NEW, theSender, NULL); @@ -3154,14 +3159,18 @@ bool NormSession::InsertRemoteSender(NormSenderNode &sender) cmd.Init(); cmd.SetSequence(sender.GetCurrentSequence()); cmd.SetSourceId(sender.GetId()); - cmd.SetDestination(sender.GetAddress()); + cmd.SetSource(sender.GetAddress()); cmd.SetInstanceId(sender.GetInstanceId()); cmd.SetGrtt(sender.GetGrttQuantized()); cmd.SetBackoffFactor((UINT8)sender.GetBackoffFactor()); // TBD - should we clean up this backoff factor casting cmd.SetGroupSize(sender.GetGroupSizeQuantized()); cmd.SetCCSequence(sender.GetCCSequence()); - // Adjust send time for any current hold time - // since it will be "rehandled" + // Adjust send time for any current hold time since it + // will be "rehandled" with a new apparent recv time. + // This makes sure a proper grtt response time is calculated + // in this target session accounting from any hold time accrued + // from when the original sender NORM_CMD(CC) was received by + // the other session the sender "state" is being transferred from. struct timeval adjustedSendTime; struct timeval currentTime; ::ProtoSystemTime(currentTime); @@ -5165,6 +5174,8 @@ void NormSession::SetGrttProbingInterval(double intervalMin, double intervalMax) intervalMin = intervalMax; intervalMax = temp; } + // TBD - If NORM_TICK_MIN is too big, does this limit rate of + // of slow start ramp up? if (intervalMin < NORM_TICK_MIN) intervalMin = NORM_TICK_MIN; if (intervalMax < NORM_TICK_MIN) diff --git a/src/pynorm/node.py b/src/pynorm/node.py index 24c7d9c..d5f165e 100755 --- a/src/pynorm/node.py +++ b/src/pynorm/node.py @@ -62,7 +62,7 @@ class Node(object): ## Private Functions def __del__(self): - libnorm.NormNodeRelease(self) + libnorm.NormNodeRelease(self._node) @property def _as_parameter_(self): diff --git a/src/pynorm/object.py b/src/pynorm/object.py index deb36c1..b3673a9 100755 --- a/src/pynorm/object.py +++ b/src/pynorm/object.py @@ -28,11 +28,9 @@ class Object(object): def getInfo(self): if not self.hasObjectInfo(): raise NormError("No object info received yet.") - length = libnorm.NormObjectGetInfoLength(self) if length == 0: raise NormError("No object info received yet.") - buf = ctypes.create_string_buffer(length) recv = libnorm.NormObjectGetInfo(self, buf, length) if recv == 0: @@ -49,7 +47,8 @@ class Object(object): libnorm.NormObjectCancel(self) def getFileName(self): - buf = ctypes.create_string_buffer(100) + # TBD - should we do something to get file name size first? + buf = ctypes.create_string_buffer(256) libnorm.NormFileGetName(self, buf, ctypes.sizeof(buf)) return buf.value @@ -120,7 +119,7 @@ class Object(object): ## Private functions def __del__(self): - libnorm.NormObjectRelease(self) + libnorm.NormObjectRelease(self._object) @property def _as_parameter_(self): diff --git a/src/pynorm/session.py b/src/pynorm/session.py index 02dee14..c2f32d8 100755 --- a/src/pynorm/session.py +++ b/src/pynorm/session.py @@ -122,16 +122,40 @@ class Session(object): libnorm.NormSetGroupSize(self, size) def fileEnqueue(self, filename, info=""): - return Object(libnorm.NormFileEnqueue(self, filename.encode('utf-8'), info.encode('utf-8'), len(info))) + # TBD - allow for case of info being None? + result = libnorm.NormFileEnqueue(self, filename.encode('utf-8'), info.encode('utf-8'), len(info)) + if ctypes.c_void_p.in_dll(libnorm, "NORM_OBJECT_INVALID") == result: + return None; # enqueue not successful due to flow control or sender cache limit + else + # Put a reference of the object in our instance "_objects" cache to avoid creation + # of duplicative Python NORM Object during event notification + obj = self._instance._objects[self._estruct.object] = Object(result) + return obj def dataEnqueue(self, data, info=""): - return Object(libnorm.NormDataEnqueue(self, data.encode('utf-8'), len(data), info.encode('utf-8'), len(info))) - - def requeueObject(self, normObject): - libnorm.NormRequeueObject(self, normObject) + # TBD - allow for case of info being None? + result = libnorm.NormDataEnqueue(self, data.encode('utf-8'), len(data), info.encode('utf-8'), len(info)) + if ctypes.c_void_p.in_dll(libnorm, "NORM_OBJECT_INVALID") == result: + return None; # enqueue not successful due to flow control or sender cache limit + else + # Put a reference of the object in our instance "_objects" cache to avoid creation + # of duplicative Python NORM Object during event notification + obj = self._instance._objects[self._estruct.object] = Object(result) + return obj def streamOpen(self, bufferSize, info=""): - return Object(libnorm.NormStreamOpen(self, bufferSize, info.encode('utf-8'), len(info))) + # TBD - allow for case of info being None? + result = libnorm.NormStreamOpen(self, bufferSize, info.encode('utf-8'), len(info)) + if ctypes.c_void_p.in_dll(libnorm, "NORM_OBJECT_INVALID") == result: + return None; # stream open/enqueue not successful due to flow control or sender cache limit + else + # Put a reference of the object in our instance "_objects" cache to avoid creation + # of duplicative Python NORM Object during event notification + obj = self._instance._objects[self._estruct.object] = Object(result) + return obj + + def requeueObject(self, normObject): + libnorm.NormRequeueObject(self, normObject) def sendCommand(self, cmdBuffer, robust=False): return libnorm.NormSendCommand(self, cmdBuffer.encode('utf-8'), len(cmdBuffer), robust)