pendingacklist.h

/***************************************************************************
 * The contents of this file are subject to the Mozilla Public             *
 * License Version 1.1 (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.mozilla.org/MPL/                              *
 *                                                                         *
 * Software distributed under the License is distributed on an "AS         *
 * IS" basis, WITHOUT WARRANTY OF ANY KIND, either express or              *
 * implied. See the License for the specific language governing            *
 * rights and limitations under the License.                               *
 *                                                                         *
 * The Original Code is Game Network Framework (GaNeF).                    *
 *                                                                         *
 * The Initial Developers of the Original Code are                         *
 * Lars Langer and Emanuel Greisen                                         *
 * Copyright (C) 2005. Lars Langer & Emanuel Greisen                       *
 * All Rights Reserved.                                                    *
 *                                                                         *
 * Contributor(s):                                                         *
 *   none yet....                                                          *
 *                                                                         *
 ***************************************************************************/
#ifndef PENDINGACKLIST_H
#define PENDINGACKLIST_H

#include <ptypes.h>
#include "oakbinheap.h"
#include "pendingack.h"
#include <set>

template<class PacketType, class SocketType>
class Framework;


/**
 * @ingroup Common
 * @brief This comparerer structure is used to sort the PendingAck in the heap.
*/
template<class PacketType, class SocketType>
struct PendingAckTimtoutCompare
{
   bool operator() (const PendingAck<PacketType, SocketType> &a1,const PendingAck<PacketType, SocketType> &a2)
   {
      return a1.timeout < a2.timeout;
   };
};

/**
 * @ingroup Common
 * @brief This struct can sort the PendingAck for easy retrieval of a certain one, when we get an Ack
*/
template<class PacketType, class SocketType>
struct PendingAckKeyCompare
{
   bool operator() (const PendingAck<PacketType, SocketType> * pendack1, const PendingAck<PacketType, SocketType> * pendack2)
  {
      if(pendack1->seq == pendack2->seq)
      {
         if(pendack1->pri == pendack2->pri)
         {
            return pendack1->user_id < pendack2->user_id;
         }
         else
         {
            return pendack1->pri < pendack2->pri;
         }
      }
      else
      {
         return (pendack1->seq < pendack2->seq);
      }
  }
};

/**
 * @ingroup Common
 * @brief This is the class keeping the list of pending acks.
 * It runs as its own thread, hence it can resend messages that are not Acknowledged in time.
 *
 * @author Lars Langer & Emanuel Greisen
*/
template<class PacketType,class SocketType>
class PendingAckList : public thread, public mutex
{
private:
   typedef typename std::set<PendingAck<PacketType, SocketType> *, PendingAckKeyCompare<PacketType, SocketType> >::iterator AckMapIt;
private:
   Framework<PacketType, SocketType> * framework;
   unsigned int minRTT;
   double RTT_multiplier;

protected:
   OakBinaryMinHeap<PendingAck<PacketType, SocketType>, PendingAckTimtoutCompare<PacketType, SocketType> > timeoutheap;
   std::set<PendingAck<PacketType, SocketType> *, PendingAckKeyCompare<PacketType, SocketType> > ackmap;
private:
   void setPendingAckAcked(PendingAck<PacketType, SocketType> * penack);

public:
   PendingAckList(Framework<PacketType, SocketType> * framework);
   ~PendingAckList();

public:
   void addPendingAck(unsigned int packet_seq, unsigned short packet_pri, unsigned int user_id, unsigned int timeout, PacketType * pack, bool haslock);
   /// This method will remove a pending ack, and return the RTT for this ack.
   unsigned int registerAck(unsigned int packet_seq, unsigned char packet_pri, unsigned int user_id);
   /// The run-method of this thread.
   void execute();
   /// Returns the number of pending acks.
   inline unsigned int size() const { return ackmap.size(); };
   /// Will set all pending-acks that whose userid matches to state cancelled
   template<typename Comp>
   void removeWithComparer( const Comp & comp );
   /// Will simply clear the map.
   void clear();
};


