tweaks

examples/normSocket.cpp

@@ -207,12 +207,22 @@ class NormSocket
         NormSessionHandle GetMulticastSession() const
             {return mcast_session;}
         
+        NormObjectHandle GetTxStream() const
+            {return tx_stream;}
     
         void InitRxStream(NormObjectHandle rxStream)
             {rx_stream = rxStream;}
         NormObjectHandle GetRxStream() const
             {return rx_stream;}
         
+        
+        void SetFlowControl(bool state)
+        {
+            tx_flow_control = state;
+            if (NORM_OBJECT_INVALID != tx_stream)
+                NormStreamSetPushEnable(tx_stream, state ? false : true);
+        }
+        
         void InitTxStream(NormObjectHandle txStream, unsigned int bufferSize, UINT16 segmentSize, UINT16 blockSize)
         {
             tx_stream = txStream;
@@ -223,6 +233,7 @@ class NormSocket
             tx_stream_bytes_remain = 0;
             tx_watermark_pending = false;
             tx_ready = true;
+            NormStreamSetPushEnable(tx_stream, tx_flow_control ? false : true);
         }
         
         bool Open(NormInstanceHandle instance = NORM_INSTANCE_INVALID);
@@ -244,13 +255,9 @@ class NormSocket
         // hard, immediate closure
         void Close();
         
-        
-        
-        
         void SetSocketInfo(NormSocketInfo* socketInfo)        // for server-side, client sockets only
             {socket_info = socketInfo;}
         
-        
         NormSocketInfo* FindSocketInfo(NormNodeHandle client)
             {return client_table.FindSocketInfo(client);}
         void RemoveSocketInfo(NormSocketInfo& socketInfo)     // for server sockets only
@@ -350,6 +357,7 @@ class NormSocket
         unsigned int        tx_stream_buffer_count;
         unsigned int        tx_stream_bytes_remain;
         bool                tx_watermark_pending;
+        bool                tx_flow_control;
         // Receive stream state
         NormObjectHandle    rx_stream;
         const void*         user_data;      // for use by user application
@@ -366,6 +374,7 @@ NormSocket::NormSocket(NormSessionHandle normSession)
    tx_stream(NORM_OBJECT_INVALID), tx_ready(false), tx_segment_size(0), 
    tx_stream_buffer_max(0), tx_stream_buffer_count(0),
    tx_stream_bytes_remain(0), tx_watermark_pending(false),
+   tx_flow_control(true),
    rx_stream(NORM_OBJECT_INVALID), user_data(NULL)
 {
     // For now we use the NormSession "user data" option to associate
@@ -732,7 +741,6 @@ bool NormSocket::Connect(const char*    serverAddr,
     else
     {
         // Set timeout for connect attempt (for "heavyweight" mcast connect, this would also be done)
-        TRACE("NormSetUserTimer(1) session: %p\n", norm_session);
         NormSetUserTimer(norm_session, NORM_DEFAULT_CONNECT_TIMEOUT);   
     }
     server_socket = NULL;  // this is a client-side socket
@@ -769,9 +777,14 @@ unsigned int NormSocket::Write(const char* buffer, unsigned int numBytes)
         InitTxStream(tx_stream, 2*1024*1024, 1400, 16);
     }
     
-    // This method uses NormStreamWrite(), but limits writes by explicit ACK-based flow control status
+    if (!tx_flow_control)
-    if (tx_stream_buffer_count < tx_stream_buffer_max)
     {
+        unsigned int bytesWritten = NormStreamWrite(tx_stream, buffer, numBytes);
+        return bytesWritten;
+    }
+    else if (tx_stream_buffer_count < tx_stream_buffer_max)
+    {
+        // This method uses NormStreamWrite(), but limits writes by explicit ACK-based flow control status
         // 1) How many buffer bytes are available?
         unsigned int bytesAvailable = tx_segment_size * (tx_stream_buffer_max - tx_stream_buffer_count);
         bytesAvailable -= tx_stream_bytes_remain;  // unflushed segment portiomn
@@ -878,7 +891,6 @@ void NormSocket::Shutdown()
         if ((IsServerSide() && IsMulticastClient()) || (NORM_OBJECT_INVALID == tx_stream))
         {
             // Use a zero-timeout to immediately post NORM_SOCKET_CLOSE notification
-            TRACE("NormSetUserTimer(2) session: %p\n", norm_session);
             NormSetUserTimer(norm_session, 0.0);
         }
         else if (NORM_OBJECT_INVALID != tx_stream)
@@ -892,7 +904,6 @@ void NormSocket::Shutdown()
     else
     {
         // Use a zero-timeout to immediately post NORM_SOCKET_CLOSE notification
-        TRACE("NormSetUserTimer(6) session: %p\n", norm_session);
         NormSetUserTimer(norm_session, 0.0);
     }
 }  // end NormSocket::Shutdown()
@@ -914,7 +925,6 @@ void NormSocket::Close()
                     NormNodeHandle node = NormGetAckingNodeHandle(mcast_session, nodeId);
                     assert(NORM_NODE_INVALID != node);
                     NormSocket* clientSocket = (NormSocket*)NormNodeGetUserData(node);
-                    TRACE("NormSetUserTimer(3) session: %p\n", clientSocket->norm_session);
                     NormSetUserTimer(clientSocket->norm_session, 0.0);
                 }
                 // for mcast server mcast_session == norm_session so it's destroyed below
