C:/knet/include/kNet/MessageConnection.h

Go to the documentation of this file.

/* Copyright 2010 Jukka Jylänki

   Licensed under the Apache License, Version 2.0 (the "License");
   you may not use this file except in compliance with the License.
   You may obtain a copy of the License at

       http://www.apache.org/licenses/LICENSE-2.0

   Unless required by applicable law or agreed to in writing, software
   distributed under the License is distributed on an "AS IS" BASIS,
   WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
   See the License for the specific language governing permissions and
   limitations under the License. */
#pragma once

#include <vector>
#include <map>
#include <utility>
#include <set>

#include "WaitFreeQueue.h"
#include "LockFreePoolAllocator.h"
#include "Lockable.h"
#include "Socket.h"
#include "IMessageHandler.h"
#include "BasicSerializedDataTypes.h"
#include "Datagram.h"
#include "FragmentedTransferManager.h"
#include "NetworkMessage.h"
#include "Event.h"
#include "DataSerializer.h"
#include "DataDeserializer.h"

#include "MaxHeap.h"
#include "Clock.h"
#include "PolledTimer.h"
#include "Thread.h"

namespace kNet
{

class MessageConnection;
class UDPMessageConnection;
class TCPMessageConnection;
class NetworkServer;
class Network;
class NetworkWorkerThread;
class FragmentedSendManager;

#ifdef WIN32
struct FragmentedSendManager::FragmentedTransfer;
#endif

struct ConnectionStatistics
{
   struct PingTrack
   {
      tick_t pingSentTick;  
      tick_t pingReplyTick; 
      unsigned long pingID;      
      bool replyReceived;        
   };
   std::vector<PingTrack> ping;

   struct TrafficTrack
   {
      tick_t tick;          
      unsigned long packetsIn;   
      unsigned long packetsOut;  
      unsigned long messagesIn;  
      unsigned long messagesOut; 
      unsigned long bytesIn;     
      unsigned long bytesOut;    
   };
   std::vector<TrafficTrack> traffic;

   struct DatagramIDTrack
   {
      tick_t tick;
      packet_id_t packetID;
   };
   std::vector<DatagramIDTrack> recvPacketIDs;
};

class NetworkMessagePriorityCmp
{
public:
   int operator ()(const NetworkMessage *a, const NetworkMessage *b)
   {
      assert(a && b);
      if (a->priority < b->priority) return -1;
      if (b->priority < a->priority) return 1;

      if (a->MessageNumber() < b->MessageNumber()) return 1;
      if (b->MessageNumber() < a->MessageNumber()) return -1;

      return 0;
   }
};

enum ConnectionState
{
   ConnectionPending, 
   ConnectionOK,      
   ConnectionDisconnecting, 
   ConnectionPeerClosed, 
   ConnectionClosed    
};

std::string ConnectionStateToString(ConnectionState state);

class MessageConnection : public RefCountable
{
public:
   virtual ~MessageConnection();

   ConnectionState GetConnectionState() const; // [main and worker thread]

   bool IsReadOpen() const; // [main and worker thread]
   
   bool IsWriteOpen() const; // [main and worker thread]

   bool IsPending() const; // [main and worker thread]

   bool Connected() const { return IsReadOpen() || IsWriteOpen(); } // [main and worker thread]

   void RunModalClient(); // [main thread]

   bool WaitToEstablishConnection(int maxMSecsToWait = 500); // [main thread]

   void Disconnect(int maxMSecsToWait = 500); // [main thread]

   void Close(int maxMSecsToWait = 500); // [main thread]

   // There are 3 ways to send messages through a MessageConnection:
   // StartNewMessage/EndAndQueueMessage, SendStruct, and Send. See below.

   NetworkMessage *StartNewMessage(unsigned long id, size_t numBytes = 0); // [main and worker thread]

   void EndAndQueueMessage(NetworkMessage *msg, size_t numBytes = (size_t)(-1), bool internalQueue = false); // [main and worker thread]

   void SendMessage(unsigned long id, bool reliable, bool inOrder, unsigned long priority, unsigned long contentID, 
                    const char *data, size_t numBytes); // [main thread]

   template<typename SerializableData>
   void SendStruct(const SerializableData &data, unsigned long id, bool inOrder, 
      bool reliable, unsigned long priority, unsigned long contentID = 0); // [main thread]

   template<typename SerializableMessage>
   void Send(const SerializableMessage &data, unsigned long contentID = 0); // [main thread]

   void PauseOutboundSends(); // [main thread]

   void ResumeOutboundSends(); // [main thread]

   size_t NumInboundMessagesPending() const { return inboundMessageQueue.Size(); } // [main and worker thread]

