UdpHostCache.cpp

Go to the documentation of this file.

/*

Copyright (C) 2006-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 "UdpHostCache.h"
#include "PingHandler.h"
#include "Imports.cpp"

using namespace Gnutella::Bootstrapping;

namespace Gnutella {
namespace Bootstrapping {

enum Constants
{
    SessionReadTimeout      = 10000     
};

struct CacheInfo
{
    NodeAddress     address;
    int             tryCount;
    int             successCount;
    int             failureCount;
    bool            isQuerying;
    bool            gotPong;
    //int               lastHostsCount;
    //int               totalHostsCount;

    CacheInfo (const NodeAddress &address_ = NodeAddress())
     :  address (address_), tryCount (0), successCount (0), failureCount (0),
        isQuerying (false), gotPong (false)
    {}
};

class UdpHostCachePrivate
{
    REFERENCE_OBJECT (UdpHostCachePrivate)

public:
    typedef QMap <UdpConnection *, CacheInfo *> ConnectingConnections;
    typedef QMap <PacketSession *, CacheInfo *> Sessions;
    typedef QQueue <CacheInfo *>                Caches;

    UdpSwitch                       *udpSwitch;
    PingHandler                     *pingHandler;
    ConnectingConnections           connectingConnections;
    Sessions                        sessions;
    Caches                          caches;

    UdpHostCachePrivate() : udpSwitch (0), pingHandler (0),
                            connectingConnections(), sessions(), caches()
    {}
};

} // namespace Bootstrapping;
} // namespace Gnutella;

UdpHostCache::UdpHostCache (UdpSwitch *udpSwitch, PingHandler *pingHandler)
 :  p (new UdpHostCachePrivate)
{
    p->udpSwitch    = udpSwitch;
    p->pingHandler  = pingHandler;

    // \todo Hardcode a list of UHCs and use it if the hosts file is empty.
    //addCache (NodeAddress (QHostAddress ("85.25.67.89"), 9999));
    addCache (NodeAddress ("secondary.udp-host-cache.com", 9999));
}

UdpHostCache::~UdpHostCache()
{
    //  \todo Make sure all active sessions are closed!
    while (!p->caches.isEmpty())
        delete p->caches.dequeue();
    delete p;
}

void UdpHostCache::addCache (const NodeAddress &cacheAddress)
{
    foreach (CacheInfo *info, p->caches) {
        if (info->address == cacheAddress) {
            // \todo Emit a singnal if the address is already added.
            return;
        }
    }
    CacheInfo *info = new CacheInfo (cacheAddress);
    p->caches.enqueue (info);
}

void UdpHostCache::removeCache (const NodeAddress &cacheAddress)
{
    // \todo Make sure no session is mapped to this info!
    foreach (CacheInfo *info, p->caches) {
        if (info->address == cacheAddress) {
            p->caches.removeAll (info);
            delete info;
            break;
        }
    }
}


void UdpHostCache::queryNode (const NodeAddress &address)
{
    doQuery (address, 0);
}

void UdpHostCache::queryCache()
{
    /*  \todo For GUI support, add a nodeAddress argument and find the info that
        has this address. */

    // Do not double query a cache at the same time:
    if (p->caches.front()->isQuerying)
        return;

    CacheInfo *info = p->caches.dequeue();
    info->tryCount++;
    info->isQuerying = true;
    info->gotPong   = false;
    p->caches.enqueue (info);

    doQuery (info->address, info);
}

void UdpHostCache::doQuery (const NodeAddress &address, CacheInfo *info)
{
    UdpConnection *connection = new UdpConnection (p->udpSwitch);

    connect (connection,    SIGNAL (connectionEstablished()),
             this,          SLOT (connectionEstablishedSlot()),
             Qt::QueuedConnection);
    connect (connection,    SIGNAL (connectionFailed()),
             this,          SLOT (connectionFailedSlot()),
             Qt::QueuedConnection);

    p->connectingConnections.insert (connection, info);
    connection->connectToNode (address);
}

PacketSession * UdpHostCache::createSession (UdpConnection *connection)
{
    PacketSession *session = new PacketSession (connection, NodeInfo());
    session->setReadTimeout (SessionReadTimeout);

    connect (session, SIGNAL (packetRead (Packet &, PacketSession *)),
             this, SLOT (processPacket (Packet &, PacketSession *)));
    connect (session, SIGNAL (sessionClosed (PacketSession *)),
             this, SLOT (closeSession (PacketSession *)),
             Qt::QueuedConnection);

    return session;
}