@@ -1068,7 +1078,6 @@ void NormSocket::GetSocketEvent(const NormEvent& event, NormSocketEvent& socketE
                                     // We use the session timer to dispatch a NORM_SOCKET_CLOSE per failed client
                                     // (This will also remove the client from this server's acking list)
                                     clientSocket->socket_state = CLOSING;
-                                    TRACE("NormSetUserTimer(4) session: %p\n", clientSocket->norm_session);
                                     NormSetUserTimer(clientSocket->norm_session, 0.0);
                                 }
                             }
@@ -1134,7 +1143,6 @@ void NormSocket::GetSocketEvent(const NormEvent& event, NormSocketEvent& socketE
                 case CONNECTING:
                     // TBD - We should validate that it's the right remote sender
                     //       (i.e., by source address and/or nodeId)
-                    TRACE("NormCancelUserTimer(1) norm_session: %p\n", norm_session);
                     NormCancelUserTimer(norm_session);
                     socketEvent.type = NORM_SOCKET_CONNECT;
                     NormSetSynStatus(norm_session, false);
@@ -1628,6 +1636,19 @@ NormSessionHandle NormGetSocketSession(NormSocketHandle normSocket)
     return s->GetSession();
 }  // end NormGetSocketSession()
 
+NormObjectHandle NormGetSocketTxStream(NormSocketHandle normSocket)
+{
+    NormSocket* s = (NormSocket*)normSocket;
+    return s->GetTxStream();
+}  // end NormGetSocketTxStream()
+
+NormObjectHandle NormGetSocketRxStream(NormSocketHandle normSocket)
+{
+    NormSocket* s = (NormSocket*)normSocket;
+    return s->GetRxStream();
+}  // end NormGetSocketRxStream()
+
+
 NormSessionHandle NormGetSocketMulticastSession(NormSocketHandle normSocket)
 {
     NormSocket* s = (NormSocket*)normSocket;
@@ -1639,4 +1660,10 @@ void NormSetSocketTrace(NormSocketHandle normSocket, bool enable)
 {
     NormSocket* s = (NormSocket*)normSocket;
     s->SetTrace(enable);
-}
+}  // end NormSetSocketTrace()
+
+void NormSetSocketFlowControl(NormSocketHandle normSocket, bool enable)
+{
+    NormSocket* s = (NormSocket*)normSocket;
+    s->SetFlowControl(enable);
+}  // end NormSetSocketFlowControl()

examples/normSocket.h

@@ -101,7 +101,10 @@ NormInstanceHandle NormGetSocketInstance(NormSocketHandle normSocket);
 NormSessionHandle NormGetSocketSession(NormSocketHandle normSocket);
 NormSessionHandle NormGetSocketMulticastSession(NormSocketHandle normSocket);
 void NormGetPeerName(NormSocketHandle normSocket, char* addr, unsigned int* addrLen, UINT16* port);
+NormObjectHandle NormGetSocketTxStream(NormSocketHandle normSocket);
+NormObjectHandle NormGetSocketRxStream(NormSocketHandle normSocket);
 
+void NormSetSocketFlowControl(NormSocketHandle normSocket, bool enable);
 void NormSetSocketTrace(NormSocketHandle normSocket, bool enable);
 
 typedef enum NormSocketEventType

include/normObject.h

@@ -52,7 +52,10 @@ class NormObject
         
         // This must be reset after each update
         void SetNotifyOnUpdate(bool state)
-            {notify_on_update = state;}
+        {
+                TRACE("enter SetNotifyOnUpdate(%d) ...\n", state);
+                notify_on_update = state;
+        }
         
         // Object information
         NormObject::Type GetType() const {return type;}

protolib

@@ -1 +1 @@
-Subproject commit 7cf50bddf3939e5bd78f7264e1baa8b8643d4cd7
+Subproject commit 0e6731b7d61ea78f8ca9bfeb32e73c19bbeb40dd

src/common/normApi.cpp

