PacketProcessor.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 "PacketProcessor.h"
#include "PongCache.h"
#include "PacketRouter.h"
//#include "QueryRouter.h"
#include "Gnutella/LocalPeer.h"
#include "Protocols/Transports/Connection.h"
#include "Gnutella/PacketProcessing/DynamicSearching/DynamicSearcher.h"
#include "Gnutella/PacketProcessing/PacketSession.h"
#include "Gnutella/PacketProcessing/QueryRouting/QrtExchanger.h"
#include "Gnutella/Packets/Ping.h"
#include "Gnutella/Packets/Pong.h"
#include "Gnutella/Packets/Query.h"
#include "Gnutella/Packets/QueryHits.h"
#include "Gnutella/Packets/Push.h"
#include "Gnutella/Packets/Bye.h"
#include "Gnutella/Packets/VendorMessages/SupportedMessages.h"
#include "Gnutella/Packets/VendorMessages/QueryStatusResponse.h"
#include "Gnutella/Packets/VendorMessages/QueryStatusRequest.h"
#include "Gnutella/NodeType.h"
#include "Gnutella/Handshaking/SlotAllocator.h"

using namespace Gnutella::Packets;
using Gnutella::Packets::VendorMessages::SupportedMessages;
using Gnutella::Packets::VendorMessages::QueryStatusResponse;
using Gnutella::Packets::VendorMessages::QueryStatusRequest;
using Gnutella::PacketProcessing::DynamicSearching::DynamicSearcher;
using Gnutella::PacketProcessing::PacketProcessor;
using Gnutella::PacketProcessing::PacketSession;
using Gnutella::PacketProcessing::PongCache;
using Gnutella::PacketProcessing::PacketRouter;
using Gnutella::PacketProcessing::QueryRouting::QrtExchanger;
using Gnutella::Bootstrapping::NodeCache;
using Gnutella::Handshaking::SlotAllocator;
using Gnutella::NodeInfo;
using Gnutella::LocalPeer;
using Protocols::Transports::Connection;

typedef PacketRouter GuidRouter;
//using Gnutella::PacketProcessing::QueryRouter;

class Gnutella::PacketProcessing::PacketProcessorPrivate
{
    REFERENCE_OBJECT (PacketProcessorPrivate)

public:
    LocalPeer           *localPeer;
    SlotAllocator       *slotAllocator;
    PacketProcessor::Sessions       sessions;
    PongCache           *pongCache;
    GuidRouter          *guidRouter;
    //QueryRouter       queryRouter;
    DynamicSearcher     *dynamicSearcher;
    QrtExchanger        *qrtExchanger;

    PacketProcessor::Sessions leafSessions() const
    {
        PacketProcessor::Sessions leafSessions;
        foreach (PacketSession *session, sessions)
            if (session->nodeInfo().type == TypeLeaf)
                leafSessions += session;
        return leafSessions;
    }

    PacketProcessorPrivate() : localPeer (0), slotAllocator (0), sessions(),
                                pongCache(0), guidRouter (0),
                                dynamicSearcher (0), qrtExchanger (0)
    {}
};

PacketProcessor::PacketProcessor (LocalPeer *lp)
 :  d (new PacketProcessorPrivate)
{
    d->pongCache = new PongCache (this, lp);
    d->guidRouter = new GuidRouter();
    d->dynamicSearcher = new DynamicSearcher (this);
    d->qrtExchanger = new QrtExchanger();
    d->localPeer = lp;
    d->slotAllocator = lp->slotAllocator();
}

PacketProcessor::~PacketProcessor()
{
    delete d->pongCache;
    delete d->guidRouter;
    delete d->dynamicSearcher;
    delete d->qrtExchanger;
    delete d;
}

