torrentclient.cpp

qt-x11-opensource-src-4.1.4.tar.gz源码

    }

    // Now make up a list of peers where the ones with more points are
    // listed many times.
    QMultiMap<int, TorrentPeer *>::ConstIterator it = pointMap.constBegin();
    while (it != pointMap.constEnd()) {
        for (int i = 0; i < it.key() + 1; ++i)
            weighedPeers << it.value();
        ++it;
    }

    return weighedPeers;
}

void TorrentClient::setupIncomingConnection(PeerWireClient *client)
{
    // Connect signals
    initializeConnection(client);

    // Initialize this client
    RateController::instance()->addSocket(client);
    d->connections << client;

    client->initialize(d->infoHash, d->pieceCount);
    client->sendPieceList(d->completedPieces);

    emit peerInfoUpdated();

    if (d->state == Searching || d->state == Connecting) {
        int completed = d->completedPieces.count(true);
        if (completed == 0)
            d->setState(WarmingUp);
        else if (d->incompletePieces.count(true) < 5 && d->pendingPieces.size() > d->incompletePieces.count(true))
            d->setState(Endgame);
    }

    if (d->connections.isEmpty())
        scheduleUploads();
}

void TorrentClient::setupOutgoingConnection()
{
    PeerWireClient *client = qobject_cast<PeerWireClient *>(sender());

    // Update connection statistics.
    foreach (TorrentPeer *peer, d->peers) {
        if (peer->port == client->peerPort() && peer->address == client->peerAddress()) {
            peer->connectTime = peer->lastVisited - peer->connectStart;
            break;
        }
    }

    // Send handshake and piece list
    client->initialize(d->infoHash, d->pieceCount);
    client->sendPieceList(d->completedPieces);

    emit peerInfoUpdated();

    if (d->state == Searching || d->state == Connecting) {
        int completed = d->completedPieces.count(true);
        if (completed == 0)
            d->setState(WarmingUp);
        else if (d->incompletePieces.count(true) < 5 && d->pendingPieces.size() > d->incompletePieces.count(true))
            d->setState(Endgame);
    }
}

void TorrentClient::initializeConnection(PeerWireClient *client)
{
    connect(client, SIGNAL(connected()),
            this, SLOT(setupOutgoingConnection()));
    connect(client, SIGNAL(disconnected()),
            this, SLOT(removeClient()));
    connect(client, SIGNAL(error(QAbstractSocket::SocketError)),
            this, SLOT(removeClient()));
    connect(client, SIGNAL(piecesAvailable(const QBitArray &)),
            this, SLOT(peerPiecesAvailable(const QBitArray &)));
    connect(client, SIGNAL(blockRequested(int, int, int)),
            this, SLOT(peerRequestsBlock(int, int, int)));
    connect(client, SIGNAL(blockReceived(int, int, const QByteArray &)),
            this, SLOT(blockReceived(int, int, const QByteArray &)));
    connect(client, SIGNAL(choked()),
            this, SLOT(peerChoked()));
    connect(client, SIGNAL(unchoked()),
            this, SLOT(peerUnchoked()));
    connect(client, SIGNAL(bytesWritten(qint64)),
            this, SLOT(peerWireBytesWritten(qint64)));
    connect(client, SIGNAL(bytesReceived(qint64)),
            this, SLOT(peerWireBytesReceived(qint64)));
}

void TorrentClient::removeClient()
{
    PeerWireClient *client = static_cast<PeerWireClient *>(sender());

    // Remove the host from our list of known peers if the connection
    // failed.
    if (client->peer() && client->error() == QAbstractSocket::ConnectionRefusedError)
        d->peers.removeAll(client->peer());

    // Remove the client from RateController and all structures.
    RateController::instance()->removeSocket(client);
    d->connections.removeAll(client);
    QMultiMap<PeerWireClient *, TorrentPiece *>::Iterator it = d->payloads.find(client);
    while (it != d->payloads.end() && it.key() == client) {
        TorrentPiece *piece = it.value();
        piece->inProgress = false;
        piece->requestedBlocks.fill(false);
        it = d->payloads.erase(it);
    }

    // Remove pending read requests.
    QMapIterator<int, PeerWireClient *> it2(d->readIds);
    while (it2.findNext(client))
        d->readIds.remove(it2.key());

    // Delete the client later.
    disconnect(client, SIGNAL(disconnected()), this, SLOT(removeClient()));
    client->deleteLater();
    ConnectionManager::instance()->removeConnection(client);

    emit peerInfoUpdated();
    d->callPeerConnector();
}

