From f49de2c510a046781c500f02d4a37a112d1f9b82 Mon Sep 17 00:00:00 2001 From: bebopagogo Date: Mon, 18 May 2020 10:19:28 -0400 Subject: [PATCH] tweaks --- examples/normSocket.cpp | 53 ++++++++++++++++++++++++++++---------- examples/normSocket.h | 3 +++ include/normObject.h | 5 +++- protolib | 2 +- src/common/normApi.cpp | 7 ++--- src/common/normObject.cpp | 16 ++++++++---- src/common/normSession.cpp | 6 +---- 7 files changed, 62 insertions(+), 30 deletions(-) diff --git a/examples/normSocket.cpp b/examples/normSocket.cpp index a3b25d8..314b22a 100644 --- a/examples/normSocket.cpp +++ b/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_stream_buffer_count < tx_stream_buffer_max) + if (!tx_flow_control) { + 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() diff --git a/examples/normSocket.h b/examples/normSocket.h index 20cbfda..e41ba38 100644 --- a/examples/normSocket.h +++ b/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 diff --git a/include/normObject.h b/include/normObject.h index 2446e79..413ec20 100644 --- a/include/normObject.h +++ b/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;} diff --git a/protolib b/protolib index 7cf50bd..0e6731b 160000 --- a/protolib +++ b/protolib @@ -1 +1 @@ -Subproject commit 7cf50bddf3939e5bd78f7264e1baa8b8643d4cd7 +Subproject commit 0e6731b7d61ea78f8ca9bfeb32e73c19bbeb40dd diff --git a/src/common/normApi.cpp b/src/common/normApi.cpp index 66f58fa..b076822 100644 --- a/src/common/normApi.cpp +++ b/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(session->QueueTxStream(bufferSize, true, infoPtr, infoLen)); + static_cast(streamObj); if (obj) objectHandle = (NormObjectHandle)obj; } instance->dispatcher.ResumeThread(); diff --git a/src/common/normObject.cpp b/src/common/normObject.cpp index 38d75bf..eb37f06 100644 --- a/src/common/normObject.cpp +++ b/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", - // (unsigned long)read_index.block.GetValue(), read_index.segment, (unsigned long)read_offset); + 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); read_ready = false; *buflen = bytesRead; if (bytesRead > 0) diff --git a/src/common/normSession.cpp b/src/common/normSession.cpp index 37a0c8d..1e83789 100644 --- a/src/common/normSession.cpp +++ b/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;