PongCache.cpp

Go to the documentation of this file.

/*

Copyright (C) 2005-2007 by Peter Dimov.

This file is part of Calitko (http://www.calitko.org).

Calitko is free software; you can redistribute it and/or modify
it under the terms of the GNU General Public License as published by
the Free Software Foundation; either version 2 of the License, or
(at your option) any later version.

Calitko is distributed in the hope that it will be useful,
but WITHOUT ANY WARRANTY; without even the implied warranty of
MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
GNU General Public License for more details.

You should have received a copy of the GNU General Public License
along with Calitko; if not, write to the Free Software
Foundation, Inc., 51 Franklin St, Fifth Floor, Boston, MA  02110-1301  USA

*/

#include "Qt.h"
#include "PongCache.h"
#include "PacketProcessor.h"
#include "Gnutella/LocalPeer.h"
#include "Gnutella/Packets/Extensions/Ggeps/DailyUptime.h"
#include "Gnutella/Packets/Extensions/Ggeps/Ultrapeer.h"
#include "Gnutella/Packets/Extensions/Ggeps/VendorCode.h"
#include "Gnutella/Packets/Extensions/Ggeps/UdpHostCache.h"
#include "Gnutella/Packets/Extensions/Ggeps/PackedHostCaches.h"
#include "Gnutella/Packets/Extensions/Ggeps/SupportsCachedPongs.h"
#include "Gnutella/Handshaking/SlotAllocator.h"
#include "Utils/Version.h"
#include <algorithm>

using Gnutella::PacketProcessing::PongCache;
using Gnutella::PacketProcessing::PacketProcessor;
using Gnutella::PacketProcessing::PacketSession;
using Gnutella::Packets::Packet;
using Gnutella::Packets::Ping;
using Gnutella::Packets::Pong;
using Gnutella::Packets::PingDescriptor;
using Gnutella::Packets::PongDescriptor;
using Gnutella::Bootstrapping::NodeCache;
using Gnutella::LocalPeer;

PongCache::SessionData::SessionData()
 :  directPong(), neighborPong(),
    lastInPingTime (QTime::currentTime().addMSecs (-MinimalPingInterval)),
    lastOutPingTime (QTime::currentTime().addMSecs (-MinimalPingInterval))
{
}

PongCache::PongCache(PacketProcessor *pp, LocalPeer *lp)
 :  localPeer (lp), packetProcessor (pp), slotAllocator (lp->slotAllocator()),
    directPing(), directPong(), hostPong(), pingTimer(), sessions()
{
    // \todo Now we have the class PingHnadler, which creates a response Pong!
    // use it instead this local implementation. Connect the pingReceived signal
    // with processPing. Let only one ping every say 10 sec flow to the handler,
    // cache the resulting pongs and answer with them!
    hostPong.setTtl (3);
    hostPong.setSharedKilobytes (1 << 21); // Square of two means we are an ultrapeer.
    hostPong.setSharedFiles (1<<10);

    using Gnutella::Packets::Extensions::GgepBlock;
    using Gnutella::Packets::Extensions::Ggeps::VendorCode;
    using Gnutella::Packets::Extensions::Ggeps::Ultrapeer;
    using Gnutella::Packets::Extensions::Ggeps::DailyUptime;
    using Utils::Version;

    GgepBlock ggepBlock;
    Ultrapeer ultrapeer (true, 0, 0); // Don't advertise free slots now (do so after 0.6.0)

    Packets::VendorCode myVendorCode ("LIME");
    Version myVersion (4, 6);
    VendorCode vendorCode (myVendorCode, myVersion);

    DailyUptime dailyUptime (34345);

    ggepBlock.addExtension (ultrapeer);
    ggepBlock.addExtension (vendorCode);
    ggepBlock.addExtension (dailyUptime);
    hostPong.setGgepBlock (ggepBlock);

    directPing.setTtl (1);
    directPing.setHops (0);

    // \todo Just to test the UHC, remove later:
    /*{
    using Gnutella::Packets::Extensions::Ggeps::SupportsCachedPongs;
    using Gnutella::Packets::Extensions::GgepBlock;
    SupportsCachedPongs scp (SupportsCachedPongs::UltrapeerPreference);

    GgepBlock ggepBlock;
    ggepBlock.addExtension (scp);
    directPing.setGgepBlock (ggepBlock);
    }*/
    // \todo remove the above

    directPong.setTtl (1);
    directPong.setHops (0);

    //hostHasFreeSlots = true;

    connect (&pingTimer, SIGNAL (timeout()), this, SLOT (refreshCache()));
    pingTimer.setInterval (PongExpirationTime);
    pingTimer.start();
}

PongCache::~PongCache()
{
}

void PongCache::addSession (PacketSession *session)
{
    Q_ASSERT (!sessions.contains (session));
    SessionData *sessionData = new SessionData();
    sessions.insert (session, sessionData);
}

void PongCache::removeSession (PacketSession *session)
{
    Q_ASSERT (sessions.contains (session));
    SessionData *sessionData = sessions.take (session);
    delete sessionData;
}