void PacketProcessor::addConnection (Connection *connection, NodeInfo nodeInfo)
{
    Q_ASSERT (connection != 0);

    PacketSession *session = new PacketSession (connection, nodeInfo);

    QObject::connect (session, SIGNAL (packetRead (Packet &, PacketSession *)),
                      this, SLOT (packetReceived (Packet &, PacketSession *)));
    QObject::connect (session, SIGNAL (sessionClosed (PacketSession *)),
                      this, SLOT (sessionClosed (PacketSession *)));
    QObject::connect (session, SIGNAL (qrtRead (PacketSession *)),
                      d->qrtExchanger, SLOT (qrtRead (PacketSession *)));

    bool gotSlot = d->slotAllocator->allocateSlot (session);
    d->sessions.insert (session);
    d->pongCache->addSession (session);
    d->dynamicSearcher->addSession (session);
    d->qrtExchanger->addSession (session);
    // \todo Is this the right was to drop the connection?
    if (!gotSlot) {
        connection->disconnectFromNode();
    } else {
        Ping ping;
        ping.setTtl (1);
        session->sendPacket (ping);
        // \todo Maybe have the VendorMessageFactory return the packet
        // it will know anyway about all types of messages it supports.
        SupportedMessages supported;
        supported.addMessage (VendorCode ("BEAR"), 11, 1); // QueryStatusRequest
        supported.addMessage (VendorCode ("BEAR"), 12, 1); // QueryStatusResponse
        session->sendPacket (supported);
        // \todo We should read the MessageseSupported message sent from the
        // remote peer and only send out vendor messages it they are supported.
    }
}

void PacketProcessor::sessionClosed (PacketSession *session)
{
    Q_ASSERT (d->sessions.contains (session));

    d->qrtExchanger->removeSession (session);
    d->dynamicSearcher->removeSession (session);
    d->pongCache->removeSession (session);
    d->sessions.remove (session);
    session->deleteLater();
}

/*
*/
void PacketProcessor::packetReceived (Packet &packet,
                                      PacketSession *session)
{
    // (Cached) Pong packets may have ttl of 0 => no doHop().
    if (packet.ttl() > 0)
        packet.doHop();

    if (d->guidRouter->isDuplicate (packet))
        return;

    // Store route paths.
    switch (packet.payloadDescriptor())
    {
    case PingDescriptor:
        d->pongCache->sendCachedPongs (Ping::castFrom (packet), session);
        break;
    case PongDescriptor:
        d->pongCache->cachePong (Pong::castFrom (packet), session);
        break;
    case QueryDescriptor:
        d->guidRouter->addQueryPath (Query::castFrom (packet), session);
        if (session->nodeInfo().type == TypeLeaf) {
            d->dynamicSearcher->startSearch (Query::castFrom (packet));
            // \todo We need to query our own leaves just the normal way!
            return; // \todo notifyHandlers() is skipped and we only want to skip routePacket()
        }
        break;
    case QueryHitsDescriptor:
        d->guidRouter->addQueryHitsPath (QueryHits::castFrom (packet), session);
        d->dynamicSearcher->countHits (QueryHits::castFrom (packet));
        break;
    case PushDescriptor:
        break;
    case ByeDescriptor:
qDebug() << "Bye packet saying:" << Bye::castFrom (packet).code() << Bye::castFrom (packet).message();
        session->closeSession (PacketSession::CloseNoBye);
        break;
    case OpenVendorDescriptor:
    case StandardVendorDescriptor:
        if (typeid (packet) == typeid (QueryStatusResponse)) {
            d->dynamicSearcher->countHits (QueryStatusResponse::castFrom (packet));
        } else if (typeid (packet) == typeid (QueryStatusRequest)) {
            // Since we used the same descriptorId for queries sent over
            // different connections we would not be able to determine to
            // exactly which node to route the QueryStatusResponse back.
            // That would not be a problem though because such unsolicited
            // packet will be handled gracefully according the specs.
            d->guidRouter->addQueryStatusRequestPath (packet, session);
        }
        break;
    case IbmcDescriptor:
    case QueryRoutingDescriptor:
    default:
        return;
    }
    notifyHandlers (packet);

    // If we are a leaf, do not forward packets.
    if (d->localPeer->nodeInfo().type == Gnutella::TypeLeaf)
        return;
    routePacket (packet);
}

void PacketProcessor::notifyHandlers (Packet &packet)
{
    switch (packet.payloadDescriptor())
    {
    case PingDescriptor:
        // With pong cacheing, the handler does not need to respond to each
        // ping. It also lacks the knowledge where exactly pongs came from, so
        // it is impossible to guarantee pongs are not routed back from where
        // they came. The PongCache does all this.
        // \todo Actually, the handler never needs to get or answer a ping.
        // It just handles the pongs.
        //if (p.pongCache->canNotifyHandler())
        //  emit receivedPing (Ping::castFrom (packet));
        break;
    case PongDescriptor:
        emit receivedPong (Pong::castFrom (packet));
        break;
    case QueryDescriptor:
        emit receivedQuery (Query::castFrom (packet));
        break;
    case QueryHitsDescriptor:
        emit receivedQueryHits (QueryHits::castFrom (packet));
        break;
    case PushDescriptor:
        emit receivedPush (Push::castFrom (packet));
        break;
    case ByeDescriptor:
        break; // \todo Do we really need the assert below?
    case OpenVendorDescriptor:
    case StandardVendorDescriptor:
        if (typeid (packet) == typeid (QueryStatusResponse)) {
            // A leaf might be sinding us a QueryStatusResponse.
            d->dynamicSearcher->countHits (QueryStatusResponse::castFrom (packet));
        } else if (typeid (packet) == typeid (QueryStatusRequest)) {
            // An ultrapeer might be sending us a QueryStatusRequest.
            emit processQueryStatusRequest (QueryStatusRequest::castFrom (packet));
        }
        break;
    case IbmcDescriptor:
    case QueryRoutingDescriptor:
    default:
        Q_ASSERT (false);
    }
}

