G3D::ReliableConduit Class Reference

#include <NetworkDevice.h>

Public Member Functions
	~ReliableConduit ()
virtual bool	messageWaiting ()
template<typename T >
void	send (uint32 type, const T &message)
void	send (uint32 type)
virtual uint32	waitingMessageType ()
template<typename T >
bool	receive (T &message)
void	receive ()
NetAddress	address () const
Public Member Functions inherited from G3D::Conduit
virtual	~Conduit ()
uint64	bytesSent () const
uint64	messagesSent () const
uint64	bytesReceived () const
uint64	messagesReceived () const
bool	ok () const
Public Member Functions inherited from G3D::ReferenceCountedObject
virtual	~ReferenceCountedObject ()

Static Public Member Functions
static ReliableConduitRef	create (const NetAddress &address)
template<typename T >
static void	multisend (const Array< ReliableConduitRef > &array, uint32 type, const T &m)

Private Types
enum	State { RECEIVING, HOLDING, NO_MESSAGE }

Private Member Functions
	ReliableConduit (const NetAddress &addr)
	ReliableConduit (const SOCKET &sock, const NetAddress &addr)
void	sendBuffer (const BinaryOutput &b)
void	receiveIntoBuffer ()
void	receiveHeader ()

Static Private Member Functions
template<typename T >
static void	serializeMessage (uint32 t, const T &m, BinaryOutput &b)

Private Attributes
enum G3D::ReliableConduit::State	state
NetAddress	addr
uint32	messageType
uint32	messageSize
void *	receiveBuffer
size_t	receiveBufferTotalSize
size_t	receiveBufferUsedSize

Friends
class	NetworkDevice
class	NetListener

Additional Inherited Members
Protected Member Functions inherited from G3D::Conduit
	Conduit ()
Protected Attributes inherited from G3D::Conduit
uint64	mSent
uint64	mReceived
uint64	bSent
uint64	bReceived
SOCKET	sock
BinaryOutput	binaryOutput

Detailed Description

A conduit that guarantees messages will arrive, intact and in order. Create on the client using NetworkDevice::createReliableConduit and on the server using NetListener::waitForConnection. Set the reference counted pointer to NULL to disconnect.

To construct a ReliableConduit:

1. Create a G3D::NetworkDevice (if you are using G3D::GApp, it creates one for you) on the client and on the server.
2. On the server, create a G3D::NetListener using G3D::NetworkDevice::createListener
3. On the server, invoke G3D::NetListener::waitForConnection.
4. On the client, call G3D::NetworkDevice::createReliableConduit. You will need the server's G3D::NetAddress. Consider using G3D::Discovery::Client to find it via broadcasting.

Deprecated:

Member Enumeration Documentation

enum G3D::ReliableConduit::State

private

Enumerator
RECEIVING
HOLDING
NO_MESSAGE

{RECEIVING, HOLDING, NO_MESSAGE} state;

Constructor & Destructor Documentation

G3D::ReliableConduit::ReliableConduit ( const NetAddress &  addr )

