DynamicSearch.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 "DynamicSearch.h"
#include "DynamicSearcher.h"
#include "Imports.cpp"

using namespace Gnutella::PacketProcessing::DynamicSearching;

// Open containing namespaces:
namespace Gnutella {
namespace PacketProcessing {
namespace DynamicSearching {

enum Constants
{
    Degree                  = 32,
    TimerHopInterval        = 2400,
    ProbesCount             = 3,
    ProbeTtl                = 2,
    MaxHosts                = 300000,
    SearchConnections       = 1,
    DesiredResults          = 150,
    MaxTtl                  = 3,
    StatusResponseTimeout   = 10000
};

enum State
{
    StateStopped,
    StateSearching,
    StateWaitingStatusResponse
};

class DynamicSearchPrivate {
    REFERENCE_OBJECT (DynamicSearchPrivate)

public:
    typedef QList <PacketSession *> Sessions;

    quint32         hostsQueried;
    quint32         receivedResults;
    Query           query;
    Sessions        remainingSessions;
    QTimer          timer;
    State           state;

    DynamicSearchPrivate() :    hostsQueried (0), receivedResults (0),
                                query(), remainingSessions(), timer(),
                                state(StateStopped)
    {}
};

} // namespace DynamicSearching;
} // namespace PacketProcessing;
} // namespace Gnutella;

DynamicSearch::DynamicSearch (const Query &query, Sessions sessions)
 :  p (new DynamicSearchPrivate)
{
    p->query                = query;
    p->remainingSessions    = sessions;

    QObject::connect (&p->timer, SIGNAL (timeout()), this, SLOT (timeout()));
    p->timer.setSingleShot (true);
}

DynamicSearch::~DynamicSearch()
{
    delete p;
}

void DynamicSearch::startSearch()
{
    doProbeQuery();
    p->state = StateSearching;
}

void DynamicSearch::stopSearch()
{
    p->timer.stop();
    p->state = StateStopped;
}

bool DynamicSearch::hasEnoughHits()
{
    return (p->receivedResults >= DesiredResults)
            || (p->hostsQueried >= MaxHosts)
            || (p->remainingSessions.isEmpty());
}


void DynamicSearch::doProbeQuery()
{
    p->query.setTtl (ProbeTtl);

    qDebug() << p->remainingSessions.count();
    p->hostsQueried = ProbesCount * hostsPerConnection (ProbeTtl);

    quint16 sessionCount = std::min<quint16> (p->remainingSessions.count(), ProbesCount);
    // Don't drop the probe sessions, we may try them at the end again:
    sendPacket (sessionCount, false);
    p->timer.start(ProbeTtl * TimerHopInterval + sessionCount * 1000);
    p->state = StateSearching;
}


void DynamicSearch::doQuery()
{
    quint16 ttl;
    quint16 remainingConns  = p->remainingSessions.count();
    quint16 connections     = std::min <quint16> (remainingConns, SearchConnections);

    if (p->receivedResults != 0) {
        float   averageHostResults = static_cast <float> (p->receivedResults)
                                     / static_cast <float> (p->hostsQueried);
        quint32 hostsToQuery = (DesiredResults - p->receivedResults)
                                / static_cast <int> (averageHostResults);
        qDebug() << remainingConns;
        ttl = computeTtl (hostsToQuery, remainingConns);
    } else {
        ttl = MaxTtl;
    }
    p->hostsQueried += connections * hostsPerConnection (ttl);
    p->query.setTtl (ttl);

    sendPacket (connections);
    p->timer.start (ttl * TimerHopInterval);
    p->state = StateSearching;
}


void DynamicSearch::sendStatusRequest()
{
    QueryStatusRequest request;
    request.setDescriptorId (p->query.descriptorId());

    p->state = StateWaitingStatusResponse;
    p->timer.start (StatusResponseTimeout);
    emit statusRequest (request);
}

void DynamicSearch::timeout()
{
    switch (p->state)
    {
    case StateSearching:
        if (hasEnoughHits()) {
            // If the leaf sends unsolicated QueryStatusResponse messages or
            // we don't have leaf guidance, and we already got enough hits:
            stopSearch();
        } else if (p->query.minimumSpeed() && 0x9000) {
            // Are bits 15 and 12 of MinSpeed were set, so leaf-guidance:
            sendStatusRequest();
        } else {
            doQuery();
        }
        break;
    case StateWaitingStatusResponse:
        stopSearch();
        break;
    case StateStopped:
    default:
        Q_ASSERT (false);
    }
}

void DynamicSearch::countHits (const QueryHits & queryHits)
{
    // Only count hits if not doing leaf-guidance.
    if (!(p->query.minimumSpeed() && 0x9000))
        p->receivedResults += queryHits.numberOfHits();
}


void DynamicSearch::countHits (const QueryStatusResponse &response)
{
    if (response.hitsCount() == 0xFFFF)
        stopSearch();
    else
        p->receivedResults = response.hitsCount();

    if (hasEnoughHits()) {
        stopSearch();
        p->state = StateStopped;
    } else {
        if (p->state == StateWaitingStatusResponse) {
            p->state = StateSearching;
            doQuery();
        }
    }
}

void DynamicSearch::addSession (PacketSession *session)
{
    // \todo if searching stopped due to insufficient PacketSessions, resume it!
    p->remainingSessions.append (session);
}

void DynamicSearch::removeSession (PacketSession *session)
{
    p->remainingSessions.removeAll (session);
}

quint32 DynamicSearch::hostsPerConnection(quint16 ttl)
{
    quint32 hosts = 0;
    quint32 temp = 1;

    for (int i=0; i < ttl; i++) {
        hosts += temp;
        temp  *= Degree - 1;
    }

    return hosts;
}

qint16 DynamicSearch::computeTtl(quint32 hostsToQuery, quint16 numSessions)
{
    int     ttl = 0;
    quint32 hosts;
    quint32 hostsPerConn = 1;

    do {
        ttl++;
        hostsPerConn = hostsPerConnection (ttl);
        hosts = numSessions * hostsPerConn;
    } while (hosts < hostsToQuery);

    return std::min <quint16> (ttl, MaxTtl);
}


void DynamicSearch::sendPacket (quint16 sessionsCount, bool dropSessions)
{
    for (quint16 i = 0; i < sessionsCount; i++) {
        PacketSession *session = p->remainingSessions.takeFirst();
        session->sendPacket (p->query);
        if (!dropSessions)
            p->remainingSessions.append (session);
    }
}

Copyright © 2006 Peter Dimov.
Powered by Doxygen.