void PacketProcessor::routePacket (const Packet &packet)
{
    // \todo Query packets with ttl of 1 (0 after doHop()) should be forwarded
    // to leaves only. Should be in QueryRouter.
    Sessions sessions;
    switch (packet.payloadDescriptor())
    {
    case PingDescriptor:
        sessions = d->pongCache->pingRoutePaths (Ping::castFrom (packet));
        break;
    case PongDescriptor:
        // The PongCache responds immediately to a ping, no need to forward
        // the pongs, they were already cached.
        break;
    case QueryDescriptor:
        //sessions = p.queryRouter->getQueryPath (Query::castFrom (packet));
        if (packet.ttl() == 0)
            sessions = d->leafSessions();
        else
            sessions = d->sessions;

        break;
    case QueryHitsDescriptor:
        sessions = d->guidRouter->queryHitsRoutePaths
                                    (QueryHits::castFrom (packet));
        break;
    case PushDescriptor:
        sessions = d->guidRouter->pushRoutePaths (Push::castFrom (packet));
        break;
    case ByeDescriptor:
        break; // \todo rethink the assert below!
    case OpenVendorDescriptor:
    case StandardVendorDescriptor:
        // Only route the messages originating from us:
        if (packet.hops() == 0 && packet.ttl() == 1) {
            if (typeid (packet) == typeid (QueryStatusRequest)){
                // Send back to leaf, i.e. where the Query came from:
                sessions = d->guidRouter->queryHitsRoutePaths (packet);
            } else if (typeid (packet) == typeid (QueryStatusResponse)) {
                // Send forward to ultrapeers, i.e. where the query was sent to:
                //Q_ASSERT (d->localPeer->nodeInfo().type == TypeLeaf);
                sessions = d->guidRouter->queryStatusResponsePaths (packet);
            }
        }
        break;
    case IbmcDescriptor:
    case QueryRoutingDescriptor:
    default:
        Q_ASSERT (false);
    }

    foreach (PacketSession *routePath, sessions) {
        // The routePath may have been closed, so check that:
        // \todo Should we make the routers take care for that?
        if (d->sessions.contains (routePath))
            routePath->sendPacket (packet);
    }
}

void PacketProcessor::sendPacket (const Packet &packet)
{
    // Only packets our node sends should be sent using this function.
    Q_ASSERT (packet.hops() == 0);

    // The pongs returned from the HostCache need to be cached. not routed.
    switch (packet.payloadDescriptor())
    {
        case PongDescriptor:
            d->pongCache->cachePong (Pong::castFrom (packet), 0);
            break;
        case QueryDescriptor:
            if (d->localPeer->nodeInfo().type == TypeLeaf)
                routePacket (packet);
            else
                d->dynamicSearcher->startSearch (Query::castFrom (packet));
            break;
        case OpenVendorDescriptor:
            if (typeid (packet) == typeid (QueryStatusResponse)) {
                // We send QueryStatusResponse only to the DynamicSearcher or
                // over the network to our ultrapeers!
                d->dynamicSearcher->countHits (QueryStatusResponse::castFrom (packet));
            } else if (typeid (packet) == typeid (QueryStatusRequest)) {
                // The DynmaciSearcher may send a QueryStatusRequest to a leaf
                // originating a query.
                emit processQueryStatusRequest (QueryStatusRequest::castFrom (packet)); // \todo don't like that!
            }
            routePacket (packet);
            break;
        case PingDescriptor:
        case QueryHitsDescriptor:
        case PushDescriptor:
        case ByeDescriptor:
        case IbmcDescriptor:
        case QueryRoutingDescriptor:
        case StandardVendorDescriptor:
        default:
            routePacket (packet);
            break;
    }
}

Copyright © 2006 Peter Dimov.
Powered by Doxygen.