private

 : 
 state(NO_MESSAGE), 
 receiveBuffer(NULL),
 receiveBufferTotalSize(0), 
 receiveBufferUsedSize(0) {

 NetworkDevice* nd = NetworkDevice::instance();
 
 messageType = 0;

 addr = _addr;
 Log::common()->print("Creating a TCP socket ");
 sock = socket(AF_INET, SOCK_STREAM, IPPROTO_IP);
 
 if (sock == SOCKET_ERROR) {
 Log::common()->println("FAIL");
 Log::common()->println(socketErrorCode());
 nd->closesocket(sock);
 return;
 }

 Log::common()->println("Ok");

 // Setup socket options (both constructors should set the same options)

 // Disable Nagle's algorithm (we send lots of small packets)
 const int T = true;
 if (setsockopt(sock, IPPROTO_TCP, TCP_NODELAY, 
 (const char*)&T, sizeof(T)) == SOCKET_ERROR) {
 
 Log::common()->println("WARNING: Disabling Nagel's "
 "algorithm failed.");
 Log::common()->println(socketErrorCode());
 } else {
 Log::common()->println("Disabled Nagel's algorithm.");
 }

 // Set the NO LINGER option so the socket doesn't hang around if
 // there is unsent data in the queue when it closes.
 struct linger ling;
 ling.l_onoff = 0;
 ling.l_linger = 0;
 if (setsockopt(sock, SOL_SOCKET, SO_LINGER, 
 (const char*)&ling, sizeof(ling)) == SOCKET_ERROR) {
 
 Log::common()->println("WARNING: Setting socket no linger failed.");
 Log::common()->println(socketErrorCode());
 } else {
 Log::common()->println("Set socket option no_linger.");
 }

 // Set reuse address so that a new server can start up soon after
 // an old one has closed.
 if (setsockopt(sock, SOL_SOCKET, SO_REUSEADDR, 
 (const char*)&T, sizeof(T)) == SOCKET_ERROR) {
 
 Log::common()->println("WARNING: Setting socket reuseaddr failed.");
 Log::common()->println(socketErrorCode());
 } else {
 Log::common()->println("Set socket option reuseaddr.");
 }

 // Ideally, we'd like to specify IPTOS_LOWDELAY as well.

 logSocketInfo(sock);

 increaseBufferSize(sock);

 Log::common()->printf("Created TCP socket %d\n", sock);

 std::string x = addr.toString();
 Log::common()->printf("Connecting to %s on TCP socket %d ", x.c_str(), sock);

 int ret = connect(sock, (struct sockaddr *) &(addr.addr), sizeof(addr.addr));

 if (ret == WSAEWOULDBLOCK) {
 RealTime t = System::time() + 5.0;
 // Non-blocking; we must wait until select returns non-zero
 while ((selectOneWriteSocket(sock) == 0) && (System::time() < t)) {
 System::sleep(0.02);
 }

 // TODO: check for failure on the select call

 } else if (ret != 0) {
 sock = (SOCKET)SOCKET_ERROR;
 Log::common()->println("FAIL");
 Log::common()->println(socketErrorCode());
 return;
 }

 Log::common()->println("Ok");
}

Here is the call graph for this function:

Here is the caller graph for this function:

G3D::ReliableConduit::ReliableConduit ( const SOCKET &  sock, const NetAddress &  addr  )

private

 : 
 state(NO_MESSAGE), 
 receiveBuffer(NULL), 
 receiveBufferTotalSize(0), 
 receiveBufferUsedSize(0) {
 sock = _sock;
 addr = _addr;

 messageType = 0;

 // Setup socket options (both constructors should set the same options)

 // Disable Nagle's algorithm (we send lots of small packets)
 const int T = true;
 if (setsockopt(sock, IPPROTO_TCP, TCP_NODELAY, 
 (const char*)&T, sizeof(T)) == SOCKET_ERROR) {
 
 Log::common()->println("WARNING: Disabling Nagel's algorithm failed.");
 Log::common()->println(socketErrorCode());
 } else {
 Log::common()->println("Disabled Nagel's algorithm.");
 }

 // Set the NO LINGER option so the socket doesn't hang around if
 // there is unsent data in the queue when it closes.
 struct linger ling;
 ling.l_onoff = 0;
 ling.l_linger = 0;
 if (setsockopt(sock, SOL_SOCKET, SO_LINGER, 
 (const char*)&ling, sizeof(ling)) == SOCKET_ERROR) {
 
 Log::common()->println("WARNING: Setting socket no linger failed.");
 Log::common()->println(socketErrorCode());
 } else {
 Log::common()->println("Set socket option no_linger.");
 }

 // Set reuse address so that a new server can start up soon after
 // an old one has closed.
 if (setsockopt(sock, SOL_SOCKET, SO_REUSEADDR, 
 (const char*)&T, sizeof(T)) == SOCKET_ERROR) {
 
 Log::common()->println("WARNING: Setting socket reuseaddr failed.");
 Log::common()->println(socketErrorCode());
 } else {
 Log::common()->println("Set socket option reuseaddr.");
 }

 // Ideally, we'd like to specify IPTOS_LOWDELAY as well.

 logSocketInfo(sock);
}

Here is the call graph for this function:

G3D::ReliableConduit::~ReliableConduit

(

)

