From 88e600cedb89efe684a87bbbf7c518c1a494c0dc Mon Sep 17 00:00:00 2001 From: bebopagogo Date: Wed, 29 Apr 2020 22:38:59 -0400 Subject: [PATCH] added NormStreamGetVacancy() and some NormSocket improvement --- examples/normServer.cpp | 7 +- examples/normSocket.cpp | 284 ++++++++++++++++++++++++++++++++- examples/normSocket.h | 3 + include/normApi.h | 6 + include/normNode.h | 2 +- include/normObject.h | 4 + include/normSession.h | 8 +- makefiles/Makefile.common | 23 ++- src/common/normApi.cpp | 88 +++++++++- src/common/normEncoderRS16.cpp | 6 +- src/common/normEncoderRS8.cpp | 6 +- src/common/normFile.cpp | 2 +- src/common/normNode.cpp | 14 +- src/common/normObject.cpp | 44 ++++- src/common/normSegment.cpp | 6 +- src/common/normSession.cpp | 65 ++++---- 16 files changed, 507 insertions(+), 61 deletions(-) diff --git a/examples/normServer.cpp b/examples/normServer.cpp index f448b54..a500243 100644 --- a/examples/normServer.cpp +++ b/examples/normServer.cpp @@ -203,7 +203,7 @@ ClientInfo NormGetClientInfo(NormNodeHandle client) return ClientInfo(version, addr, port); } // end NormGetClientInfo(NormNodeHandle) -ClientInfo NormGetSocketInfo(NormSocketHandle socket) +static ClientInfo NormGetSocketInfo(NormSocketHandle socket) { char addr[16]; // big enough for IPv6 unsigned int addrLen = 16; @@ -490,6 +490,11 @@ int main(int argc, char* argv[]) { if (event.socket == serverSocket) { + + // TBD - now that the NormSocket code manages its own client_table by remote addr/port + // and should eliminate the 'duplicative' connect itself, we can just keep track + // of client sockets by their NormSocketHandle + // Possibly a new "client" connecting to our "server" // First confirm that this really is a new client. if (NORM_SOCKET_INVALID != FindClientSocket(clientMap, clientInfo)) diff --git a/examples/normSocket.cpp b/examples/normSocket.cpp index ad6767b..a3b25d8 100644 --- a/examples/normSocket.cpp +++ b/examples/normSocket.cpp @@ -4,6 +4,9 @@ #include // for assert() #include // for strlen() +#include "protoTree.h" +#include "protoAddress.h" + #ifdef WIN32 #include // for inet_ntoa() (TBD - change to use Protolib routines?) #include // for inet_ntop() @@ -64,6 +67,110 @@ const char* NormNodeGetAddressString(NormNodeHandle node) } } // end NormNodeGetAddressString() + +class NormSocketInfo : public ProtoTree::Item +{ + public: + NormSocketInfo(unsigned int remoteAddrLen, const char* remoteAddr, UINT16 remotePort) + : norm_socket(NULL) + { + info_keysize = MakeKey(info_key, remoteAddrLen, remoteAddr, remotePort); + } + + // copy constructor + NormSocketInfo(const NormSocketInfo& s) + {*this = s;} + + ~NormSocketInfo() {} + + void SetSocket(class NormSocket* theSocket) + {norm_socket = theSocket;} + class NormSocket* GetSocket() + {return norm_socket;} + + static unsigned int MakeKey(unsigned char* key, unsigned int remoteAddrLen, const char* remoteAddr, UINT16 remotePort) + { + key[0] = remoteAddrLen; + memcpy(key + 1, remoteAddr, remoteAddrLen); + unsigned int keysize = remoteAddrLen + 1; + memcpy(key + keysize, &remotePort, 2); + keysize += 2; + keysize <<= 3; // to size in 'bits' + return keysize; + } + + void GetRemoteAddress(ProtoAddress& theAddr) const + { + int remoteAddrLen = info_key[0]; + const char* remoteAddrPtr = (char*)info_key + 1; + ProtoAddress::Type addrType; + switch (remoteAddrLen) + { + case 4: + addrType = ProtoAddress::IPv4; + break; + case 16: + addrType = ProtoAddress::IPv6; + break; + default: + theAddr.Invalidate(); + ASSERT(0); + return; + } + theAddr.SetRawHostAddress(addrType, remoteAddrPtr, remoteAddrLen); + UINT16 remotePort; + memcpy(&remotePort, remoteAddrPtr + remoteAddrLen, 2); + theAddr.SetPort(remotePort); + } + + const char* GetKey() const + {return (const char*)info_key;} + unsigned int GetKeysize() const + {return info_keysize;} + + private: + // remoteAddrLen + remoteAddr + remotePort + // 1 + 16 max + 2 + unsigned char info_key[19]; + unsigned int info_keysize; + class NormSocket* norm_socket; // may be NULL if it is pending acceptance + +}; // end class NormSocketInfo + +// helper function +NormSocketInfo NormGetSocketInfo(NormNodeHandle client) +{ + char remoteAddr[16]; // big enough for IPv6 + unsigned int remoteAddrLen = 16; + UINT16 remotePort; + NormNodeGetAddress(client, remoteAddr, &remoteAddrLen, &remotePort); + return NormSocketInfo(remoteAddrLen, remoteAddr, remotePort); +} // end NormGetSocketInfo() + +class NormSocketTable : public ProtoTreeTemplate +{ + public: + NormSocketInfo* FindSocketInfo(UINT16 remotePort, unsigned int remoteAddrLen, const char* remoteAddr) + { + unsigned char key[19]; + unsigned int keysize = NormSocketInfo::MakeKey(key, remoteAddrLen, remoteAddr, remotePort); + return Find((char*)key, keysize); + } + + NormSocketInfo* FindSocketInfo(NormNodeHandle client) + { + NormSocketInfo socketInfo = NormGetSocketInfo(client); + return Find(socketInfo.GetKey(), socketInfo.GetKeysize()); + } + + void RemoveSocketInfo(NormSocketInfo& socketInfo) + { + // safety dance + if (NULL == Find(socketInfo.GetKey(), socketInfo.GetKeysize())) return; // not on the dance floor + Remove(socketInfo); + } +}; // end class NormSocketTable + class NormSocket { public: @@ -89,6 +196,9 @@ class NormSocket {return (NULL != server_socket);} bool IsClientSide() const {return (NULL == server_socket);} + + NormSocket* GetServerSocket() + {return server_socket;} NormInstanceHandle GetInstance() const {return NormGetInstance(norm_session);} @@ -134,6 +244,18 @@ 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 + {client_table.RemoveSocketInfo(socketInfo);} + void SetUserData(const void* userData) {user_data = userData;} const void* GetUserData() const @@ -169,6 +291,12 @@ class NormSocket void RemoveAckingNode(NormNodeId nodeId) {NormRemoveAckingNode(norm_session, nodeId);} + UINT16 GetLocalPort() const + {return (NORM_SESSION_INVALID != norm_session) ? NormGetRxPort(norm_session) : 0;} + + //bool GetLocalAddress(char* addr, unsigned int& addrLen, UINT16& port) + // {return NormGetRxBindAddress(norm_session, addr, addrLen, port)} + void GetPeerName(char* addr, unsigned int* addrLen, UINT16* port) { if (NULL == addrLen) return; @@ -206,6 +334,8 @@ class NormSocket NormSessionHandle norm_session; NormSessionHandle mcast_session; // equals norm_session for a multicast server NormSocket* server_socket; // only applies to server-side sockets + NormSocketTable client_table; // only applies to server sockets + NormSocketInfo* socket_info; // only applies to server-side, client sockets unsigned int client_count; // only applies to mcast server sockets NormNodeId client_id; // only applies to mcast client socket NormNodeHandle remote_node; // client socket peer info @@ -227,10 +357,11 @@ class NormSocket }; // end class NormSocket + NormSocket::NormSocket(NormSessionHandle normSession) : socket_state(CLOSED), norm_session(normSession), mcast_session(NORM_SESSION_INVALID), server_socket(NULL), - client_count(0), client_id(NORM_NODE_NONE), + socket_info(NULL), client_count(0), client_id(NORM_NODE_NONE), remote_node(NORM_NODE_INVALID), remote_version(0), remote_port(0), tx_stream(NORM_OBJECT_INVALID), tx_ready(false), tx_segment_size(0), tx_stream_buffer_max(0), tx_stream_buffer_count(0), @@ -277,10 +408,30 @@ bool NormSocket::Listen(UINT16 serverPort, const char* groupAddr, const char* se { if (OPEN != socket_state) { - fprintf(stderr, "NormSocket::Listen() error: socket not open?!\n"); - return false; + /* This wasn't a good idea (yet and maybe never) + if ((CLOSED == socket_state) && (NORM_SESSION_INVALID != norm_session)) + { + // closed socekt, not in use, so re-open socket .. + NormInstanceHandle instance = NormGetInstance(norm_session); + NormSessionHandle oldSession = norm_session; + if (!Open(instance)) + { + norm_session = oldSession; + perror("NormSocket::Listen() error: unable to reopen socket"); + return false; + } + else + { + NormDestroySession(oldSession); + } + } + else*/ + { + fprintf(stderr, "NormSocket::Listen() error: socket not open!?\n"); + return false; + } } - // THe code below will be cleaned/tightened up somewhat once all is working + // The code below will be cleaned/tightened up somewhat once all is working // Note that port reuse here lets us manage our "client" rx-only unicast connections the // way we need, but does allow a second multicast server to be started on this group which leads @@ -479,6 +630,32 @@ bool NormSocket::Connect(const char* serverAddr, const char* groupAddr, NormNodeId clientId) { + if (OPEN != socket_state) + { + /* Not a good idea (yet and maybe never) + if ((CLOSED == socket_state) && (NORM_SESSION_INVALID != norm_session)) + { + // closed socekt, not in use, so re-open socket .. + NormInstanceHandle instance = NormGetInstance(norm_session); + NormSessionHandle oldSession = norm_session; + if (!Open(instance)) + { + norm_session = oldSession; + perror("NormSocket::Connect() error: unable to reopen socket"); + return false; + } + else + { + NormDestroySession(oldSession); + } + } + else + */ + { + fprintf(stderr, "NormSocket::Connect() error: socket not open!?\n"); + return false; + } + } // For unicast connections, the "client" manages a single NormSession for send and receive // (For multicast connections, there are two sessions: The same unicast session that will // be set to txOnly upon CONNECT and a separate NormSession for multicast reception) @@ -555,6 +732,7 @@ 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 @@ -700,6 +878,7 @@ 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) @@ -710,6 +889,12 @@ void NormSocket::Shutdown() } socket_state = CLOSING; } + 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() void NormSocket::Close() @@ -729,6 +914,7 @@ 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 @@ -750,9 +936,30 @@ void NormSocket::Close() } if (NORM_SESSION_INVALID != norm_session) { + NormCancelUserTimer(norm_session); NormStopSender(norm_session); NormStopReceiver(norm_session); } + if (NULL != socket_info) + { + if (NULL != server_socket) + server_socket->RemoveSocketInfo(*socket_info); + delete socket_info; + socket_info = NULL; + } + + // Iterate through remaining socket info and disassociate from any clients remaining + NormSocketTable::Iterator iterator(client_table); + NormSocketInfo* socketInfo; + while (NULL != (socketInfo = iterator.GetNextItem())) + { + NormSocket* clientSocket = socketInfo->GetSocket(); + if (NULL != clientSocket) + clientSocket->SetSocketInfo(NULL); + client_table.Remove(*socketInfo); + delete socketInfo; + } + server_socket = NULL; remote_node = NORM_NODE_INVALID; tx_stream = NORM_OBJECT_INVALID; @@ -861,6 +1068,7 @@ 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); } } @@ -891,8 +1099,33 @@ void NormSocket::GetSocketEvent(const NormEvent& event, NormSocketEvent& socketE switch (socket_state) { case LISTENING: - socketEvent.type = NORM_SOCKET_ACCEPT; + { + NormSocketInfo* socketInfo = client_table.FindSocketInfo(event.sender); + if (NULL == socketInfo) + { + // Add info for client socket pending acceptance + socketInfo = new NormSocketInfo(NormGetSocketInfo(event.sender)); + if (NULL != socketInfo) + { + client_table.Insert(*socketInfo); + socketEvent.type = NORM_SOCKET_ACCEPT; + } + else + { + perror("NormSocket::GetSocketEvent() error: unable to add pending client info to server socket:\n"); + } + } + else // duplicative accept event for existing socket, so ignore + { + ProtoAddress remoteAddr; + socketInfo->GetRemoteAddress(remoteAddr); + fprintf(stderr, "NormSocket::GetSocketEvent() warning: duplicative %s from client %s/%hu...\n", + (NORM_REMOTE_SENDER_NEW == event.type) ? "new" : "reset", + remoteAddr.GetHostString(), remoteAddr.GetPort()); + // TBD - should we go ahead and delete this event.sender??? + } break; + } case ACCEPTING: if (IsServerSide() && IsClientSocket() && (NORM_NODE_INVALID != remote_node)) { @@ -901,6 +1134,7 @@ 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); @@ -983,6 +1217,7 @@ void NormSocket::GetSocketEvent(const NormEvent& event, NormSocketEvent& socketE case NORM_SEND_ERROR: { + TRACE("NormSocket got SEND ERROR\n"); switch (socket_state) { case CONNECTING: @@ -1153,6 +1388,7 @@ void NormSocket::GetSocketEvent(const NormEvent& event, NormSocketEvent& socketE else { // This still allows at least a chance of an ACK to be sent upon completion + TRACE("NormSetUserTimer(5) session: %p\n", norm_session); NormSetUserTimer(norm_session, 0.0); } socketEvent.type = NORM_SOCKET_CLOSING; @@ -1224,10 +1460,44 @@ NormSocketHandle NormAccept(NormSocketHandle serverSocket, NormNodeHandle client NormSuspendInstance(serverInstance); NormSocketHandle clientSocket = s->Accept(client, instance); NormResumeInstance(serverInstance); + if (NORM_SOCKET_INVALID != clientSocket) + { + // Keep track of this client socket in our serverSocket socket_table + NormSocketInfo* socketInfo = s->FindSocketInfo(client); + ASSERT(NULL != socketInfo); + NormSocket* c = (NormSocket*)clientSocket; + socketInfo->SetSocket(c); + c->SetSocketInfo(socketInfo); + } return clientSocket; } // end NormAccept() +NORM_API_LINKAGE +extern bool NormSendCommandTo(NormSessionHandle sessionHandle, + const char* cmdBuffer, + unsigned int cmdLength, + const char* addr, + UINT16 port); + +void NormReject(NormSocketHandle serverSocket, + NormNodeHandle clientNode) +{ + // Simple, single "reject" command for moment (TBD - do something more stateful so app will be bothered less) + // Send "reject" command to source + char buffer[2]; + buffer[0] = NORM_SOCKET_VERSION; + buffer[1] = NORM_SOCKET_CMD_REJECT; + NormSocket* s = (NormSocket*)serverSocket; + NormSocketInfo socketInfo = NormGetSocketInfo(clientNode); + ProtoAddress dest; + socketInfo.GetRemoteAddress(dest); + char destString[64]; + destString[63] = '\0'; + dest.GetHostString(destString, 63); + NormSendCommandTo(s->GetSession(), buffer, 2,destString, dest.GetPort()); +} // end NormReject() + // TBD - provide options for binding to a specific local address, interface, etc bool NormConnect(NormSocketHandle normSocket, const char* serverAddr, UINT16 serverPort, UINT16 localPort, const char* groupAddr, NormNodeId clientId) { @@ -1248,7 +1518,7 @@ ssize_t NormWrite(NormSocketHandle normSocket, const void *buf, size_t nbyte) NormSocket* s = (NormSocket*)normSocket; NormInstanceHandle instance = s->GetInstance(); NormSuspendInstance(instance); - ssize_t result = (ssize_t)s->Write((const char*)buf, nbyte); + ssize_t result = (ssize_t)s->Write((const char*)buf, (unsigned int)nbyte); NormResumeInstance(instance); return result; } // end NormWrite() @@ -1271,7 +1541,7 @@ ssize_t NormRead(NormSocketHandle normSocket, void *buf, size_t nbyte) NormInstanceHandle instance = s->GetInstance(); NormSuspendInstance(instance); // TBD - make sure s->rx_stream is valid - unsigned int numBytes = nbyte; + unsigned int numBytes = (unsigned int)nbyte; ssize_t result; if (s->Read((char*)buf, numBytes)) result = numBytes; diff --git a/examples/normSocket.h b/examples/normSocket.h index d62de1f..763e2c2 100644 --- a/examples/normSocket.h +++ b/examples/normSocket.h @@ -78,6 +78,9 @@ bool NormConnect(NormSocketHandle normSocket, NormSocketHandle NormAccept(NormSocketHandle serverSocket, NormNodeHandle clientNode, NormInstanceHandle instance = NORM_INSTANCE_INVALID); + +void NormReject(NormSocketHandle serverSocket, + NormNodeHandle clientNode); void NormShutdown(NormSocketHandle normSocket); diff --git a/include/normApi.h b/include/normApi.h index 9ee1d00..cc6e2bc 100644 --- a/include/normApi.h +++ b/include/normApi.h @@ -367,6 +367,9 @@ void NormSetRxPortReuse(NormSessionHandle sessionHandle, NORM_API_LINKAGE UINT16 NormGetRxPort(NormSessionHandle sessionHandle); +NORM_API_LINKAGE +bool NormGetRxBindAddress(NormSessionHandle sessionHandle, char* addr, unsigned int& addrLen, UINT16& port); + // TBD - We should probably have a "NormSetCCMode(NormCCMode ccMode)" function for users NORM_API_LINKAGE void NormSetEcnSupport(NormSessionHandle sessionHandle, @@ -580,6 +583,9 @@ void NormStreamSetPushEnable(NormObjectHandle streamHandle, NORM_API_LINKAGE bool NormStreamHasVacancy(NormObjectHandle streamHandle); +NORM_API_LINKAGE +unsigned int NormStreamGetVacancy(NormObjectHandle streamHandle, unsigned int bytesWanted = 0); + NORM_API_LINKAGE void NormStreamMarkEom(NormObjectHandle streamHandle); diff --git a/include/normNode.h b/include/normNode.h index cc24c5a..cb2355a 100644 --- a/include/normNode.h +++ b/include/normNode.h @@ -134,7 +134,7 @@ class NormLossEstimator2 history[1] = (unsigned int)((1.0 / lossFraction) + 0.5); } unsigned long CurrentLossInterval() {return history[0];} - unsigned int LastLossInterval() {return history[1];} + unsigned long LastLossInterval() {return history[1];} void SetIgnoreLoss(bool state) {ignore_loss = state;} diff --git a/include/normObject.h b/include/normObject.h index 7f0212f..2446e79 100644 --- a/include/normObject.h +++ b/include/normObject.h @@ -484,6 +484,10 @@ class NormStreamObject : public NormObject bool HasVacancy() {return (stream_closing ? false : write_vacancy);} + // Returns how many bytes can be written to stream without blocking + // (up to 'wanted' for non-zero 'wanted', otherwise max vacancy available) + unsigned int GetVacancy(unsigned int wanted = 0); + NormBlock* StreamBlockLo() {return stream_buffer.Find(stream_buffer.RangeLo());} void SetLastNackTime(NormBlockId blockId, const ProtoTime& theTime) diff --git a/include/normSession.h b/include/normSession.h index f444cca..6208e8e 100644 --- a/include/normSession.h +++ b/include/normSession.h @@ -236,6 +236,9 @@ class NormSession UINT16 GetRxPort() const; + const ProtoAddress& GetRxBindAddr() const + {return rx_bind_addr;} + // "SetEcnSupport(true)" sets up raw packet capture (pcap) so that incoming packet // ECN status may be checked // NOTE: only effective _before_ sndr/rcvr startup! @@ -428,6 +431,9 @@ class NormSession bool SenderSendCmd(const char* cmdBuffer, unsigned int cmdLength, bool robust); void SenderCancelCmd(); + // The following method is currently only used for NormSocket purposes + bool SenderSendAppCmd(const char* buffer, unsigned int length, const ProtoAddress& dst); + void SenderSetSynStatus(bool state) {syn_status = state;} @@ -697,8 +703,6 @@ class NormSession bool SenderBuildRepairAdv(NormCmdRepairAdvMsg& cmd); void SenderUpdateGroupSize(); bool SenderQueueAppCmd(); - // The following method is only used for NormSocket purposes - bool SenderSendAppCmd(const char* buffer, unsigned int length, const ProtoAddress& dst); // Receiver message handling routines void ReceiverHandleObjectMessage(const struct timeval& currentTime, diff --git a/makefiles/Makefile.common b/makefiles/Makefile.common index aa5ee09..f5bf585 100644 --- a/makefiles/Makefile.common +++ b/makefiles/Makefile.common @@ -187,7 +187,7 @@ normStreamer: $(STREAMER_OBJ) libnorm.a $(LIBPROTO) mkdir -p ../bin cp $@ ../bin/$@ -# (nocmCast) file sender/receiver +# (normCast) file sender/receiver CAST_SRC = $(EXAMPLE)/normCast.cpp CAST_OBJ = $(CAST_SRC:.cpp=.o) @@ -195,6 +195,27 @@ normCast: $(CAST_OBJ) libnorm.a $(LIBPROTO) $(CC) $(CFLAGS) -o $@ $(CAST_OBJ) $(LDFLAGS) libnorm.a $(LIBPROTO) $(LIBS) mkdir -p ../bin cp $@ ../bin/$@ + +# These are the new "NormSocket" API extension examples +SERVER_SRC = $(EXAMPLE)/normServer.cpp $(EXAMPLE)/normSocket.cpp +SERVER_OBJ = $(SERVER_SRC:.cpp=.o) + +normServer: $(SERVER_OBJ) libnorm.a $(LIBPROTO) + $(CC) $(CFLAGS) -o $@ $(SERVER_OBJ) $(LDFLAGS) libnorm.a $(LIBPROTO) $(LIBS) + mkdir -p ../bin + cp $@ ../bin/$@ + + +CLIENT_SRC = $(EXAMPLE)/normClient.cpp $(EXAMPLE)/normSocket.cpp +CLIENT_OBJ = $(CLIENT_SRC:.cpp=.o) + +normClient: $(CLIENT_OBJ) libnorm.a $(LIBPROTO) + $(CC) $(CFLAGS) -o $@ $(CLIENT_OBJ) $(LDFLAGS) libnorm.a $(LIBPROTO) $(LIBS) + mkdir -p ../bin + cp $@ ../bin/$@ + + + # (pcap2norm) - parses pcap (e.g. tcpdump) file and prints NORM trace PCAP_SRC = $(COMMON)/pcap2norm.cpp diff --git a/src/common/normApi.cpp b/src/common/normApi.cpp index 09e4ed9..66f58fa 100644 --- a/src/common/normApi.cpp +++ b/src/common/normApi.cpp @@ -220,8 +220,11 @@ 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; } @@ -554,6 +557,7 @@ bool NormInstance::GetNextEvent(NormEvent* theEvent) case NORM_SEND_ERROR: { NormSession* session = (NormSession*)next->event.session; + TRACE("calling ClearSendError() ....\n"); session->ClearSendError(); break; } @@ -1073,6 +1077,40 @@ UINT16 NormGetRxPort(NormSessionHandle sessionHandle) } // end NormGetRxPort() +NORM_API_LINKAGE +bool NormGetRxBindAddress(NormSessionHandle sessionHandle, char* addr, unsigned int& addrLen, UINT16& port) +{ + bool result = false; + NormInstance* instance = NormInstance::GetInstanceFromSession(sessionHandle); + if ((NULL != instance) && instance->dispatcher.SuspendThread()) + { + NormSession* session = (NormSession*)sessionHandle; + port = session->GetRxPort(); + ProtoAddress bindAddr = session->GetRxBindAddr(); + if (!bindAddr.IsValid()) + { + addrLen = 0; + result = true; + } + else if (addrLen < bindAddr.GetLength()) + { + addrLen = bindAddr.GetLength(); + } + else + { + addrLen = bindAddr.GetLength(); + memcpy(addr, bindAddr.GetRawHostAddress(), addrLen); + result = true; + } + instance->dispatcher.ResumeThread(); + } + else + { + addrLen = port = 0; + } + return result; +} // end NormGetRxBindAddress() + NORM_API_LINKAGE bool NormSetTxPort(NormSessionHandle sessionHandle, UINT16 txPort, @@ -1132,7 +1170,7 @@ bool NormPresetObjectInfo(NormSessionHandle sessionHandle, NormSession* session = (NormSession*)sessionHandle; if (session) { - result = session->SetPresetFtiData(objectSize, segmentSize, numData, numParity); + result = session->SetPresetFtiData((unsigned int)objectSize, segmentSize, numData, numParity); if (result) session->SenderSetFtiMode(NormSession::FTI_PRESET); } instance->dispatcher.ResumeThread(); @@ -1511,7 +1549,7 @@ NormSessionId NormGetRandomSessionId() { ProtoTime currentTime; currentTime.GetCurrentTime(); - srand(currentTime.usec()); // seed random number generator + srand((unsigned int)currentTime.usec()); // seed random number generator return (NormSessionId)rand(); } // end NormGetRandomSessionId() @@ -1956,6 +1994,22 @@ bool NormStreamHasVacancy(NormObjectHandle streamHandle) return result; } // end NormStreamHasVacancy() +NORM_API_LINKAGE +unsigned int NormStreamGetVacancy(NormObjectHandle streamHandle, unsigned int bytesWanted) +{ + bool result = false; + NormInstance* instance = NormInstance::GetInstanceFromObject(streamHandle); + if (instance && instance->dispatcher.SuspendThread()) + { + NormStreamObject* stream = + static_cast((NormObject*)streamHandle); + if (NULL != stream) + result = stream->GetVacancy(bytesWanted); + instance->dispatcher.ResumeThread(); + } + return result; +} // end NormStreamGetVacancy() + NORM_API_LINKAGE void NormStreamMarkEom(NormObjectHandle streamHandle) { @@ -2247,6 +2301,32 @@ void NormCancelCommand(NormSessionHandle sessionHandle) } } // end NormCancelCommand() + +// This one is not part of the public API (for NormSocket use currently) +NORM_API_LINKAGE +bool NormSendCommandTo(NormSessionHandle sessionHandle, + const char* cmdBuffer, + unsigned int cmdLength, + const char* addr, + UINT16 port) +{ + bool result = false; + NormInstance* instance = NormInstance::GetInstanceFromSession(sessionHandle); + if (instance && instance->dispatcher.SuspendThread()) + { + NormSession* session = (NormSession*)sessionHandle; + ProtoAddress dest; + if (dest.ResolveFromString(addr)) + { + dest.SetPort(port); + result = session->SenderSendAppCmd(cmdBuffer, cmdLength, dest); + } + instance->dispatcher.ResumeThread(); + } + return result; +} // end NormSendCommandTo() + + NORM_API_LINKAGE void NormSetSynStatus(NormSessionHandle sessionHandle, bool state) { @@ -2457,7 +2537,7 @@ bool NormPreallocateRemoteSender(NormSessionHandle sessionHandle, if (instance && instance->dispatcher.SuspendThread()) { NormSession* session = (NormSession*)sessionHandle; - result = session->PreallocateRemoteSender(bufferSize, segmentSize, numData, numParity, streamBufferSize); + result = session->PreallocateRemoteSender((unsigned int)bufferSize, segmentSize, numData, numParity, streamBufferSize); instance->dispatcher.ResumeThread(); } return result; @@ -2754,7 +2834,7 @@ NORM_API_LINKAGE bool NormNodeGetAddress(NormNodeHandle nodeHandle, char* addrBuffer, unsigned int* bufferLen, - UINT16* port) + UINT16* port) { bool result = false; if (NORM_NODE_INVALID != nodeHandle) diff --git a/src/common/normEncoderRS16.cpp b/src/common/normEncoderRS16.cpp index 1aba353..5cf0757 100644 --- a/src/common/normEncoderRS16.cpp +++ b/src/common/normEncoderRS16.cpp @@ -156,7 +156,7 @@ inline gf gf_mul(int x, int y) } #define init_mul_table() -#define USE_GF_MULC register gf * __gf_mulc_ +#define USE_GF_MULC gf * __gf_mulc_ #define GF_MULC0(c) __gf_mulc_ = &gf_exp[ gf_log[c] ] #define GF_ADDMULC(dst, x) { if (x) dst ^= __gf_mulc_[ gf_log[x] ] ; } @@ -261,8 +261,8 @@ static void generate_gf() static void addmul1(gf* dst1, gf* src1, gf c, int sz) { USE_GF_MULC ; - register gf* dst = dst1; - register gf* src = src1 ; + gf* dst = dst1; + gf* src = src1 ; gf* lim = &dst[sz - UNROLL + 1] ; GF_MULC0(c) ; diff --git a/src/common/normEncoderRS8.cpp b/src/common/normEncoderRS8.cpp index 778efba..a1334ca 100644 --- a/src/common/normEncoderRS8.cpp +++ b/src/common/normEncoderRS8.cpp @@ -133,7 +133,7 @@ static inline gf modnn(int x) static gf gf_mul_table[GF_SIZE + 1][GF_SIZE + 1]; #define gf_mul(x,y) gf_mul_table[x][y] -#define USE_GF_MULC register gf * __gf_mulc_ +#define USE_GF_MULC gf * __gf_mulc_ #define GF_MULC0(c) __gf_mulc_ = gf_mul_table[c] #define GF_ADDMULC(dst, x) dst ^= __gf_mulc_[x] @@ -262,8 +262,8 @@ static void generate_gf() static void addmul1(gf* dst1, gf* src1, gf c, int sz) { USE_GF_MULC ; - register gf* dst = dst1; - register gf* src = src1 ; + gf* dst = dst1; + gf* src = src1 ; gf* lim = &dst[sz - UNROLL + 1] ; GF_MULC0(c) ; diff --git a/src/common/normFile.cpp b/src/common/normFile.cpp index 535d8b9..bb920b5 100644 --- a/src/common/normFile.cpp +++ b/src/common/normFile.cpp @@ -690,7 +690,7 @@ bool NormDirectoryIterator::GetNextFile(char* fileName) NormFile::Type type = NormFile::GetType(fileName); if (NormFile::NORMAL == type) { - int nameLen = strlen(fileName); + size_t nameLen = strlen(fileName); nameLen = MIN(PATH_MAX, nameLen); nameLen -= path_len; memmove(fileName, fileName+path_len, nameLen); diff --git a/src/common/normNode.cpp b/src/common/normNode.cpp index 2393a1d..b83b85c 100644 --- a/src/common/normNode.cpp +++ b/src/common/normNode.cpp @@ -238,7 +238,7 @@ bool NormSenderNode::AllocateBuffers(unsigned int bufferSpace, unsigned long numSegments = numBlocks * segPerBlock; - if (!block_pool.Init(numBlocks, blockSize)) + if (!block_pool.Init((UINT32)numBlocks, blockSize)) { PLOG(PL_FATAL, "NormSenderNode::AllocateBuffers() block_pool init error\n"); Close(); @@ -246,7 +246,7 @@ bool NormSenderNode::AllocateBuffers(unsigned int bufferSpace, } // Segment buffers include space for NORM_OBJECT_STREAM stream payload header - if (!segment_pool.Init(numSegments, segmentSize+NormDataMsg::GetStreamPayloadHeaderLength())) + if (!segment_pool.Init((unsigned int)numSegments, segmentSize+NormDataMsg::GetStreamPayloadHeaderLength())) { PLOG(PL_FATAL, "NormSenderNode::AllocateBuffers() segment_pool init error\n"); Close(); @@ -1571,8 +1571,12 @@ void NormSenderNode::HandleObjectMessage(const NormObjectMsg& msg) } // else wait for NORM_INFO message with sender FTI } - if (gotFTI && !AllocateBuffers(session.RemoteSenderBufferSize(), fecId, ftiData.GetFecInstanceId(), ftiData.GetFecFieldSize(), - ftiData.GetSegmentSize(), ftiData.GetFecMaxBlockLen(), ftiData.GetFecNumParity())) + if (gotFTI && !AllocateBuffers((unsigned int)session.RemoteSenderBufferSize(), + fecId, ftiData.GetFecInstanceId(), + ftiData.GetFecFieldSize(), + ftiData.GetSegmentSize(), + ftiData.GetFecMaxBlockLen(), + ftiData.GetFecNumParity())) { PLOG(PL_ERROR, "NormSenderNode::HandleObjectMessage() node>%lu sender>%lu buffer allocation error\n", (unsigned long)LocalNodeId(), (unsigned long)GetId()); @@ -1980,7 +1984,7 @@ void NormSenderNode::Sync(NormObjectId objectId) incrementResyncCount = true; // more than just a trim } unsigned long numBits = (UINT16)(objectId - firstPending); - rx_pending_mask.UnsetBits(firstPending, numBits); + rx_pending_mask.UnsetBits(firstPending, (UINT32)numBits); if (incrementResyncCount) IncrementResyncCount(); } } diff --git a/src/common/normObject.cpp b/src/common/normObject.cpp index 77218d9..38d75bf 100644 --- a/src/common/normObject.cpp +++ b/src/common/normObject.cpp @@ -3982,6 +3982,48 @@ void NormStreamObject::Terminate() session.TouchSender(); } // end NormStreamObject::Terminate() +unsigned int NormStreamObject::GetVacancy(unsigned int wanted) +{ + // Computes how many bytes are available for _immediate_ writing + ASSERT(Compare(write_index.block, tx_index.block) >= 0); + unsigned int maxDelta = block_pool.GetTotal() >> 1; + UINT32 blockDelta = (UINT32)Difference(write_index.block, tx_index.block); + if (blockDelta > maxDelta) return 0; + UINT32 nBytes = 0; + NormBlock* block = stream_buffer.Find(write_index.block); + if (NULL != block) + { + char* segment = block->GetSegment(write_index.segment); + if (NULL != segment) + { + UINT16 index = NormDataMsg::ReadStreamPayloadLength(segment); + nBytes += segment_size - index; + } + else + { + nBytes += segment_size; + } + nBytes += (ndata - write_index.segment - 1) * segment_size; + } + unsigned int blocksAllowed = maxDelta - blockDelta; + unsigned int poolCount = block_pool.GetCount(); + if (poolCount >= blocksAllowed) + poolCount = blocksAllowed; + nBytes += poolCount * ndata * segment_size; + + NormBlockBuffer::Iterator iterator(block_buffer); + while ((NULL != (block = iterator.GetNextBlock())) && + (blocksAllowed > 0) && + ((0 == wanted) || (nBytes < wanted))) + { + double delay = session.GetFlowControlDelay() - block->GetNackAge(); + if (block->IsPending() || (delay >= 1.0e-06)) break; + nBytes += (segment_size * ndata); + blocksAllowed--; + } + return nBytes; +} // end NormStreamObject::GetVacancy() + UINT32 NormStreamObject::Write(const char* buffer, UINT32 len, bool eom) { UINT32 nBytes = 0; @@ -3999,7 +4041,7 @@ UINT32 NormStreamObject::Write(const char* buffer, UINT32 len, bool eom) } // This old code detected buffer "fullness" by offset instead of segment index // but, the problem there was when apps wrote & flushed messages smaller than - // the segment_size, the buffer used up before this detected it. + // the segment_size, the buffer was used up before this detected it. //INT32 deltaOffset = write_offset - tx_offset; // (TBD) deprecate tx_offset //ASSERT(deltaOffset >= 0); //if (deltaOffset >= (INT32)object_size.LSB()) diff --git a/src/common/normSegment.cpp b/src/common/normSegment.cpp index 822c17d..e2a2fb7 100644 --- a/src/common/normSegment.cpp +++ b/src/common/normSegment.cpp @@ -789,7 +789,7 @@ NormBlockId NormBlockBuffer::RangeMin() const if (range_max > 1) { NormBlockId rangeMin = range_hi; - Decrement(rangeMin, range_max - 1); + Decrement(rangeMin, (UINT32)range_max - 1); return rangeMin; } else @@ -841,7 +841,7 @@ bool NormBlockBuffer::Insert(NormBlock* theBlock) // else if (blockId < range_lo) else if (Compare(blockId, range_lo) < 0) { - UINT32 newRange = (UINT32)Difference(range_lo, blockId) + range; + UINT32 newRange = (UINT32)Difference(range_lo, blockId) + (UINT32)range; if (newRange > range_max) return false; range_lo = blockId; range = newRange; @@ -849,7 +849,7 @@ bool NormBlockBuffer::Insert(NormBlock* theBlock) // else if (blockId > range_hi) else if (Compare(blockId, range_hi) > 0) { - UINT32 newRange = (UINT32)Difference(blockId, range_hi) + range; + UINT32 newRange = (UINT32)Difference(blockId, range_hi) + (UINT32)range; if (newRange > range_max) return false; range_hi = blockId; range = newRange; diff --git a/src/common/normSession.cpp b/src/common/normSession.cpp index b857f31..37a0c8d 100644 --- a/src/common/normSession.cpp +++ b/src/common/normSession.cpp @@ -254,6 +254,8 @@ 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)) @@ -774,14 +776,14 @@ bool NormSession::StartSender(UINT16 instanceId, if (numBlocks < 2) numBlocks = 2; unsigned long numSegments = numBlocks * numParity; - if (!block_pool.Init(numBlocks, blockSize)) + if (!block_pool.Init((UINT32)numBlocks, blockSize)) { PLOG(PL_FATAL, "NormSession::StartSender() block_pool init error\n"); StopSender(); return false; } - if (!segment_pool.Init(numSegments, segmentSize + NormDataMsg::GetStreamPayloadHeaderLength())) + if (!segment_pool.Init((unsigned int)numSegments, segmentSize + NormDataMsg::GetStreamPayloadHeaderLength())) { PLOG(PL_FATAL, "NormSession::StartSender() segment_pool init error\n"); StopSender(); @@ -2098,9 +2100,9 @@ bool NormSession::SetTxCacheBounds(NormObjectSize sizeMax, { bool result = true; tx_cache_size_max = sizeMax; - tx_cache_count_min = (countMin < countMax) ? countMin : countMax; + tx_cache_count_min = (unsigned int)((countMin < countMax) ? countMin : countMax); if (tx_cache_count_min < 1) tx_cache_count_min = 1; - tx_cache_count_max = (countMax > countMin) ? countMax : countMin; + tx_cache_count_max = (unsigned int)((countMax > countMin) ? countMax : countMin); if (tx_cache_count_max < 1) tx_cache_count_max = 1; tx_cache_count_min &= 0x00007fff; // limited to one-half of 16-bit NormObjectId space @@ -2127,15 +2129,15 @@ bool NormSession::SetTxCacheBounds(NormObjectSize sizeMax, if (countMax != tx_table.GetRangeMax()) { tx_table.SetRangeMax((UINT16)countMax); - result = tx_pending_mask.Resize(countMax); - result &= tx_repair_mask.Resize(countMax); + result = tx_pending_mask.Resize((UINT32)countMax); + result &= tx_repair_mask.Resize((UINT32)countMax); if (!result) { countMax = tx_pending_mask.GetSize(); if (tx_repair_mask.GetSize() < countMax) countMax = tx_repair_mask.GetSize(); if (tx_cache_count_max > countMax) - tx_cache_count_max = countMax; + tx_cache_count_max = (unsigned int)countMax; if (tx_cache_count_min > tx_cache_count_max) tx_cache_count_min = tx_cache_count_max; } @@ -2238,6 +2240,7 @@ 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 @@ -2359,12 +2362,16 @@ void NormSession::RxSocketRecvHandler(ProtoSocket& theSocket, } else { + 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; } } @@ -3324,7 +3331,7 @@ void NormSession::SenderHandleCCFeedback(struct timeval currentTime, ccFlags = NormCC::RTT; ccRtt = savedRtt; ccLoss = savedLoss; - ccRate = savedRate, + ccRate = savedRate; ccSequence = savedSequence; currentTime = savedTime; } @@ -4344,6 +4351,27 @@ void NormSession::SenderCancelCmd() } } // end NormSession::SenderCancelCmd() +bool NormSession::SenderSendAppCmd(const char* buffer, unsigned int length, const ProtoAddress& dst) +{ + // Build/immediately send a NORM_CMD(APPLICATION) message + NormCmdAppMsg appMsg; + appMsg.Init(); + appMsg.SetDestination(address); + appMsg.SetGrtt(grtt_quantized); + appMsg.SetBackoffFactor((unsigned char)backoff_factor); + appMsg.SetGroupSize(gsize_quantized); + // We use a surrogate segment_size in case sender not configured (e.g. server-listener) + appMsg.SetContent(buffer, length, segment_size ? segment_size : 64); + appMsg.SetDestination(dst); + if (MSG_SEND_OK != SendMessage(appMsg)) + PLOG(PL_ERROR, "NormSession::SenderSendAppCmd() node>%lu sender unable to send app-defined cmd ...\n", + (unsigned long)LocalNodeId()); + else + PLOG(PL_DEBUG, "NormSession::SenderSendAppCmd() node>%lu sender sending app-defined cmd len:%u...\n", + (unsigned long)LocalNodeId(), appMsg.GetLength()); + return true; +} // end NormSession::SenderSendAppCmd() + bool NormSession::SenderQueueAppCmd() { if (0 == cmd_count) return false; @@ -4389,27 +4417,6 @@ bool NormSession::OnCmdTimeout(ProtoTimer& theTimer) return true; } // end NormSession::OnCmdTimeout() -bool NormSession::SenderSendAppCmd(const char* buffer, unsigned int length, const ProtoAddress& dst) -{ - // Build/immediately send a NORM_CMD(APPLICATION) message - NormCmdAppMsg appMsg; - appMsg.Init(); - appMsg.SetDestination(address); - appMsg.SetGrtt(grtt_quantized); - appMsg.SetBackoffFactor((unsigned char)backoff_factor); - appMsg.SetGroupSize(gsize_quantized); - // We use a surrogate segment_size in case sender not configured (e.g. server-listener) - appMsg.SetContent(buffer, length, segment_size ? segment_size : 64); - appMsg.SetDestination(dst); - if (MSG_SEND_OK != SendMessage(appMsg)) - PLOG(PL_ERROR, "NormSession::SenderSendAppCmd() node>%lu sender unable to send app-defined cmd ...\n", - (unsigned long)LocalNodeId()); - else - PLOG(PL_DEBUG, "NormSession::SenderSendAppCmd() node>%lu sender sending app-defined cmd len:%u...\n", - (unsigned long)LocalNodeId(), appMsg.GetLength()); - return true; -} // end NormSession::SenderSendAppCmd() - void NormSession::ActivateFlowControl(double delay, NormObjectId objectId, NormController::Event event) { flow_control_object = objectId;