   size_t NumOutboundMessagesPending() const { return outboundQueue.Size() + outboundAcceptQueue.Size(); } // [main and worker thread]

   Socket *GetSocket() { return socket; }

   EndPoint LocalEndPoint() const; // [main and worker thread]

   EndPoint RemoteEndPoint() const; // [main and worker thread]

   void SetMaximumDataSendRate(int numBytesPerSec, int numDatagramsPerSec);

   void RegisterInboundMessageHandler(IMessageHandler *handler); // [main thread]

   void Process(int maxMessagesToProcess = 100); // [main thread]
   
   void WaitForMessage(int maxMSecsToWait); // [main thread]

   NetworkMessage *ReceiveMessage(int maxMSecsToWait = -1); // [main thread]

   void FreeMessage(NetworkMessage *msg); // [main and worker thread]
   
   std::string ToString() const; // [main and worker thread]

   void DumpStatus() const; // [main thread]

   // MessageConnection Statistics -related functions:

   float RoundTripTime() const { return rtt; } // [main and worker thread]

   float Latency() const { return rtt / 2.f; } // [main and worker thread]

   float LastHeardTime() const { return Clock::TicksToMillisecondsF(Clock::TicksInBetween(Clock::Tick(), lastHeardTime)); } // [main and worker thread]

   float PacketsInPerSec() const { return packetsInPerSec; } // [main and worker thread]
   float PacketsOutPerSec() const { return packetsOutPerSec; } // [main and worker thread]
   float MsgsInPerSec() const { return msgsInPerSec; } // [main and worker thread]
   float MsgsOutPerSec() const { return msgsOutPerSec; } // [main and worker thread]
   float BytesInPerSec() const { return bytesInPerSec; } // [main and worker thread]
   float BytesOutPerSec() const { return bytesOutPerSec; } // [main and worker thread]

   Lockable<ConnectionStatistics> statistics; // [main and worker thread]

protected:
   friend class NetworkWorkerThread;

   Network *owner; // [set and read only by the main thread]

   NetworkServer *ownerServer; // [set and read only by the main thread]

   NetworkWorkerThread *workerThread; // [set and read only by worker thread]

#ifdef _DEBUG

   ThreadId workerThreadId; // [set by worker thread on thread startup, read by both main and worker thread]

   bool InWorkerThreadContext() const; // [main and worker thread]

   bool InMainThreadContext() const; // [main and worker thread]

   bool IsWorkerThreadRunning() const { return workerThread != 0; } // [main and worker thread]
#endif

   WaitFreeQueue<NetworkMessage*> outboundAcceptQueue; // [produced by main thread, consumed by worker thread]

   WaitFreeQueue<NetworkMessage*> inboundMessageQueue; // [produced by worker thread, consumed by main thread]

// MaxHeap<NetworkMessage*, NetworkMessagePriorityCmp> outboundQueue;
   WaitFreeQueue<NetworkMessage*> outboundQueue; // [worker thread]

   Lockable<FragmentedSendManager> fragmentedSends; // [worker thread]

   FragmentedReceiveManager fragmentedReceives; // [worker thread]

   LockFreePoolAllocator<NetworkMessage> messagePool; // [main and worker thread]

   PolledTimer pingTimer; // [worker thread]

   PolledTimer statsRefreshTimer; // [worker thread]

   enum PacketSendResult
   {
      PacketSendOK, 
      PacketSendSocketClosed, 
      PacketSendSocketFull, 
      PacketSendNoMessages, 
      PacketSendThrottled   
   };

   virtual PacketSendResult SendOutPacket() = 0; // [worker thread]

   virtual void SendOutPackets() = 0; // [worker thread]

   virtual unsigned long TimeUntilCanSendPacket() const = 0; // [worker thread]

   void UpdateConnection(); // [worker thread]

   virtual void DoUpdateConnection() {} // [worker thread]

   void SetPeerClosed(); // [worker thread]

   virtual void DumpConnectionStatus() const {} // [main thread]

   Event NewOutboundMessagesEvent() const; // [main and worker thread]

   enum SocketReadResult
   {
      SocketReadOK,        
      SocketReadError,     
      SocketReadThrottled, 
   };

   virtual SocketReadResult ReadSocket(size_t &bytesRead) = 0; // [worker thread]

   SocketReadResult ReadSocket(); // [worker thread]

   void SetWorkerThread(NetworkWorkerThread *thread); // [worker thread]

   void HandleInboundMessage(packet_id_t packetID, const char *data, size_t numBytes); // [worker thread]

   NetworkMessage *AllocateNewMessage();

   // Ping/RTT management operations:
   void SendPingRequestMessage(); // [worker thread]

   void HandlePingRequestMessage(const char *data, size_t numBytes); // [worker thread]