Closes the socket.

 {
 free(receiveBuffer);
 receiveBuffer = NULL;
 receiveBufferTotalSize = 0;
 receiveBufferUsedSize = 0;
}

Member Function Documentation

NetAddress G3D::ReliableConduit::address

(

)

const

The address of the other end of the conduit

 {
 return addr;
}

ReliableConduitRef G3D::ReliableConduit::create ( const NetAddress &  address )

static

Client invokes this to connect to a server. The call blocks until the conduit is opened. The conduit will not be ok() if it fails.

 {
 return ReliableConduitRef(new ReliableConduit(address));
}

Here is the call graph for this function:

bool G3D::ReliableConduit::messageWaiting ( )

virtual

The message is actually copied from the socket to an internal buffer during this call. Receive only deserializes.

Reimplemented from G3D::Conduit.

 {
 switch (state) {
 case HOLDING:
 // We've already read the message and are waiting
 // for a receive call.
 return true;

 case RECEIVING:

 if (! ok()) {
 return false;
 }
 // We're currently receiving the message. Read a little more.
 receiveIntoBuffer();
 
 if (messageSize == receiveBufferUsedSize) {
 // We've read the whole mesage. Switch to holding state 
 // and return true.
 state = HOLDING;
 return true;
 } else {
 // There are more bytes left to read. We'll read them on
 // the next call. Because the *entire* message is not ready,
 // return false.
 return false;
 }
 break;

 case NO_MESSAGE:
 if (Conduit::messageWaiting()) {
 // Message incoming. Read the header.

 state = RECEIVING;
 receiveHeader();
 
 // Loop back around now that we're in the receive state; we
 // may be able to read the whole message before returning 
 // to the caller.
 return messageWaiting();
 } else {
 // No message incoming.
 return false;
 }
 }

 debugAssertM(false, "Should not reach this point");
 return false;
}

Here is the call graph for this function:

Here is the caller graph for this function:

template<typename T >

static void G3D::ReliableConduit::multisend ( const Array< ReliableConduitRef > &  array, uint32  type, const T &  m  )

inlinestatic

Send the same message to a number of conduits. Useful for sending data from a server to many clients (only serializes once).

 {
 
 if (array.size() > 0) {
 array[0]->binaryOutput.reset();
 serializeMessage(type, m, array[0]->binaryOutput);

 for (int i = 0; i < array.size(); ++i) {
 array[i]->sendBuffer(array[0]->binaryOutput);
 }
 }
 }

Here is the call graph for this function:

template<typename T >

bool G3D::ReliableConduit::receive ( T &  message )

inline

If a message is waiting, deserializes the waiting message into message and returns true, otherwise returns false. You can determine the type of the message (and therefore, the class of message) using G3D::ReliableConduit::waitingMessageType().

 {
 if (! messageWaiting()) {
 return false;
 }

 debugAssert(state == HOLDING);
 // Deserialize
 BinaryInput b((uint8*)receiveBuffer, receiveBufferUsedSize, G3D_LITTLE_ENDIAN, BinaryInput::NO_COPY);
 message.deserialize(b);
 
 // Don't let anyone read this message again. We leave the buffer
 // allocated for the next caller, however.
 receiveBufferUsedSize = 0;
 state = NO_MESSAGE;
 messageType = 0;
 messageSize = 0;

 // Potentially read the next message.
 messageWaiting();

 return true;
 }

Here is the call graph for this function:

void G3D::ReliableConduit::receive ( )

inline

Removes the current message from the queue.

 {
 if (! messageWaiting()) {
 return;
 }
 receiveBufferUsedSize = 0;
 state = NO_MESSAGE;
 messageType = 0;
 messageSize = 0;

 // Potentially read the next message.
 messageWaiting();
 }

Here is the call graph for this function:

void G3D::ReliableConduit::receiveHeader ( )

private