void UdpHostCache::connectionAcceptedSlot (UdpConnection *incomingConnection)
{
    PacketSession *session = createSession (incomingConnection);
    p->sessions.insert (session, 0);
}

void UdpHostCache::connectionFailedSlot()
{
    UdpConnection *connection = reinterpret_cast <UdpConnection *> (sender());
    Q_ASSERT (connection != 0);
    Q_ASSERT (p->connectingConnections.contains (connection));

    CacheInfo *info = p->connectingConnections.value (connection);
    info->failureCount++;
    info->isQuerying = false;
    p->connectingConnections.remove (connection);

    connection->disconnect (this);
    delete connection;
}

void UdpHostCache::connectionEstablishedSlot()
{
    UdpConnection *connection = reinterpret_cast <UdpConnection *> (sender());
    Q_ASSERT (connection != 0);
    Q_ASSERT (p->connectingConnections.contains (connection));

    CacheInfo *info = p->connectingConnections.value (connection);
    p->connectingConnections.remove (connection);

    PacketSession *session = createSession (connection);
    p->sessions.insert (session, info);

    SupportsCachedPongs scp (SupportsCachedPongs::UltrapeerPreference);

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

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

    session->sendPacket (ping);
}

void UdpHostCache::processPacket (Packet &packet, PacketSession *session)
{
    Q_ASSERT (p->sessions.contains (session));
    using namespace Gnutella::Packets;

    switch (packet.payloadDescriptor())
    {
    case PingDescriptor:
        {
qDebug() << "received UDP ping, sending a pong back";
            Ping &ping = Ping::castFrom (packet);
            Pong pong = p->pingHandler->makePongFor (ping);
            session->sendPacket (pong);
        }
        break;
    case PongDescriptor:
        {
            CacheInfo *info = p->sessions.value (session);
            if (info) {
                info->gotPong = true;
                info->successCount++;
            }
            Pong &pong = Pong::castFrom (packet);
qDebug() << "received UDP pong: " << pong.ipAddress().toString() << pong.port();
            emit receivedPong (pong);
            // \todo Wait for further pongs?
            //session->disconnectFromNode();
        }
        break;
    case QueryDescriptor:
    case QueryHitsDescriptor:
    case PushDescriptor:
    case ByeDescriptor:
    case IbmcDescriptor:
    case QueryRoutingDescriptor:
    case OpenVendorDescriptor:
    case StandardVendorDescriptor:
    default:
        break;
    }
}

void UdpHostCache::closeSession (PacketSession *session)
{
    Q_ASSERT (p->sessions.contains (session));

    CacheInfo *info = p->sessions.value (session);
    if (info) {
        if (!info->gotPong)
            info->failureCount++;
        info->isQuerying    = false;
        info->gotPong       = false;
    }
    p->sessions.remove (session);
    session->disconnect (this);
    delete session;
}

bool UdpHostCache::loadCaches (const QString &filename)
{
    QFile file (filename);
    if (!file.exists() || !file.open (QIODevice::ReadOnly))
        return false;

    QDataStream stream (&file);
    // \todo Introduce format versioning.
    int count = 0;
    stream >> count;
    // \todo Verify the host is not duplicated due to corrupted file
    // \todo get the end() iterators in advance.
    while (count-- > 0) {
        CacheInfo *info = new CacheInfo;

        stream >> info->address;
        stream >> info->tryCount;
        stream >> info->successCount;
        stream >> info->failureCount;

        info->isQuerying    = false;
        info->gotPong       = false;

        p->caches.append (info);
    }
    return true;
}

bool UdpHostCache::storeCaches (const QString &filename)
{
    // \todo Do not store the hosts that were unavailable?
    // The list was sordered by access time, which means that
    QFile file (filename);

    if (!file.open (QIODevice::WriteOnly))
        return false;

    QDataStream stream (&file);

    // We now do not store the host that were unavailable.
    stream << p->caches.count();
    UdpHostCachePrivate::Caches::iterator i = p->caches.begin();
    UdpHostCachePrivate::Caches::iterator e = p->caches.end();
    for (; i != e; i++) {
        CacheInfo *info = *i;
        stream << info->address;
        stream << info->tryCount;
        stream << info->successCount;
        stream << info->failureCount;
    }
    return true;
}

Copyright © 2006 Peter Dimov.
Powered by Doxygen.