void TorrentClient::peerPiecesAvailable(const QBitArray &pieces)
{
    PeerWireClient *client = qobject_cast<PeerWireClient *>(sender());

    // Find the peer in our list of announced peers. If it's there,
    // then we can use the piece list into to gather statistics that
    // help us decide what peers to connect to.
    TorrentPeer *peer = 0;
    QList<TorrentPeer *>::Iterator it = d->peers.begin();
    while (it != d->peers.end()) {
        if ((*it)->address == client->peerAddress() && (*it)->port == client->peerPort()) {
            peer = *it;
            break;
        }
        ++it;
    }

    // If the peer is a seed, and we are in seeding mode, then the
    // peer is uninteresting.
    if (pieces.count(true) == d->pieceCount) {
        if (peer)
            peer->seed = true;
        emit peerInfoUpdated();
        if (d->state == Seeding) {
            client->abort();
            return;
        } else {
            if (peer)
                peer->interesting = true;
            if ((client->peerWireState() & PeerWireClient::InterestedInPeer) == 0)
                client->sendInterested();
            d->callScheduler();
            return;
        }
    }

    // Update our list of available pieces.
    if (peer) {
        peer->pieces = pieces;
        peer->numCompletedPieces = pieces.count(true);
    }

    // Check for interesting pieces, and tell the peer whether we are
    // interested or not.
    bool interested = false;
    int piecesSize = pieces.size();
    for (int pieceIndex = 0; pieceIndex < piecesSize; ++pieceIndex) {
        if (!pieces.testBit(pieceIndex))
            continue;
        if (!d->completedPieces.testBit(pieceIndex)) {
            interested = true;
            if ((client->peerWireState() & PeerWireClient::InterestedInPeer) == 0) {
                if (peer)
                    peer->interesting = true;
                client->sendInterested();
            }

            QMultiMap<PeerWireClient *, TorrentPiece *>::Iterator it = d->payloads.find(client);
            int inProgress = 0;
            while (it != d->payloads.end() && it.key() == client) {
                if (it.value()->inProgress)
                    inProgress += it.value()->requestedBlocks.count(true);
                ++it;
            }
            if (!inProgress)
                d->callScheduler();
            break;
        }
    }
    if (!interested && (client->peerWireState() & PeerWireClient::InterestedInPeer)) {
        if (peer)
            peer->interesting = false;
        client->sendNotInterested();
    }
}

void TorrentClient::peerRequestsBlock(int pieceIndex, int begin, int length)
{
    PeerWireClient *client = qobject_cast<PeerWireClient *>(sender());

    // Silently ignore requests from choked peers
    if (client->peerWireState() & PeerWireClient::ChokingPeer)
        return;

    // Silently ignore requests for pieces we don't have.
    if (!d->completedPieces.testBit(pieceIndex))
        return;

    // Request the block from the file manager
    d->readIds.insert(d->fileManager.read(pieceIndex, begin, length),
                      qobject_cast<PeerWireClient *>(sender()));
}

void TorrentClient::blockReceived(int pieceIndex, int begin, const QByteArray &data)
{
    PeerWireClient *client = qobject_cast<PeerWireClient *>(sender());
    if (data.size() == 0) {
        client->abort();
        return;
    }

    // Ignore it if we already have this block.
    int blockBit = begin / BlockSize;
    TorrentPiece *piece = d->pendingPieces.value(pieceIndex);
    if (!piece || piece->completedBlocks.testBit(blockBit)) {
        // Discard blocks that we already have, and fill up the pipeline.
        requestMore(client);
        return;
    }

    // If we are in warmup or endgame mode, cancel all duplicate
    // requests for this block.
    if (d->state == WarmingUp || d->state == Endgame) {
        QMultiMap<PeerWireClient *, TorrentPiece *>::Iterator it = d->payloads.begin();
        while (it != d->payloads.end()) {
            PeerWireClient *otherClient = it.key();
            if (otherClient != client && it.value()->index == pieceIndex) {
                if (otherClient->incomingBlocks().contains(TorrentBlock(pieceIndex, begin, data.size())))
                    it.key()->cancelRequest(pieceIndex, begin, data.size());
            }
            ++it;
        }
    }

    if (d->state != Downloading && d->state != Endgame && d->completedPieces.count(true) > 0)
        d->setState(Downloading);

    // Store this block
    d->fileManager.write(pieceIndex, begin, data);
    piece->completedBlocks.setBit(blockBit);
    piece->requestedBlocks.clearBit(blockBit);

    if (blocksLeftForPiece(piece) == 0) {
        // Ask the file manager to verify the newly downloaded piece
        d->fileManager.verifyPiece(piece->index);
        
        // Remove this piece from all payloads
        QMultiMap<PeerWireClient *, TorrentPiece *>::Iterator it = d->payloads.begin();
        while (it != d->payloads.end()) {
            if (!it.value() || it.value()->index == piece->index)
                it = d->payloads.erase(it);
            else
                ++it;
        }
    }

    // Fill up the pipeline.
    requestMore(client);
}