@@ -220,11 +220,8 @@ void NormInstance::Notify(NormController::Event   event,
     {
         case SEND_OK:
             // Purge any pending NORM_SEND_ERROR notifications for session
-            TRACE("purging SEND_ERROR ...\n");
             PurgeNotifications(session, NORM_SEND_ERROR);
             return;
-        case SEND_ERROR:
-            TRACE("got SEND_ERROR\n");
         default:
             break;
     }
@@ -557,7 +554,6 @@ bool NormInstance::GetNextEvent(NormEvent* theEvent)
             case NORM_SEND_ERROR:
             {
                 NormSession* session = (NormSession*)next->event.session;
-                TRACE("calling ClearSendError() ....\n");
                 session->ClearSendError();
                 break;
             }
@@ -1878,8 +1874,9 @@ NormObjectHandle NormStreamOpen(NormSessionHandle  sessionHandle,
         NormSession* session = (NormSession*)sessionHandle;
         if (session)
         {
+            NormStreamObject* streamObj = session->QueueTxStream(bufferSize, true, infoPtr, infoLen);
             NormObject* obj = 
-                static_cast<NormObject*>(session->QueueTxStream(bufferSize, true, infoPtr, infoLen));
+                static_cast<NormObject*>(streamObj);
             if (obj) objectHandle = (NormObjectHandle)obj;
         }
         instance->dispatcher.ResumeThread();

src/common/normObject.cpp

@@ -3521,14 +3521,20 @@ bool NormStreamObject::Read(char* buffer, unsigned int* buflen, bool seekMsgStar
         bytesWanted = *buflen;
     }
     bool result = ReadPrivate(buffer, buflen, seekMsgStart);
-    if (!read_ready) notify_on_update = true;
+    if (!read_ready) 
+    {
+        notify_on_update = true;
+    }
     if (!seekMsgStart && result && (0 != *buflen) && (*buflen < bytesWanted))
     {
         char dummyBuffer[8];
         unsigned int dummyCount = 8;
         stream_broken = ReadPrivate(dummyBuffer, &dummyCount, false) ? false : true;
         ASSERT(0 == dummyCount);
-        if (!read_ready) notify_on_update = true;
+        if (!read_ready) 
+        {
+            notify_on_update = true;
+        }
     }
     return result;
 }  // end NormStreamObject::Read()
@@ -3553,7 +3559,7 @@ bool NormStreamObject::ReadPrivate(char* buffer, unsigned int* buflen, bool seek
         NormBlock* block = stream_buffer.Find(read_index.block);
         if (NULL == block)
         {
-            //DMSG(0, "NormStreamObject::ReadPrivate() stream buffer empty (1) (sbEmpty:%d)\n", stream_buffer.IsEmpty());
+            PLOG(PL_DETAIL, "NormStreamObject::ReadPrivate() stream buffer empty (1) (sbEmpty:%d)\n", stream_buffer.IsEmpty());
             read_ready = false;
             *buflen = bytesRead;
             if (bytesRead > 0)
@@ -3614,8 +3620,8 @@ bool NormStreamObject::ReadPrivate(char* buffer, unsigned int* buflen, bool seek
         
         if (NULL == segment)
         {
-            //DMSG(0, "NormStreamObject::ReadPrivate(%lu:%hu) stream buffer empty (read_offset>%lu) (2)\n",
+            PLOG(PL_DETAIL, "NormStreamObject::ReadPrivate(%lu:%hu) stream buffer empty (read_offset>%lu) (2)\n",
-            //        (unsigned long)read_index.block.GetValue(), read_index.segment, (unsigned long)read_offset);
+                    (unsigned long)read_index.block.GetValue(), read_index.segment, (unsigned long)read_offset);
             read_ready = false;
             *buflen = bytesRead;
             if (bytesRead > 0)

src/common/normSession.cpp

@@ -254,8 +254,6 @@ bool NormSession::Open()
     if (!tx_socket->SetFragmentation(fragmentation))
         PLOG(PL_WARN, "NormSession::Open() warning: tx_socket.SetFragmentation() error\n");
     
-    
-    TRACE("NormSession::Open() address %s\n", address.GetHostString());
     if (address.IsMulticast())
     {
         if (!tx_socket->SetTTL(ttl))
@@ -2240,7 +2238,6 @@ void NormSession::TxSocketRecvHandler(ProtoSocket&       theSocket,
             }
             else
             {
-                TRACE("NormSession::TxSocketRecvHandler() RecvFrom error\n");
                 // Probably an ICMP "port unreachable" error
                 // Note we purposefull do _not_ set the "posted_send_error"
                 // status here because we do not want this notification
@@ -2362,14 +2359,13 @@ void NormSession::RxSocketRecvHandler(ProtoSocket&       theSocket,
             }
             else
             {   
-                TRACE("NormSession::RxSocketRecvHandler() RecvFrom error  address:%s (unicast:%d)\n", Address().GetHostString(), Address().IsUnicast());
+                //TRACE("NormSession::RxSocketRecvHandler() RecvFrom error  address:%s (unicast:%d)\n", Address().GetHostString(), Address().IsUnicast());
                 // Probably an ICMP "port unreachable" error
                 // Note we purposefull do _not_ set the "posted_send_error"
                 // status here because we do not want this notification
                 // cleared due to SEND_OK status since it's receiver driven
 				if (Address().IsUnicast())
                 {
-                    TRACE("posting send error ...\n");
                     Notify(NormController::SEND_ERROR, NULL, NULL);
                 }
                 break;