Receives the messageType and messageSize from the socket.

 {
 NetworkDevice* nd = NetworkDevice::instance();
 debugAssert(state == RECEIVING);

 // Read the type
 uint32 tmp;
 int ret = recv(sock, (char*)&tmp, sizeof(tmp), 0);

 // The type is the first four bytes. It is little endian.
 if (System::machineEndian() == G3D_LITTLE_ENDIAN) {
 messageType = tmp;
 } else {
 // Swap the byte order
 for (int i = 0; i < 4; ++i) {
 ((char*)&messageType)[i] = ((char*)&tmp)[3 - i];
 }
 }

 if ((ret == SOCKET_ERROR) || (ret != sizeof(messageType))) {
 Log::common()->printf("Call to recv failed. ret = %d,"
 " sizeof(messageType) = %d\n", 
 (int)ret, (int)sizeof(messageType));
 Log::common()->println(socketErrorCode());
 nd->closesocket(sock);
 messageType = 0;
 return;
 }

 // Read the size
 ret = recv(sock, (char*)&messageSize, sizeof(messageSize), 0);

 if ((ret == SOCKET_ERROR) || (ret != sizeof(messageSize))) {
 Log::common()->printf("Call to recv failed. ret = %d,"
 " sizeof(len) = %d\n", (int)ret,
 (int)sizeof(messageSize));
 Log::common()->println(socketErrorCode());
 nd->closesocket(sock);
 messageType = 0;
 return;
 }

 messageSize = ntohl(messageSize);
 debugAssert(messageSize < 6e7);

 debugAssert(receiveBufferUsedSize == 0);

 // Extend the size of the buffer.
 if (messageSize > receiveBufferTotalSize) {
 receiveBuffer = realloc(receiveBuffer, messageSize);
 receiveBufferTotalSize = messageSize;
 }

 if (receiveBuffer == NULL) {
 Log::common()->println("Could not allocate a memory buffer "
 "during receivePacket.");
 nd->closesocket(sock);
 }

 bReceived += 4;
}

Here is the call graph for this function:

Here is the caller graph for this function:

void G3D::ReliableConduit::receiveIntoBuffer ( )

private

Accumulates whatever part of the message (not the header) is still waiting on the socket into the receiveBuffer during state = RECEIVING mode. Closes the socket if anything goes wrong. When receiveBufferUsedSize == messageSize, the entire message has arrived.

 {
 NetworkDevice* nd = NetworkDevice::instance();

 debugAssert(state == RECEIVING);
 debugAssert(messageType != 0);
 debugAssertM(receiveBufferUsedSize < messageSize, "Message already received.");
 debugAssertM(messageSize >= receiveBufferUsedSize, "Message size overflow.");

 // Read the data itself
 int ret = 0;
 uint32 left = messageSize - (uint32)receiveBufferUsedSize;
 int count = 0;
 while ((ret != SOCKET_ERROR) && (left > 0) && (count < 100)) {

 ret = recv(sock, ((char*)receiveBuffer) + (uint32)receiveBufferUsedSize, left, 0);

 if (ret > 0) {
 left -= ret;
 receiveBufferUsedSize += ret;
 bReceived += ret;

 if (left > 0) {
 // There's still more. Give the machine a chance to read
 // more data, but don't wait forever.

 ++count;
 System::sleep(0.001);
 }
 } else {
 // Something went wrong; our blocking read returned nothing.
 break;
 }
 }

 if ((ret == 0) || (ret == SOCKET_ERROR)) {

 if (ret == SOCKET_ERROR) {
 Log::common()->printf("Call to recv failed. ret = %d,"
 " sizeof(messageSize) = %d\n", ret, messageSize);
 Log::common()->println(socketErrorCode());
 } else {
 Log::common()->printf("recv returned 0\n");
 }
 nd->closesocket(sock);
 return;
 }

 ++mReceived;
}

Here is the call graph for this function:

Here is the caller graph for this function:

template<typename T >

void G3D::ReliableConduit::send ( uint32  type, const T &  message  )

inline

Serializes the message and schedules it to be sent as soon as possible, and then returns immediately. The message can be any class with a serialize and deserialize method. On the receiving side, use G3D::ReliableConduit::waitingMessageType() to detect the incoming message and then invoke G3D::ReliableConduit::receive(msg) where msg is of the same class as the message that was sent.

The actual data sent across the network is preceeded by the message type and the size of the serialized message as a 32-bit integer. The size is sent because TCP is a stream protocol and doesn't have a concept of discrete messages.

 {
 binaryOutput.reset();
 serializeMessage(type, message, binaryOutput);
 sendBuffer(binaryOutput);
 }