void TorrentClient::peerWireBytesWritten(qint64 size)
{
    if (!d->transferRateTimer)
        d->transferRateTimer = startTimer(RateControlTimerDelay);

    d->uploadRate[0] += size;
    d->uploadedBytes += size;
    emit dataSent(size);
}

void TorrentClient::peerWireBytesReceived(qint64 size)
{
    if (!d->transferRateTimer)
        d->transferRateTimer = startTimer(RateControlTimerDelay);

    d->downloadRate[0] += size;
    d->downloadedBytes += size;
    emit dataSent(size);
}

int TorrentClient::blocksLeftForPiece(const TorrentPiece *piece) const
{
    int blocksLeft = 0;
    int completedBlocksSize = piece->completedBlocks.size();
    for (int i = 0; i < completedBlocksSize; ++i) {
        if (!piece->completedBlocks.testBit(i))
            ++blocksLeft;
    }
    return blocksLeft;
}

void TorrentClient::scheduleUploads()
{
    // Generate a list of clients sorted by their transfer
    // speeds.  When leeching, we sort by download speed, and when
    // seeding, we sort by upload speed. Seeds are left out; there's
    // no use in unchoking them.
    QList<PeerWireClient *> allClients = d->connections;
    QMultiMap<int, PeerWireClient *> transferSpeeds;
    foreach (PeerWireClient *client, allClients) {
        if (client->state() == QAbstractSocket::ConnectedState
            && client->availablePieces().count(true) != d->pieceCount) {
            if (d->state == Seeding) {
                transferSpeeds.insert(client->uploadSpeed(), client);
            } else {
                transferSpeeds.insert(client->downloadSpeed(), client);
            }
        }
    }

    // Unchoke the top 'MaxUploads' downloaders (peers that we are
    // uploading to) and choke all others.
    int maxUploaders = MaxUploads;
    QMapIterator<int, PeerWireClient *> it(transferSpeeds);
    it.toBack();
    while (it.hasPrevious()) {
        PeerWireClient *client = it.previous().value();
        bool interested = (client->peerWireState() & PeerWireClient::PeerIsInterested);

        if (maxUploaders) {
            allClients.removeAll(client);
            if (client->peerWireState() & PeerWireClient::ChokingPeer)
                client->unchokePeer();
            --maxUploaders;
            continue;
        }

        if ((client->peerWireState() & PeerWireClient::ChokingPeer) == 0) {
            if ((rand() % 10) == 0) 
                client->abort();
            else
                client->chokePeer();
            allClients.removeAll(client);
        }
        if (!interested)
            allClients.removeAll(client);
    }

    // Only interested peers are left in allClients. Unchoke one
    // random peer to allow it to compete for a position among the
    // downloaders.  (This is known as an "optimistic unchoke".)
    if (!allClients.isEmpty()) {
        PeerWireClient *client = allClients[rand() % allClients.size()];
        if (client->peerWireState() & PeerWireClient::ChokingPeer)
            client->unchokePeer();
    }
}

void TorrentClient::scheduleDownloads()
{
    d->schedulerCalled = false;

    if (d->state == Stopping || d->state == Paused || d->state == Idle)
        return;

    // Check what each client is doing, and assign payloads to those
    // who are either idle or done.
    foreach (PeerWireClient *client, d->connections)
        schedulePieceForClient(client);
}

void TorrentClient::schedulePieceForClient(PeerWireClient *client)
{
    // Only schedule connected clients.
    if (client->state() != QTcpSocket::ConnectedState)
        return;

    // The peer has choked us; try again later.