   void HandlePingReplyMessage(const char *data, size_t numBytes); // [worker thread]

   // Frees all internal dynamically allocated message data.
   void FreeMessageData(); // [main thread]

   void DetectConnectionTimeOut(); // [worker thread]

   void ComputeStats(); // [worker thread]
   
   void AddOutboundStats(unsigned long numBytes, unsigned long numPackets, unsigned long numMessages); // [worker thread]

   void AddInboundStats(unsigned long numBytes, unsigned long numPackets, unsigned long numMessages); // [worker thread]

   void AcceptOutboundMessages(); // [worker thread]

   virtual void PerformDisconnection() = 0; 

   IMessageHandler *inboundMessageHandler; // [main thread]

   Socket *socket; // [set by main thread before the worker thread is running. Read-only when worker thread is running. Read by main and worker thread]

   ConnectionState connectionState; // [main and worker thread]

   bool bOutboundSendsPaused; // [set by main thread, read by worker thread]

   friend class NetworkServer;
   friend class Network;

   Event eventMsgsOutAvailable; // [main and worker thread]

   void operator=(const MessageConnection &); 
   MessageConnection(const MessageConnection &); 

   float rtt; 
   tick_t lastHeardTime; 
   float packetsInPerSec; 
   float packetsOutPerSec; 
   float msgsInPerSec; 
   float msgsOutPerSec; 
   float bytesInPerSec; 
   float bytesOutPerSec; 

   unsigned long outboundMessageNumberCounter; // [worker thread]

   unsigned long outboundReliableMessageNumberCounter; // [worker thread]

   typedef std::pair<u32, u32> MsgContentIDPair;

   typedef std::map<MsgContentIDPair, std::pair<packet_id_t, tick_t> > ContentIDReceiveTrack;

   ContentIDReceiveTrack inboundContentIDStamps; // [worker thread]

   typedef std::map<MsgContentIDPair, NetworkMessage*> ContentIDSendTrack;

   ContentIDSendTrack outboundContentIDMessages; // [worker thread]

   void CheckAndSaveOutboundMessageWithContentID(NetworkMessage *msg); // [worker thread]

   void ClearOutboundMessageWithContentID(NetworkMessage *msg); // [worker thread]

   bool CheckAndSaveContentIDStamp(u32 messageID, u32 contentID, packet_id_t packetID); // [worker thread]

   void SplitAndQueueMessage(NetworkMessage *message, bool internalQueue, size_t maxFragmentSize); // [main and worker thread]

   static const unsigned long MsgIdPingRequest = 1;
   static const unsigned long MsgIdPingReply = 2;
   static const unsigned long MsgIdFlowControlRequest = 3;
   static const unsigned long MsgIdPacketAck = 4;
   static const unsigned long MsgIdDisconnect = 0x3FFFFFFF;
   static const unsigned long MsgIdDisconnectAck = 0x3FFFFFFE;

   explicit MessageConnection(Network *owner, NetworkServer *ownerServer, Socket *socket, ConnectionState startingState);

   virtual void Initialize() {} // [main thread]

   virtual bool HandleMessage(packet_id_t /*packetID*/, u32 /*messageID*/, const char * /*data*/, size_t /*numBytes*/) { return false; } // [main thread]
};

template<typename SerializableData>
void MessageConnection::SendStruct(const SerializableData &data, unsigned long id, bool inOrder, 
      bool reliable, unsigned long priority, unsigned long contentID)
{
   assert(InMainThreadContext());

   const size_t dataSize = data.Size();

   NetworkMessage *msg = StartNewMessage(id, dataSize);

   if (dataSize > 0)
   {
      DataSerializer mb(msg->data, dataSize);
      data.SerializeTo(mb);
      assert(mb.BytesFilled() == dataSize); // The SerializableData::Size() estimate must be exact!
   }

   msg->id = id;
   msg->contentID = contentID;
   msg->inOrder = inOrder;
   msg->priority = priority;
   msg->reliable = reliable;

   EndAndQueueMessage(msg);
}

template<typename SerializableMessage>
void MessageConnection::Send(const SerializableMessage &data, unsigned long contentID)
{
   assert(InMainThreadContext());

   const size_t dataSize = data.Size();

   NetworkMessage *msg = StartNewMessage(data.MessageID(), dataSize);

   if (dataSize > 0)
   {
      DataSerializer mb(msg->data, dataSize);
      data.SerializeTo(mb);
      assert(mb.BytesFilled() == dataSize);
   }

   msg->id = data.MessageID();
   msg->contentID = contentID;
   msg->inOrder = data.inOrder;
   msg->priority = data.priority;
   msg->reliable = data.reliable;

   EndAndQueueMessage(msg);
}

} // ~kNet