template<class PacketType, class SocketType>
PendingAckList<PacketType, SocketType>::PendingAckList( Framework<PacketType, SocketType> * framework ) :
framework(framework),
minRTT(60),
RTT_multiplier(1.25),
thread(false)
{
   ackmap.clear();
}

template<class PacketType, class SocketType>
PendingAckList<PacketType, SocketType>::~PendingAckList()
{

}

template<class PacketType, class SocketType>
template<typename Comp>
void PendingAckList<PacketType, SocketType>::removeWithComparer( const Comp & comp )
{
   enter();
   for(AckMapIt it = ackmap.begin(); it != ackmap.end(); ++it)
   {
      PendingAck<PacketType, SocketType> * penack = *it;
      if(comp(penack->user_id) && !penack->acked)
      {
         std::cout << "Erasing pending Ack: " << penack->packet << std::endl;
         setPendingAckAcked( penack );
      }
      else
      {
         //std::cout << "Skipping pending Ack: " << penack->user_id << std::endl;
      }
   }
   leave();
}

template<class PacketType, class SocketType>
void PendingAckList<PacketType, SocketType>::clear( )
{
   enter();
   for(AckMapIt it = ackmap.begin(); it != ackmap.end(); ++it)
   {
      PendingAck<PacketType, SocketType> * penack = *it;
      setPendingAckAcked( penack );
   }
   leave();
}


template<class PacketType, class SocketType>
void PendingAckList<PacketType, SocketType>::addPendingAck(unsigned int packet_seq, unsigned short packet_pri, unsigned int user_id, unsigned int timeout, PacketType * pack, bool haslock)
{
   // Insert data in to a map with unique key so we can ack it
   PendingAck<PacketType, SocketType> key(packet_seq, packet_pri, user_id, 0, 0);
   key.sent_time = Framework<PacketType, SocketType>::currentTimeMillis();
   timeout = (unsigned int)(timeout * RTT_multiplier);

   if(!haslock)
      enter();
   // Replace old - if any exist
   if(ackmap.find(&key) != ackmap.end())
   {
      PendingAck<PacketType, SocketType> * penAck = *ackmap.find(&key);
      //std::cout << "Rep PendingAck:("<<timeoutheap.size()<<"/"<<ackmap.size()<<")["<<packet_seq<<","<<packet_pri<<","<<user_id<<"]";
      //std::cout << "["<<penAck->seq<<","<<penAck->pri<<","<<penAck->user_id<<"]\n";
      penAck->timeout = timeout + key.sent_time;
      timeoutheap.heapify(penAck); // Make sure it finds it way down
   }
   else
   {
      // Inserting the packet in the minheap to be able to spot timeouts
      //std::cout << "New PendingAck:("<<timeoutheap.size()<<"/"<<ackmap.size()<<")["<<packet_seq<<","<<packet_pri<<","<<user_id<<"]" << std::endl;

      PendingAck<PacketType, SocketType> * penAck = new PendingAck<PacketType, SocketType>(packet_seq, packet_pri, user_id, (timeout+key.sent_time), pack);
      penAck->sent_time = key.sent_time;
      timeoutheap.insert(penAck);
      ackmap.insert(penAck);
   }
   // And now insert in the ackmap

   if(!haslock)
      leave();
}

