BodgeMaster
/
NORM-mirror
mirror of https://github.com/USNavalResearchLaboratory/norm.git
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

added NormStreamGetVacancy() and some NormSocket improvement

pull/9/head
bebopagogo 2020-04-29 22:38:59 -04:00
parent 1c6b5304c7
commit 88e600cedb
16 changed files with 507 additions and 61 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

7
examples/normServer.cpp
View File

@ -203,7 +203,7 @@ ClientInfo NormGetClientInfo(NormNodeHandle client)
return ClientInfo(version, addr, port); return ClientInfo(version, addr, port);
} // end NormGetClientInfo(NormNodeHandle) } // end NormGetClientInfo(NormNodeHandle)
ClientInfo NormGetSocketInfo(NormSocketHandle socket) static ClientInfo NormGetSocketInfo(NormSocketHandle socket)
{ {
char addr[16]; // big enough for IPv6 char addr[16]; // big enough for IPv6
unsigned int addrLen = 16; unsigned int addrLen = 16;
@ -490,6 +490,11 @@ int main(int argc, char* argv[])
{ {
if (event.socket == serverSocket) 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" // Possibly a new "client" connecting to our "server"
// First confirm that this really is a new client. // First confirm that this really is a new client.
if (NORM_SOCKET_INVALID != FindClientSocket(clientMap, clientInfo)) if (NORM_SOCKET_INVALID != FindClientSocket(clientMap, clientInfo))

284
examples/normSocket.cpp
View File

@ -4,6 +4,9 @@
#include <assert.h> // for assert() #include <assert.h> // for assert()
#include <string.h> // for strlen() #include <string.h> // for strlen()
#include "protoTree.h"
#include "protoAddress.h"
#ifdef WIN32 #ifdef WIN32
#include <Winsock2.h> // for inet_ntoa() (TBD - change to use Protolib routines?) #include <Winsock2.h> // for inet_ntoa() (TBD - change to use Protolib routines?)
#include <Ws2tcpip.h> // for inet_ntop() #include <Ws2tcpip.h> // for inet_ntop()
@ -64,6 +67,110 @@ const char* NormNodeGetAddressString(NormNodeHandle node)
} }
} // end NormNodeGetAddressString() } // 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<NormSocketInfo>
{
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 class NormSocket
{ {
public: public:
@ -89,6 +196,9 @@ class NormSocket
{return (NULL != server_socket);} {return (NULL != server_socket);}
bool IsClientSide() const bool IsClientSide() const
{return (NULL == server_socket);} {return (NULL == server_socket);}
NormSocket* GetServerSocket()
{return server_socket;}
NormInstanceHandle GetInstance() const NormInstanceHandle GetInstance() const
{return NormGetInstance(norm_session);} {return NormGetInstance(norm_session);}
@ -134,6 +244,18 @@ class NormSocket
// hard, immediate closure // hard, immediate closure
void Close(); 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) void SetUserData(const void* userData)
{user_data = userData;} {user_data = userData;}
const void* GetUserData() const const void* GetUserData() const
@ -169,6 +291,12 @@ class NormSocket
void RemoveAckingNode(NormNodeId nodeId) void RemoveAckingNode(NormNodeId nodeId)
{NormRemoveAckingNode(norm_session, 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) void GetPeerName(char* addr, unsigned int* addrLen, UINT16* port)
{ {
if (NULL == addrLen) return; if (NULL == addrLen) return;
@ -206,6 +334,8 @@ class NormSocket
NormSessionHandle norm_session; NormSessionHandle norm_session;
NormSessionHandle mcast_session; // equals norm_session for a multicast server NormSessionHandle mcast_session; // equals norm_session for a multicast server
NormSocket* server_socket; // only applies to server-side sockets 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 unsigned int client_count; // only applies to mcast server sockets
NormNodeId client_id; // only applies to mcast client socket NormNodeId client_id; // only applies to mcast client socket
NormNodeHandle remote_node; // client socket peer info NormNodeHandle remote_node; // client socket peer info
@ -227,10 +357,11 @@ class NormSocket
}; // end class NormSocket }; // end class NormSocket
NormSocket::NormSocket(NormSessionHandle normSession) NormSocket::NormSocket(NormSessionHandle normSession)
: socket_state(CLOSED), norm_session(normSession), : socket_state(CLOSED), norm_session(normSession),
mcast_session(NORM_SESSION_INVALID), server_socket(NULL), 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), 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(NORM_OBJECT_INVALID), tx_ready(false), tx_segment_size(0),
tx_stream_buffer_max(0), tx_stream_buffer_count(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) if (OPEN != socket_state)
{ {
fprintf(stderr, "NormSocket::Listen() error: socket not open?!\n"); /* This wasn't a good idea (yet and maybe never)
return false; 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 // 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 // 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, const char* groupAddr,
NormNodeId clientId) 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 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 // (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) // be set to txOnly upon CONNECT and a separate NormSession for multicast reception)
@ -555,6 +732,7 @@ bool NormSocket::Connect(const char* serverAddr,
else else
{ {
// Set timeout for connect attempt (for "heavyweight" mcast connect, this would also be done) // 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); NormSetUserTimer(norm_session, NORM_DEFAULT_CONNECT_TIMEOUT);
} }
server_socket = NULL; // this is a client-side socket server_socket = NULL; // this is a client-side socket
@ -700,6 +878,7 @@ void NormSocket::Shutdown()
if ((IsServerSide() && IsMulticastClient()) || (NORM_OBJECT_INVALID == tx_stream)) if ((IsServerSide() && IsMulticastClient()) || (NORM_OBJECT_INVALID == tx_stream))
{ {
// Use a zero-timeout to immediately post NORM_SOCKET_CLOSE notification // Use a zero-timeout to immediately post NORM_SOCKET_CLOSE notification
TRACE("NormSetUserTimer(2) session: %p\n", norm_session);
NormSetUserTimer(norm_session, 0.0); NormSetUserTimer(norm_session, 0.0);
} }
else if (NORM_OBJECT_INVALID != tx_stream) else if (NORM_OBJECT_INVALID != tx_stream)
@ -710,6 +889,12 @@ void NormSocket::Shutdown()
} }
socket_state = CLOSING; 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() } // end NormSocket::Shutdown()
void NormSocket::Close() void NormSocket::Close()
@ -729,6 +914,7 @@ void NormSocket::Close()
NormNodeHandle node = NormGetAckingNodeHandle(mcast_session, nodeId); NormNodeHandle node = NormGetAckingNodeHandle(mcast_session, nodeId);
assert(NORM_NODE_INVALID != node); assert(NORM_NODE_INVALID != node);
NormSocket* clientSocket = (NormSocket*)NormNodeGetUserData(node); NormSocket* clientSocket = (NormSocket*)NormNodeGetUserData(node);
TRACE("NormSetUserTimer(3) session: %p\n", clientSocket->norm_session);
NormSetUserTimer(clientSocket->norm_session, 0.0); NormSetUserTimer(clientSocket->norm_session, 0.0);
} }
// for mcast server mcast_session == norm_session so it's destroyed below // 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) if (NORM_SESSION_INVALID != norm_session)
{ {
NormCancelUserTimer(norm_session);
NormStopSender(norm_session); NormStopSender(norm_session);
NormStopReceiver(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; server_socket = NULL;
remote_node = NORM_NODE_INVALID; remote_node = NORM_NODE_INVALID;
tx_stream = NORM_OBJECT_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 // 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) // (This will also remove the client from this server's acking list)
clientSocket->socket_state = CLOSING; clientSocket->socket_state = CLOSING;
TRACE("NormSetUserTimer(4) session: %p\n", clientSocket->norm_session);
NormSetUserTimer(clientSocket->norm_session, 0.0); NormSetUserTimer(clientSocket->norm_session, 0.0);
} }
} }
@ -891,8 +1099,33 @@ void NormSocket::GetSocketEvent(const NormEvent& event, NormSocketEvent& socketE
switch (socket_state) switch (socket_state)
{ {
case LISTENING: 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; break;
}
case ACCEPTING: case ACCEPTING:
if (IsServerSide() && IsClientSocket() && (NORM_NODE_INVALID != remote_node)) if (IsServerSide() && IsClientSocket() && (NORM_NODE_INVALID != remote_node))
{ {
@ -901,6 +1134,7 @@ void NormSocket::GetSocketEvent(const NormEvent& event, NormSocketEvent& socketE
case CONNECTING: case CONNECTING:
// TBD - We should validate that it's the right remote sender // TBD - We should validate that it's the right remote sender
// (i.e., by source address and/or nodeId) // (i.e., by source address and/or nodeId)
TRACE("NormCancelUserTimer(1) norm_session: %p\n", norm_session);
NormCancelUserTimer(norm_session); NormCancelUserTimer(norm_session);
socketEvent.type = NORM_SOCKET_CONNECT; socketEvent.type = NORM_SOCKET_CONNECT;
NormSetSynStatus(norm_session, false); NormSetSynStatus(norm_session, false);
@ -983,6 +1217,7 @@ void NormSocket::GetSocketEvent(const NormEvent& event, NormSocketEvent& socketE
case NORM_SEND_ERROR: case NORM_SEND_ERROR:
{ {
TRACE("NormSocket got SEND ERROR\n");
switch (socket_state) switch (socket_state)
{ {
case CONNECTING: case CONNECTING:
@ -1153,6 +1388,7 @@ void NormSocket::GetSocketEvent(const NormEvent& event, NormSocketEvent& socketE
else else
{ {
// This still allows at least a chance of an ACK to be sent upon completion // 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); NormSetUserTimer(norm_session, 0.0);
} }
socketEvent.type = NORM_SOCKET_CLOSING; socketEvent.type = NORM_SOCKET_CLOSING;
@ -1224,10 +1460,44 @@ NormSocketHandle NormAccept(NormSocketHandle serverSocket, NormNodeHandle client
NormSuspendInstance(serverInstance); NormSuspendInstance(serverInstance);
NormSocketHandle clientSocket = s->Accept(client, instance); NormSocketHandle clientSocket = s->Accept(client, instance);
NormResumeInstance(serverInstance); 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; return clientSocket;
} // end NormAccept() } // 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 // 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) 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; NormSocket* s = (NormSocket*)normSocket;
NormInstanceHandle instance = s->GetInstance(); NormInstanceHandle instance = s->GetInstance();
NormSuspendInstance(instance); 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); NormResumeInstance(instance);
return result; return result;
} // end NormWrite() } // end NormWrite()
@ -1271,7 +1541,7 @@ ssize_t NormRead(NormSocketHandle normSocket, void *buf, size_t nbyte)
NormInstanceHandle instance = s->GetInstance(); NormInstanceHandle instance = s->GetInstance();
NormSuspendInstance(instance); NormSuspendInstance(instance);
// TBD - make sure s->rx_stream is valid // TBD - make sure s->rx_stream is valid
unsigned int numBytes = nbyte; unsigned int numBytes = (unsigned int)nbyte;
ssize_t result; ssize_t result;
if (s->Read((char*)buf, numBytes)) if (s->Read((char*)buf, numBytes))
result = numBytes; result = numBytes;

3
examples/normSocket.h
View File

@ -78,6 +78,9 @@ bool NormConnect(NormSocketHandle normSocket,
NormSocketHandle NormAccept(NormSocketHandle serverSocket, NormSocketHandle NormAccept(NormSocketHandle serverSocket,
NormNodeHandle clientNode, NormNodeHandle clientNode,
NormInstanceHandle instance = NORM_INSTANCE_INVALID); NormInstanceHandle instance = NORM_INSTANCE_INVALID);
void NormReject(NormSocketHandle serverSocket,
NormNodeHandle clientNode);
void NormShutdown(NormSocketHandle normSocket); void NormShutdown(NormSocketHandle normSocket);

6
include/normApi.h
View File

@ -367,6 +367,9 @@ void NormSetRxPortReuse(NormSessionHandle sessionHandle,
NORM_API_LINKAGE NORM_API_LINKAGE
UINT16 NormGetRxPort(NormSessionHandle sessionHandle); 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 // TBD - We should probably have a "NormSetCCMode(NormCCMode ccMode)" function for users
NORM_API_LINKAGE NORM_API_LINKAGE
void NormSetEcnSupport(NormSessionHandle sessionHandle, void NormSetEcnSupport(NormSessionHandle sessionHandle,
@ -580,6 +583,9 @@ void NormStreamSetPushEnable(NormObjectHandle streamHandle,
NORM_API_LINKAGE NORM_API_LINKAGE
bool NormStreamHasVacancy(NormObjectHandle streamHandle); bool NormStreamHasVacancy(NormObjectHandle streamHandle);
NORM_API_LINKAGE
unsigned int NormStreamGetVacancy(NormObjectHandle streamHandle, unsigned int bytesWanted = 0);
NORM_API_LINKAGE NORM_API_LINKAGE
void NormStreamMarkEom(NormObjectHandle streamHandle); void NormStreamMarkEom(NormObjectHandle streamHandle);

2
include/normNode.h
View File

@ -134,7 +134,7 @@ class NormLossEstimator2
history[1] = (unsigned int)((1.0 / lossFraction) + 0.5); history[1] = (unsigned int)((1.0 / lossFraction) + 0.5);
} }
unsigned long CurrentLossInterval() {return history[0];} unsigned long CurrentLossInterval() {return history[0];}
unsigned int LastLossInterval() {return history[1];} unsigned long LastLossInterval() {return history[1];}
void SetIgnoreLoss(bool state) void SetIgnoreLoss(bool state)
{ignore_loss = state;} {ignore_loss = state;}

4
include/normObject.h
View File

@ -484,6 +484,10 @@ class NormStreamObject : public NormObject
bool HasVacancy() bool HasVacancy()
{return (stream_closing ? false : write_vacancy);} {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() NormBlock* StreamBlockLo()
{return stream_buffer.Find(stream_buffer.RangeLo());} {return stream_buffer.Find(stream_buffer.RangeLo());}
void SetLastNackTime(NormBlockId blockId, const ProtoTime& theTime) void SetLastNackTime(NormBlockId blockId, const ProtoTime& theTime)

8
include/normSession.h
View File

@ -236,6 +236,9 @@ class NormSession
UINT16 GetRxPort() const; UINT16 GetRxPort() const;
const ProtoAddress& GetRxBindAddr() const
{return rx_bind_addr;}
// "SetEcnSupport(true)" sets up raw packet capture (pcap) so that incoming packet // "SetEcnSupport(true)" sets up raw packet capture (pcap) so that incoming packet
// ECN status may be checked // ECN status may be checked
// NOTE: only effective _before_ sndr/rcvr startup! // NOTE: only effective _before_ sndr/rcvr startup!
@ -428,6 +431,9 @@ class NormSession
bool SenderSendCmd(const char* cmdBuffer, unsigned int cmdLength, bool robust); bool SenderSendCmd(const char* cmdBuffer, unsigned int cmdLength, bool robust);
void SenderCancelCmd(); 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) void SenderSetSynStatus(bool state)
{syn_status = state;} {syn_status = state;}
@ -697,8 +703,6 @@ class NormSession
bool SenderBuildRepairAdv(NormCmdRepairAdvMsg& cmd); bool SenderBuildRepairAdv(NormCmdRepairAdvMsg& cmd);
void SenderUpdateGroupSize(); void SenderUpdateGroupSize();
bool SenderQueueAppCmd(); 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 // Receiver message handling routines
void ReceiverHandleObjectMessage(const struct timeval& currentTime, void ReceiverHandleObjectMessage(const struct timeval& currentTime,

23
makefiles/Makefile.common
View File

@ -187,7 +187,7 @@ normStreamer: $(STREAMER_OBJ) libnorm.a $(LIBPROTO)
mkdir -p ../bin mkdir -p ../bin
cp $@ ../bin/$@ cp $@ ../bin/$@
# (nocmCast) file sender/receiver # (normCast) file sender/receiver
CAST_SRC = $(EXAMPLE)/normCast.cpp CAST_SRC = $(EXAMPLE)/normCast.cpp
CAST_OBJ = $(CAST_SRC:.cpp=.o) 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) $(CC) $(CFLAGS) -o $@ $(CAST_OBJ) $(LDFLAGS) libnorm.a $(LIBPROTO) $(LIBS)
mkdir -p ../bin mkdir -p ../bin
cp $@ ../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 # (pcap2norm) - parses pcap (e.g. tcpdump) file and prints NORM trace
PCAP_SRC = $(COMMON)/pcap2norm.cpp PCAP_SRC = $(COMMON)/pcap2norm.cpp

88
src/common/normApi.cpp
View File

@ -220,8 +220,11 @@ void NormInstance::Notify(NormController::Event event,
{ {
case SEND_OK: case SEND_OK:
// Purge any pending NORM_SEND_ERROR notifications for session // Purge any pending NORM_SEND_ERROR notifications for session
TRACE("purging SEND_ERROR ...\n");
PurgeNotifications(session, NORM_SEND_ERROR); PurgeNotifications(session, NORM_SEND_ERROR);
return; return;
case SEND_ERROR:
TRACE("got SEND_ERROR\n");
default: default:
break; break;
} }
@ -554,6 +557,7 @@ bool NormInstance::GetNextEvent(NormEvent* theEvent)
case NORM_SEND_ERROR: case NORM_SEND_ERROR:
{ {
NormSession* session = (NormSession*)next->event.session; NormSession* session = (NormSession*)next->event.session;
TRACE("calling ClearSendError() ....\n");
session->ClearSendError(); session->ClearSendError();
break; break;
} }
@ -1073,6 +1077,40 @@ UINT16 NormGetRxPort(NormSessionHandle sessionHandle)
} // end NormGetRxPort() } // 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 NORM_API_LINKAGE
bool NormSetTxPort(NormSessionHandle sessionHandle, bool NormSetTxPort(NormSessionHandle sessionHandle,
UINT16 txPort, UINT16 txPort,
@ -1132,7 +1170,7 @@ bool NormPresetObjectInfo(NormSessionHandle sessionHandle,
NormSession* session = (NormSession*)sessionHandle; NormSession* session = (NormSession*)sessionHandle;
if (session) 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); if (result) session->SenderSetFtiMode(NormSession::FTI_PRESET);
} }
instance->dispatcher.ResumeThread(); instance->dispatcher.ResumeThread();
@ -1511,7 +1549,7 @@ NormSessionId NormGetRandomSessionId()
{ {
ProtoTime currentTime; ProtoTime currentTime;
currentTime.GetCurrentTime(); currentTime.GetCurrentTime();
srand(currentTime.usec()); // seed random number generator srand((unsigned int)currentTime.usec()); // seed random number generator
return (NormSessionId)rand(); return (NormSessionId)rand();
} // end NormGetRandomSessionId() } // end NormGetRandomSessionId()
@ -1956,6 +1994,22 @@ bool NormStreamHasVacancy(NormObjectHandle streamHandle)
return result; return result;
} // end NormStreamHasVacancy() } // 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<NormStreamObject*>((NormObject*)streamHandle);
if (NULL != stream)
result = stream->GetVacancy(bytesWanted);
instance->dispatcher.ResumeThread();
}
return result;
} // end NormStreamGetVacancy()
NORM_API_LINKAGE NORM_API_LINKAGE
void NormStreamMarkEom(NormObjectHandle streamHandle) void NormStreamMarkEom(NormObjectHandle streamHandle)
{ {
@ -2247,6 +2301,32 @@ void NormCancelCommand(NormSessionHandle sessionHandle)
} }
} // end NormCancelCommand() } // 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 NORM_API_LINKAGE
void NormSetSynStatus(NormSessionHandle sessionHandle, bool state) void NormSetSynStatus(NormSessionHandle sessionHandle, bool state)
{ {
@ -2457,7 +2537,7 @@ bool NormPreallocateRemoteSender(NormSessionHandle sessionHandle,
if (instance && instance->dispatcher.SuspendThread()) if (instance && instance->dispatcher.SuspendThread())
{ {
NormSession* session = (NormSession*)sessionHandle; 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(); instance->dispatcher.ResumeThread();
} }
return result; return result;
@ -2754,7 +2834,7 @@ NORM_API_LINKAGE
bool NormNodeGetAddress(NormNodeHandle nodeHandle, bool NormNodeGetAddress(NormNodeHandle nodeHandle,
char* addrBuffer, char* addrBuffer,
unsigned int* bufferLen, unsigned int* bufferLen,
UINT16* port) UINT16* port)
{ {
bool result = false; bool result = false;
if (NORM_NODE_INVALID != nodeHandle) if (NORM_NODE_INVALID != nodeHandle)

6
src/common/normEncoderRS16.cpp
View File

@ -156,7 +156,7 @@ inline gf gf_mul(int x, int y)
} }
#define init_mul_table() #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_MULC0(c) __gf_mulc_ = &gf_exp[ gf_log[c] ]
#define GF_ADDMULC(dst, x) { if (x) dst ^= __gf_mulc_[ gf_log[x] ] ; } #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) static void addmul1(gf* dst1, gf* src1, gf c, int sz)
{ {
USE_GF_MULC ; USE_GF_MULC ;
register gf* dst = dst1; gf* dst = dst1;
register gf* src = src1 ; gf* src = src1 ;
gf* lim = &dst[sz - UNROLL + 1] ; gf* lim = &dst[sz - UNROLL + 1] ;
GF_MULC0(c) ; GF_MULC0(c) ;

6
src/common/normEncoderRS8.cpp
View File

@ -133,7 +133,7 @@ static inline gf modnn(int x)
static gf gf_mul_table[GF_SIZE + 1][GF_SIZE + 1]; static gf gf_mul_table[GF_SIZE + 1][GF_SIZE + 1];
#define gf_mul(x,y) gf_mul_table[x][y] #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_MULC0(c) __gf_mulc_ = gf_mul_table[c]
#define GF_ADDMULC(dst, x) dst ^= __gf_mulc_[x] #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) static void addmul1(gf* dst1, gf* src1, gf c, int sz)
{ {
USE_GF_MULC ; USE_GF_MULC ;
register gf* dst = dst1; gf* dst = dst1;
register gf* src = src1 ; gf* src = src1 ;
gf* lim = &dst[sz - UNROLL + 1] ; gf* lim = &dst[sz - UNROLL + 1] ;
GF_MULC0(c) ; GF_MULC0(c) ;

2
src/common/normFile.cpp
View File

@ -690,7 +690,7 @@ bool NormDirectoryIterator::GetNextFile(char* fileName)
NormFile::Type type = NormFile::GetType(fileName); NormFile::Type type = NormFile::GetType(fileName);
if (NormFile::NORMAL == type) if (NormFile::NORMAL == type)
{ {
int nameLen = strlen(fileName); size_t nameLen = strlen(fileName);
nameLen = MIN(PATH_MAX, nameLen); nameLen = MIN(PATH_MAX, nameLen);
nameLen -= path_len; nameLen -= path_len;
memmove(fileName, fileName+path_len, nameLen); memmove(fileName, fileName+path_len, nameLen);

14
src/common/normNode.cpp
View File

@ -238,7 +238,7 @@ bool NormSenderNode::AllocateBuffers(unsigned int bufferSpace,
unsigned long numSegments = numBlocks * segPerBlock; 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"); PLOG(PL_FATAL, "NormSenderNode::AllocateBuffers() block_pool init error\n");
Close(); Close();
@ -246,7 +246,7 @@ bool NormSenderNode::AllocateBuffers(unsigned int bufferSpace,
} }
// Segment buffers include space for NORM_OBJECT_STREAM stream payload header // 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"); PLOG(PL_FATAL, "NormSenderNode::AllocateBuffers() segment_pool init error\n");
Close(); Close();
@ -1571,8 +1571,12 @@ void NormSenderNode::HandleObjectMessage(const NormObjectMsg& msg)
} }
// else wait for NORM_INFO message with sender FTI // else wait for NORM_INFO message with sender FTI
} }
if (gotFTI && !AllocateBuffers(session.RemoteSenderBufferSize(), fecId, ftiData.GetFecInstanceId(), ftiData.GetFecFieldSize(), if (gotFTI && !AllocateBuffers((unsigned int)session.RemoteSenderBufferSize(),
ftiData.GetSegmentSize(), ftiData.GetFecMaxBlockLen(), ftiData.GetFecNumParity())) fecId, ftiData.GetFecInstanceId(),
ftiData.GetFecFieldSize(),
ftiData.GetSegmentSize(),
ftiData.GetFecMaxBlockLen(),
ftiData.GetFecNumParity()))
{ {
PLOG(PL_ERROR, "NormSenderNode::HandleObjectMessage() node>%lu sender>%lu buffer allocation error\n", PLOG(PL_ERROR, "NormSenderNode::HandleObjectMessage() node>%lu sender>%lu buffer allocation error\n",
(unsigned long)LocalNodeId(), (unsigned long)GetId()); (unsigned long)LocalNodeId(), (unsigned long)GetId());
@ -1980,7 +1984,7 @@ void NormSenderNode::Sync(NormObjectId objectId)
incrementResyncCount = true; // more than just a trim incrementResyncCount = true; // more than just a trim
} }
unsigned long numBits = (UINT16)(objectId - firstPending); unsigned long numBits = (UINT16)(objectId - firstPending);
rx_pending_mask.UnsetBits(firstPending, numBits); rx_pending_mask.UnsetBits(firstPending, (UINT32)numBits);
if (incrementResyncCount) IncrementResyncCount(); if (incrementResyncCount) IncrementResyncCount();
} }
} }

44
src/common/normObject.cpp
View File

@ -3982,6 +3982,48 @@ void NormStreamObject::Terminate()
session.TouchSender(); session.TouchSender();
} // end NormStreamObject::Terminate() } // 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 NormStreamObject::Write(const char* buffer, UINT32 len, bool eom)
{ {
UINT32 nBytes = 0; 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 // This old code detected buffer "fullness" by offset instead of segment index
// but, the problem there was when apps wrote & flushed messages smaller than // 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 //INT32 deltaOffset = write_offset - tx_offset; // (TBD) deprecate tx_offset
//ASSERT(deltaOffset >= 0); //ASSERT(deltaOffset >= 0);
//if (deltaOffset >= (INT32)object_size.LSB()) //if (deltaOffset >= (INT32)object_size.LSB())

6
src/common/normSegment.cpp
View File

@ -789,7 +789,7 @@ NormBlockId NormBlockBuffer::RangeMin() const
if (range_max > 1) if (range_max > 1)
{ {
NormBlockId rangeMin = range_hi; NormBlockId rangeMin = range_hi;
Decrement(rangeMin, range_max - 1); Decrement(rangeMin, (UINT32)range_max - 1);
return rangeMin; return rangeMin;
} }
else else
@ -841,7 +841,7 @@ bool NormBlockBuffer::Insert(NormBlock* theBlock)
// else if (blockId < range_lo) // else if (blockId < range_lo)
else if (Compare(blockId, range_lo) < 0) 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; if (newRange > range_max) return false;
range_lo = blockId; range_lo = blockId;
range = newRange; range = newRange;
@ -849,7 +849,7 @@ bool NormBlockBuffer::Insert(NormBlock* theBlock)
// else if (blockId > range_hi) // else if (blockId > range_hi)
else if (Compare(blockId, range_hi) > 0) 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; if (newRange > range_max) return false;
range_hi = blockId; range_hi = blockId;
range = newRange; range = newRange;

65
src/common/normSession.cpp
View File

@ -254,6 +254,8 @@ bool NormSession::Open()
if (!tx_socket->SetFragmentation(fragmentation)) if (!tx_socket->SetFragmentation(fragmentation))
PLOG(PL_WARN, "NormSession::Open() warning: tx_socket.SetFragmentation() error\n"); PLOG(PL_WARN, "NormSession::Open() warning: tx_socket.SetFragmentation() error\n");
TRACE("NormSession::Open() address %s\n", address.GetHostString());
if (address.IsMulticast()) if (address.IsMulticast())
{ {
if (!tx_socket->SetTTL(ttl)) if (!tx_socket->SetTTL(ttl))
@ -774,14 +776,14 @@ bool NormSession::StartSender(UINT16 instanceId,
if (numBlocks < 2) numBlocks = 2; if (numBlocks < 2) numBlocks = 2;
unsigned long numSegments = numBlocks * numParity; 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"); PLOG(PL_FATAL, "NormSession::StartSender() block_pool init error\n");
StopSender(); StopSender();
return false; 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"); PLOG(PL_FATAL, "NormSession::StartSender() segment_pool init error\n");
StopSender(); StopSender();
@ -2098,9 +2100,9 @@ bool NormSession::SetTxCacheBounds(NormObjectSize sizeMax,
{ {
bool result = true; bool result = true;
tx_cache_size_max = sizeMax; 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; 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; 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 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()) if (countMax != tx_table.GetRangeMax())
{ {
tx_table.SetRangeMax((UINT16)countMax); tx_table.SetRangeMax((UINT16)countMax);
result = tx_pending_mask.Resize(countMax); result = tx_pending_mask.Resize((UINT32)countMax);
result &= tx_repair_mask.Resize(countMax); result &= tx_repair_mask.Resize((UINT32)countMax);
if (!result) if (!result)
{ {
countMax = tx_pending_mask.GetSize(); countMax = tx_pending_mask.GetSize();
if (tx_repair_mask.GetSize() < countMax) if (tx_repair_mask.GetSize() < countMax)
countMax = tx_repair_mask.GetSize(); countMax = tx_repair_mask.GetSize();
if (tx_cache_count_max > countMax) 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) if (tx_cache_count_min > tx_cache_count_max)
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 else
{ {
TRACE("NormSession::TxSocketRecvHandler() RecvFrom error\n");
// Probably an ICMP "port unreachable" error // Probably an ICMP "port unreachable" error
// Note we purposefull do _not_ set the "posted_send_error" // Note we purposefull do _not_ set the "posted_send_error"
// status here because we do not want this notification // status here because we do not want this notification
@ -2359,12 +2362,16 @@ void NormSession::RxSocketRecvHandler(ProtoSocket& theSocket,
} }
else else
{ {
TRACE("NormSession::RxSocketRecvHandler() RecvFrom error address:%s (unicast:%d)\n", Address().GetHostString(), Address().IsUnicast());
// Probably an ICMP "port unreachable" error // Probably an ICMP "port unreachable" error
// Note we purposefull do _not_ set the "posted_send_error" // Note we purposefull do _not_ set the "posted_send_error"
// status here because we do not want this notification // status here because we do not want this notification
// cleared due to SEND_OK status since it's receiver driven // cleared due to SEND_OK status since it's receiver driven
if (Address().IsUnicast()) if (Address().IsUnicast())
{
TRACE("posting send error ...\n");
Notify(NormController::SEND_ERROR, NULL, NULL); Notify(NormController::SEND_ERROR, NULL, NULL);
}
break; break;
} }
} }
@ -3324,7 +3331,7 @@ void NormSession::SenderHandleCCFeedback(struct timeval currentTime,
ccFlags = NormCC::RTT; ccFlags = NormCC::RTT;
ccRtt = savedRtt; ccRtt = savedRtt;
ccLoss = savedLoss; ccLoss = savedLoss;
ccRate = savedRate, ccRate = savedRate;
ccSequence = savedSequence; ccSequence = savedSequence;
currentTime = savedTime; currentTime = savedTime;
} }
@ -4344,6 +4351,27 @@ void NormSession::SenderCancelCmd()
} }
} // end 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() bool NormSession::SenderQueueAppCmd()
{ {
if (0 == cmd_count) return false; if (0 == cmd_count) return false;
@ -4389,27 +4417,6 @@ bool NormSession::OnCmdTimeout(ProtoTimer& theTimer)
return true; return true;
} // end NormSession::OnCmdTimeout() } // 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) void NormSession::ActivateFlowControl(double delay, NormObjectId objectId, NormController::Event event)
{ {
flow_control_object = objectId; flow_control_object = objectId;