Here is the call graph for this function:

Here is the caller graph for this function:

void G3D::ReliableConduit::send

(

uint32

type

)

Sends an empty message with the given type. Useful for sending commands that have no parameters.

 {
 static Dummy dummy;
 send(type, dummy);
}

Here is the call graph for this function:

void G3D::ReliableConduit::sendBuffer ( const BinaryOutput &  b )

private

 {
 NetworkDevice* nd = NetworkDevice::instance();
 int ret = ::send(sock, (const char*)b.getCArray(), (int)b.size(), 0);
 
 if (ret == SOCKET_ERROR) {
 Log::common()->println("Error occured while sending message.");
 Log::common()->println(socketErrorCode());
 nd->closesocket(sock);
 return;
 }

 ++mSent;
 bSent += b.size();

 // Verify the packet was actually sent
 // Conversion to unsigned is safe because -1 is caught earlier
 debugAssert(ret == b.size());
}

Here is the call graph for this function:

Here is the caller graph for this function:

template<typename T >

static void G3D::ReliableConduit::serializeMessage ( uint32  t, const T &  m, BinaryOutput &  b  )

inlinestaticprivate

 {

 b.writeUInt32(t);

 // Reserve space for the 4 byte size header
 b.writeUInt32(0);

 size_t L = (size_t)b.length();
 m.serialize(b);
 if ((size_t)b.length() == L) {
 // No data was created by serialization.
 // We need to send at least one byte because receive assumes that
 // a zero length message is an error.
 b.writeUInt8(0xFF);
 }
 
 uint32 len = (uint32)b.size() - 8;
 
 // We send the length first to tell recv how much data to read.
 // Here we abuse BinaryOutput a bit and write directly into
 // its buffer, violating the abstraction.
 // Note that we write to the second set of 4 bytes, which is
 // the size field.
 uint32* lenPtr = ((uint32*)b.getCArray()) + 1;
 #if defined(__GNUC__)
 *lenPtr = gcchtonl(len);
 #else
 *lenPtr = htonl(len);
 #endif
 }

Here is the call graph for this function:

Here is the caller graph for this function:

uint32 G3D::ReliableConduit::waitingMessageType ( )

virtual

Returns the type of the waiting message (i.e. the type supplied with send). The return value is zero when there is no message waiting.

One way to use this is to have a Table mapping message types to pre-allocated subclasses so receiving looks like:

    // My base class for messages.
    class Message {
        virtual void serialize(BinaryOutput&) const;
        virtual void deserialize(BinaryInput&);
        virtual void process() = 0;
    };

    Message* m = table[conduit->waitingMessageType()];
    conduit->receive(m);
    m->process();

Another is to simply switch on the message type:

    switch (conduit->waitingMessageType()) {
    case 0:
       // No message
       break;

    case ENTITY_SPAWN_MSG:
       {
          EntitySpawnMsg m;
          condiut->receive(m);
          spawnEntity(m.id, m.position, m.modelID);
       }
       break;
       ...
    }
 

Implements G3D::Conduit.

 {
 // The messageWaiting call is what actually receives the message.
 if (messageWaiting()) {
 return messageType;
 } else {
 return 0;
 }
}

Here is the call graph for this function:

Friends And Related Function Documentation

friend class NetListener

friend

friend class NetworkDevice

friend

Member Data Documentation

NetAddress G3D::ReliableConduit::addr

private

uint32 G3D::ReliableConduit::messageSize

private

Total size of the incoming message (read from the header).

uint32 G3D::ReliableConduit::messageType

private

Type of the incoming message.

void* G3D::ReliableConduit::receiveBuffer

private

Shared buffer for receiving messages.

size_t G3D::ReliableConduit::receiveBufferTotalSize

private

Total size of the receiveBuffer.

size_t G3D::ReliableConduit::receiveBufferUsedSize

private

Size occupied by the current message... so far. This will be equal to messageSize when the whole message has arrived.

enum G3D::ReliableConduit::State G3D::ReliableConduit::state

private

---

The documentation for this class was generated from the following files:

• dep/g3dlite/include/G3D/NetworkDevice.h
• dep/g3dlite/source/NetworkDevice.cpp