template<class PacketType, class SocketType>
unsigned int PendingAckList<PacketType, SocketType>::registerAck( unsigned int seq, unsigned char pri, unsigned int user_id )
{
   unsigned int rtt_for_this_ack = std::numeric_limits<unsigned int>::max();
   //std::cout << "registerAck( " << seq << ", " << pri << ", " << user_id << " )\n";
   enter();
   PendingAck<PacketType, SocketType> key(seq, pri, user_id, 0, 0);
   if(ackmap.find(&key) == ackmap.end())
   {
      //std::cout << "No! PendingAck<PacketType, SocketType>:("<<timeoutheap.size()<<"/"<<ackmap.size()<<")["<<seq<<","<<pri<<","<<user_id<<"]\n";
   }
   else
   {
      PendingAck<PacketType, SocketType> * penAck = *ackmap.find(&key);
      //std::cout << "Ack PendingAck<PacketType, SocketType>:("<<timeoutheap.size()<<"/"<<ackmap.size()<<")["<<seq<<","<<pri<<","<<user_id<<"]:";
      //std::cout << "["<<penAck->seq<<","<<penAck->pri<<","<<penAck->user_id<<"]\n";
      if(!penAck->acked)
      {
         // The pending ack is there so we get it
         rtt_for_this_ack = Framework<PacketType, SocketType>::currentTimeMillis() - penAck->sent_time;
         /*
         if(penAck->seq != seq || penAck->pri != pri || penAck->user_id != user_id)
         {
         std::cout << "ERROR:\n";
         std::cout << "GOT PendingAck<PacketType, SocketType>:("<<timeoutheap.size()<<"/"<<ackmap.size()<<")["<<seq<<","<<pri<<","<<user_id<<"]\n";
         std::cout << "HAD PendingAck<PacketType, SocketType>:("<<timeoutheap.size()<<"/"<<ackmap.size()<<")["<<penAck->seq<<","<<penAck->pri<<","<<penAck->user_id<<"]\n";
      }
         */
         setPendingAckAcked(penAck);
      }
      //std::cout << "Ack PendingAck:("<<timeoutheap.size()<<"/"<<ackmap.size()<<")["<<seq<<","<<pri<<","<<user_id<<"]:";
      //std::cout << "["<<penAck->seq<<","<<penAck->pri<<","<<penAck->user_id<<"]\n";
   }
   leave();
   return rtt_for_this_ack;
}


template<class PacketType, class SocketType>
void PendingAckList<PacketType, SocketType>::execute( )
{
   while(true)
   {
      PendingAck<PacketType, SocketType> * tmpAck;

      // There has to be at least one element due to the semaphore wait()
      timeoutheap.peek();
      // Now that we know there is one, lets enter the Mutex and handle it here.
      enter();
      tmpAck = timeoutheap.peek();
      if( tmpAck->acked )
      {
         // Its been acked, remove it from both list and heap
         //std::cout << "Rem PendingAck:("<<timeoutheap.size()<<"/"<<ackmap.size()<<")["<<tmpAck->seq<<","<<tmpAck->pri<<","<<tmpAck->user_id<<"]\n";

         timeoutheap.getMin();
         ackmap.erase(tmpAck);
         delete tmpAck;
      }
      else if(tmpAck->timeout < Framework<PacketType, SocketType>::currentTimeMillis())
      {
         // Make sure it stays down in the heap
         tmpAck->timeout = Framework<PacketType, SocketType>::currentTimeMillis() + (unsigned int)(minRTT * RTT_multiplier);
         timeoutheap.heapify(tmpAck); // Move the element in the heap

         // Schedule for Retransmission (if it's already queued the queue will discard the queing of this pack):
         framework->retransmitPacket(tmpAck->packet);
      }
      else
      {
         leave();
         unsigned int time_to_sleep = std::min((unsigned int)(minRTT * RTT_multiplier), tmpAck->timeout - Framework<PacketType, SocketType>::currentTimeMillis());
         //std::cout << "PendingAckList.sleep(" << time_to_sleep << ")\n";
         thread::relax(time_to_sleep);
         continue;
      }
      leave();
   }
}

template<class PacketType, class SocketType>
void PendingAckList<PacketType, SocketType>::setPendingAckAcked( PendingAck<PacketType, SocketType> * penAck )
{
   //std::cout << "PendingAck on Pack:" << penAck->packet << std::endl;
   penAck->acked = true; // execute() now know that it can remove the ack from the heap
   penAck->timeout = 0; // Make sure it goes to the top of the heap
   timeoutheap.heapify(penAck);
   if(penAck->packet->hasBeenSent())
   {
      delete penAck->packet; // Since it't not in the queue, we can delete it
      penAck->packet = 0;
   }
   else
   {
      penAck->packet->setStateAborted(); // We must wait untill the sender deletes it.
   }
}






#endif

---