pynorm improvement to keep cache sender enqueued objects in pynorm.instance._objects cache

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 $<BUILD_INTERFACE:${CMAKE_CURRENT_SOURCE_DIR}/include> $<INSTALL_INTERFACE:include>)
 
 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()

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

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;

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

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)

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()
 

src/common/normObject.cpp

@@ -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)

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,
+NormDataObject *NormSession::QueueTxData(const char* dataPtr,
-                                         UINT32 dataLen,
+                                         UINT32      dataLen,
-                                         const char *infoPtr,
+                                         const char* infoPtr,
-                                         UINT16 infoLen)
+                                         UINT16      infoLen)
 {
     if (!IsSender())
     {
@@ -1952,10 +1952,10 @@ NormDataObject *NormSession::QueueTxData(const char *dataPtr,
     }
 } // end NormSession::QueueTxData()
 
-NormStreamObject *NormSession::QueueTxStream(UINT32 bufferSize,
+NormStreamObject *NormSession::QueueTxStream(UINT32      bufferSize,
-                                             bool doubleBuffer,
+                                             bool        doubleBuffer,
-                                             const char *infoPtr,
+                                             const char* infoPtr,
-                                             UINT16 infoLen)
+                                             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())
+        if (notify)
-            static_cast<NormFileObject*>(obj)->CloseFile();
+        {
-        Notify(NormController::TX_OBJECT_PURGED, (NormSenderNode *)NULL, obj);
+            if (NormObject::FILE == obj->GetType())
+                static_cast<NormFileObject*>(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 &currentTime,
                 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 &currentTime,
             {
                 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
+    // Adjust send time for any current hold time since it 
-    // since it will be "rehandled"
+    // 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)

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):

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):

src/pynorm/session.py

@@ -122,17 +122,41 @@ 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)))
+        # 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=""):
+        # 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 streamOpen(self, bufferSize, info=""):
-        return Object(libnorm.NormStreamOpen(self, bufferSize, info.encode('utf-8'), len(info)))
-
     def sendCommand(self, cmdBuffer, robust=False):
         return libnorm.NormSendCommand(self, cmdBuffer.encode('utf-8'), len(cmdBuffer), robust)