void PongCache::sendCachedPongs (const Ping &ping, PacketSession *session)
{
    // Direct Ping
    if (ping.ttl() == 1 && ping.hops() == 0) {
        directPong.setDescriptorId (ping.descriptorId());
        directPong.setIpAddress (localPeer->nodeInfo().address.hostAddress());
        session->sendPacket (directPong);
    } else if (ping.ttl() == 2 && ping.hops() == 0) { // Browse Ping
        foreach (PacketSession *otherSession, sessions.keys()) {
            if (session == otherSession)
                continue;
            SessionData *sessionData = sessions [otherSession];
            sessionData->neighborPong.setDescriptorId (ping.descriptorId());
            otherSession->sendPacket (sessionData->neighborPong);
        }
    } else {
        SessionData *data = sessions [session];
        // Ignore ping if sent less than 1 sec after the previous one.
        if (data->lastInPingTime.elapsed() >= MinimalPingInterval) {
            data->lastInPingTime.start();
            for (int i = 0; i < MaximalPongHops; i++) {
                // Forward the first PongsPerHops number of pongs, which did not
                // come over \a session.
                int pongsSent = 0;
                // to the back. Stop iteration after count() times.
                foreach (PongInfo *pongInfo, cachedPongForwardQueue[i]) {
                    if (pongInfo->second != session) {
                        pongInfo->first.setDescriptorId (ping.descriptorId());
                        session->sendPacket (pongInfo->first);
                        if (++pongsSent == PongsPerHops)
                            break;
                    }
                }
                // Now move the forwarded pongs to the back of the queue.
                if (pongsSent > 0) {
                    foreach (PongInfo *pongInfo, cachedPongForwardQueue[i]) {
                        if (pongInfo->second != session) {
                            cachedPongForwardQueue[i].removeAll (pongInfo);
                            cachedPongForwardQueue[i].enqueue (pongInfo);
                            if (--pongsSent == 0)
                                break;
                        }
                    }
                }
            }
            // \todo Send also pongs for ultrapeer hosts whose connections were rejected
            // during handshaking. These hosts DO have free slots. Send a few for them.

            // \todo Use localPeer->nodeInfo to check the slots and create the
            // pong with all ggep extensions.
            if (slotAllocator->hasFreePeerSlots() || slotAllocator->hasFreeLeafSlots()) {
                hostPong.setDescriptorId (ping.descriptorId());
                hostPong.setIpAddress (localPeer->nodeInfo().address.hostAddress());
                hostPong.setPort (localPeer->nodeInfo().address.hostPort());
                session->sendPacket (hostPong);
            }
        }
    }
}

PongCache::Paths PongCache::pingRoutePaths (const Ping &)
{
    Paths paths;
    foreach (PacketSession *session, sessions.keys()) {
        SessionData *data = this->sessions [session];
        // Ignore ping if sent less than 1 sec after the previous one.
        if (data->lastOutPingTime.elapsed() >= MinimalPingInterval) {
            data->lastOutPingTime.start();
            paths += session;
        }
    }
    return paths;
}

void PongCache::cachePong (const Pong &pong, PacketSession *session)
{
    if (pong.ttl() == 1 && pong.hops() == 0) {
        SessionData *data = sessions [session];
        data->directPong = pong;
        data->neighborPong = pong;
        data->neighborPong.setHops (1);
    } else {
        if (pong.hops() < MaximalPongHops) {
            PongInfo *pongInfo = new PongInfo (pong, session);
            uchar hops = pongInfo->first.hops();
            pongForwardQueue[hops].append (pongInfo);
            pongInfo->first.doHop();
            if (pongInfo->first.ttl() > Gnutella::Packets::MaximalTtl)
                pongInfo->first.setTtl (0);
        }
    }
    // \todo Verify that pongs with hops 0 have the same address as session->remoteNodeAddress()
}

void PongCache::refreshCache()
{
    for (int i = 0; i < MaximalPongHops; i++) {
        // Make sure we did get new pongs before we erase the old cached ones.
        if (pongForwardQueue[i].size() > 0) {
            foreach (PongInfo *pongInfo, cachedPongForwardQueue[i])
                delete pongInfo;
            cachedPongForwardQueue[i] = pongForwardQueue[i];
            std::random_shuffle (cachedPongForwardQueue[i].begin(), cachedPongForwardQueue[i].end());
            pongForwardQueue[i] = PongForwardQueue();
        }
    }

    // Send a PingPacket over all connections.
    Ping ping;

    using Gnutella::Packets::Extensions::Ggeps::SupportsCachedPongs;
    using Gnutella::Packets::Extensions::GgepBlock;
    SupportsCachedPongs scp (SupportsCachedPongs::UltrapeerPreference);

    GgepBlock ggepBlock;
    ggepBlock.addExtension (scp);
    ping.setGgepBlock (ggepBlock);

    /*  Make the PacketProcessor try to route the ping, causing our
        pingRoutePaths() getting called. Thus, the ping is forwarded only over
        sessions that were not recently pinged. */
    packetProcessor->sendPacket (ping);
}

Copyright © 2006 Peter Dimov.
Powered by Doxygen.