torrentclient.cpp Example File network/torrent/torrentclient.cpp
#include "connectionmanager.h"
#include "filemanager.h"
#include "metainfo.h"
#include "torrentclient.h"
#include "torrentserver.h"
#include "trackerclient.h"
#include "peerwireclient.h"
#include "ratecontroller.h"
#include <QtCore>
#include <QNetworkInterface>
static const int ServerMinPort = 6881;
static const int ServerMaxPort = 7000;
static const int BlockSize = 16384;
static const int MaxBlocksInProgress = 5;
static const int MaxBlocksInMultiMode = 2;
static const int MaxConnectionPerPeer = 1;
static const int RateControlWindowLength = 10;
static const int RateControlTimerDelay = 1000;
static const int MinimumTimeBeforeRevisit = 30;
static const int MaxUploads = 4;
static const int UploadScheduleInterval = 10000;
static const int EndGamePieces = 5;
class TorrentPiece {
public:
int index;
int length;
QBitArray completedBlocks;
QBitArray requestedBlocks;
bool inProgress;
};
class TorrentClientPrivate
{
public:
TorrentClientPrivate(TorrentClient *qq);
void setError(TorrentClient::Error error);
void setState(TorrentClient::State state);
TorrentClient::Error error;
TorrentClient::State state;
QString errorString;
QString stateString;
QString destinationFolder;
MetaInfo metaInfo;
QByteArray peerId;
QByteArray infoHash;
TrackerClient trackerClient;
FileManager fileManager;
QList<PeerWireClient *> connections;
QList<TorrentPeer *> peers;
bool schedulerCalled;
void callScheduler();
bool connectingToClients;
void callPeerConnector();
int uploadScheduleTimer;
QMap<int, PeerWireClient *> readIds;
QMultiMap<PeerWireClient *, TorrentPiece *> payloads;
QMap<int, TorrentPiece *> pendingPieces;
QBitArray completedPieces;
QBitArray incompletePieces;
int pieceCount;
int lastProgressValue;
qint64 downloadedBytes;
qint64 uploadedBytes;
int downloadRate[RateControlWindowLength];
int uploadRate[RateControlWindowLength];
int transferRateTimer;
TorrentClient *q;
};
TorrentClientPrivate::TorrentClientPrivate(TorrentClient *qq)
: trackerClient(qq), q(qq)
{
error = TorrentClient::UnknownError;
state = TorrentClient::Idle;
errorString = QT_TRANSLATE_NOOP(TorrentClient, "Unknown error");
stateString = QT_TRANSLATE_NOOP(TorrentClient, "Idle");
schedulerCalled = false;
connectingToClients = false;
uploadScheduleTimer = 0;
lastProgressValue = -1;
pieceCount = 0;
downloadedBytes = 0;
uploadedBytes = 0;
memset(downloadRate, 0, sizeof(downloadRate));
memset(uploadRate, 0, sizeof(uploadRate));
transferRateTimer = 0;
}
void TorrentClientPrivate::setError(TorrentClient::Error errorCode)
{
this->error = errorCode;
switch (error) {
case TorrentClient::UnknownError:
errorString = QT_TRANSLATE_NOOP(TorrentClient, "Unknown error");
break;
case TorrentClient::TorrentParseError:
errorString = QT_TRANSLATE_NOOP(TorrentClient, "Invalid torrent data");
break;
case TorrentClient::InvalidTrackerError:
errorString = QT_TRANSLATE_NOOP(TorrentClient, "Unable to connect to tracker");
break;
case TorrentClient::FileError:
errorString = QT_TRANSLATE_NOOP(TorrentClient, "File error");
break;
case TorrentClient::ServerError:
errorString = QT_TRANSLATE_NOOP(TorrentClient, "Unable to initialize server");
break;
}
emit q->error(errorCode);
}
void TorrentClientPrivate::setState(TorrentClient::State state)
{
this->state = state;
switch (state) {
case TorrentClient::Idle:
stateString = QT_TRANSLATE_NOOP(TorrentClient, "Idle");
break;
case TorrentClient::Paused:
stateString = QT_TRANSLATE_NOOP(TorrentClient, "Paused");
break;
case TorrentClient::Stopping:
stateString = QT_TRANSLATE_NOOP(TorrentClient, "Stopping");
break;
case TorrentClient::Preparing:
stateString = QT_TRANSLATE_NOOP(TorrentClient, "Preparing");
break;
case TorrentClient::Searching:
stateString = QT_TRANSLATE_NOOP(TorrentClient, "Searching");
break;
case TorrentClient::Connecting:
stateString = QT_TRANSLATE_NOOP(TorrentClient, "Connecting");
break;
case TorrentClient::WarmingUp:
stateString = QT_TRANSLATE_NOOP(TorrentClient, "Warming up");
break;
case TorrentClient::Downloading:
stateString = QT_TRANSLATE_NOOP(TorrentClient, "Downloading");
break;
case TorrentClient::Endgame:
stateString = QT_TRANSLATE_NOOP(TorrentClient, "Finishing");
break;
case TorrentClient::Seeding:
stateString = QT_TRANSLATE_NOOP(TorrentClient, "Seeding");
break;
}
emit q->stateChanged(state);
}
void TorrentClientPrivate::callScheduler()
{
if (!schedulerCalled) {
schedulerCalled = true;
QMetaObject::invokeMethod(q, "scheduleDownloads", Qt::QueuedConnection);
}
}
void TorrentClientPrivate::callPeerConnector()
{
if (!connectingToClients) {
connectingToClients = true;
QTimer::singleShot(10000, q, SLOT(connectToPeers()));
}
}
TorrentClient::TorrentClient(QObject *parent)
: QObject(parent), d(new TorrentClientPrivate(this))
{
connect(&d->fileManager, SIGNAL(dataRead(int,int,int,QByteArray)),
this, SLOT(sendToPeer(int,int,int,QByteArray)));
connect(&d->fileManager, SIGNAL(verificationProgress(int)),
this, SLOT(updateProgress(int)));
connect(&d->fileManager, SIGNAL(verificationDone()),
this, SLOT(fullVerificationDone()));
connect(&d->fileManager, SIGNAL(pieceVerified(int,bool)),
this, SLOT(pieceVerified(int,bool)));
connect(&d->fileManager, SIGNAL(error()),
this, SLOT(handleFileError()));
connect(&d->trackerClient, SIGNAL(peerListUpdated(QList<TorrentPeer>)),
this, SLOT(addToPeerList(QList<TorrentPeer>)));
connect(&d->trackerClient, SIGNAL(stopped()),
this, SIGNAL(stopped()));
}
TorrentClient::~TorrentClient()
{
qDeleteAll(d->peers);
qDeleteAll(d->pendingPieces);
delete d;
}
bool TorrentClient::setTorrent(const QString &fileName)
{
QFile file(fileName);
if (!file.open(QIODevice::ReadOnly) || !setTorrent(file.readAll())) {
d->setError(TorrentParseError);
return false;
}
return true;
}
bool TorrentClient::setTorrent(const QByteArray &torrentData)
{
if (!d->metaInfo.parse(torrentData)) {
d->setError(TorrentParseError);
return false;
}
QByteArray infoValue = d->metaInfo.infoValue();
d->infoHash = QCryptographicHash::hash(infoValue, QCryptographicHash::Sha1);
return true;
}
MetaInfo TorrentClient::metaInfo() const
{
return d->metaInfo;
}
void TorrentClient::setDestinationFolder(const QString &directory)
{
d->destinationFolder = directory;
}
QString TorrentClient::destinationFolder() const
{
return d->destinationFolder;
}
void TorrentClient::setDumpedState(const QByteArray &dumpedState)
{
QDataStream stream(dumpedState);
quint16 version = 0;
stream >> version;
if (version != 2)
return;
stream >> d->completedPieces;
while (!stream.atEnd()) {
int index;
int length;
QBitArray completed;
stream >> index >> length >> completed;
if (stream.status() != QDataStream::Ok) {
d->completedPieces.clear();
break;
}
TorrentPiece *piece = new TorrentPiece;
piece->index = index;
piece->length = length;
piece->completedBlocks = completed;
piece->requestedBlocks.resize(completed.size());
piece->inProgress = false;
d->pendingPieces[index] = piece;
}
}
QByteArray TorrentClient::dumpedState() const
{
QByteArray partials;
QDataStream stream(&partials, QIODevice::WriteOnly);
stream << quint16(2);
stream << d->completedPieces;
QMap<int, TorrentPiece *>::ConstIterator it = d->pendingPieces.constBegin();
while (it != d->pendingPieces.constEnd()) {
TorrentPiece *piece = it.value();
if (blocksLeftForPiece(piece) > 0 && blocksLeftForPiece(piece) < piece->completedBlocks.size()) {
stream << piece->index;
stream << piece->length;
stream << piece->completedBlocks;
}
++it;
}
return partials;
}
qint64 TorrentClient::progress() const
{
return d->lastProgressValue;
}
void TorrentClient::setDownloadedBytes(qint64 bytes)
{
d->downloadedBytes = bytes;
}
qint64 TorrentClient::downloadedBytes() const
{
return d->downloadedBytes;
}
void TorrentClient::setUploadedBytes(qint64 bytes)
{
d->uploadedBytes = bytes;
}
qint64 TorrentClient::uploadedBytes() const
{
return d->uploadedBytes;
}
int TorrentClient::connectedPeerCount() const
{
int tmp = 0;
foreach (PeerWireClient *client, d->connections) {
if (client->state() == QAbstractSocket::ConnectedState)
++tmp;
}
return tmp;
}
int TorrentClient::seedCount() const
{
int tmp = 0;
foreach (PeerWireClient *client, d->connections) {
if (client->availablePieces().count(true) == d->pieceCount)
++tmp;
}
return tmp;
}
TorrentClient::State TorrentClient::state() const
{
return d->state;
}
QString TorrentClient::stateString() const
{
return d->stateString;
}
TorrentClient::Error TorrentClient::error() const
{
return d->error;
}
QString TorrentClient::errorString() const
{
return d->errorString;
}
QByteArray TorrentClient::peerId() const
{
return d->peerId;
}
QByteArray TorrentClient::infoHash() const
{
return d->infoHash;
}
void TorrentClient::start()
{
if (d->state != Idle)
return;
TorrentServer::instance()->addClient(this);
d->setState(Preparing);
d->fileManager.setMetaInfo(d->metaInfo);
d->fileManager.setDestinationFolder(d->destinationFolder);
d->fileManager.setCompletedPieces(d->completedPieces);
d->fileManager.start(QThread::LowestPriority);
d->fileManager.startDataVerification();
}
void TorrentClient::stop()
{
if (d->state == Stopping)
return;
TorrentServer::instance()->removeClient(this);
State oldState = d->state;
d->setState(Stopping);
if (d->transferRateTimer) {
killTimer(d->transferRateTimer);
d->transferRateTimer = 0;
}
foreach (PeerWireClient *client, d->connections) {
RateController::instance()->removeSocket(client);
ConnectionManager::instance()->removeConnection(client);
client->abort();
}
d->connections.clear();
if (oldState > Preparing) {
d->trackerClient.stop();
} else {
d->setState(Idle);
emit stopped();
}
}
void TorrentClient::setPaused(bool paused)
{
if (paused) {
d->setState(Paused);
foreach (PeerWireClient *client, d->connections)
client->abort();
d->connections.clear();
TorrentServer::instance()->removeClient(this);
} else {
d->setState(d->completedPieces.count(true) == d->fileManager.pieceCount()
? Seeding : Searching);
connectToPeers();
TorrentServer::instance()->addClient(this);
}
}
void TorrentClient::timerEvent(QTimerEvent *event)
{
if (event->timerId() == d->uploadScheduleTimer) {
scheduleUploads();
return;
}
if (event->timerId() != d->transferRateTimer) {
QObject::timerEvent(event);
return;
}
qint64 uploadBytesPerSecond = 0;
qint64 downloadBytesPerSecond = 0;
for (int i = 0; i < RateControlWindowLength; ++i) {
uploadBytesPerSecond += d->uploadRate[i];
downloadBytesPerSecond += d->downloadRate[i];
}
uploadBytesPerSecond /= qint64(RateControlWindowLength);
downloadBytesPerSecond /= qint64(RateControlWindowLength);
for (int i = RateControlWindowLength - 2; i >= 0; --i) {
d->uploadRate[i + 1] = d->uploadRate[i];
d->downloadRate[i + 1] = d->downloadRate[i];
}
d->uploadRate[0] = 0;
d->downloadRate[0] = 0;
emit uploadRateUpdated(int(uploadBytesPerSecond));
emit downloadRateUpdated(int(downloadBytesPerSecond));
if (downloadBytesPerSecond == 0 && uploadBytesPerSecond == 0) {
killTimer(d->transferRateTimer);
d->transferRateTimer = 0;
}
}
void TorrentClient::sendToPeer(int readId, int pieceIndex, int begin, const QByteArray &data)
{
PeerWireClient *client = d->readIds.value(readId);
if (client) {
if ((client->peerWireState() & PeerWireClient::ChokingPeer) == 0)
client->sendBlock(pieceIndex, begin, data);
}
d->readIds.remove(readId);
}
void TorrentClient::fullVerificationDone()
{
d->completedPieces = d->fileManager.completedPieces();
d->incompletePieces.resize(d->completedPieces.size());
d->pieceCount = d->completedPieces.size();
for (int i = 0; i < d->fileManager.pieceCount(); ++i) {
if (!d->completedPieces.testBit(i))
d->incompletePieces.setBit(i);
}
updateProgress();
QMap<int, TorrentPiece *>::Iterator it = d->pendingPieces.begin();
while (it != d->pendingPieces.end()) {
if (d->completedPieces.testBit(it.key()))
it = d->pendingPieces.erase(it);
else
++it;
}
d->uploadScheduleTimer = startTimer(UploadScheduleInterval);
TorrentServer *server = TorrentServer::instance();
if (!server->isListening()) {
for (int i = ServerMinPort; i <= ServerMaxPort; ++i) {
if (server->listen(QHostAddress::Any, i))
break;
}
if (!server->isListening()) {
d->setError(ServerError);
return;
}
}
d->setState(d->completedPieces.count(true) == d->pieceCount ? Seeding : Searching);
d->trackerClient.start(d->metaInfo);
}
void TorrentClient::pieceVerified(int pieceIndex, bool ok)
{
TorrentPiece *piece = d->pendingPieces.value(pieceIndex);
QMultiMap<PeerWireClient *, TorrentPiece *>::Iterator it = d->payloads.begin();
while (it != d->payloads.end()) {
if (it.value()->index == pieceIndex)
it = d->payloads.erase(it);
else
++it;
}
if (!ok) {
piece->inProgress = false;
piece->completedBlocks.fill(false);
piece->requestedBlocks.fill(false);
d->callScheduler();
return;
}
foreach (TorrentPeer *peer, d->peers) {
if (!peer->interesting)
continue;
bool interesting = false;
for (int i = 0; i < d->pieceCount; ++i) {
if (peer->pieces.testBit(i) && d->incompletePieces.testBit(i)) {
interesting = true;
break;
}
}
peer->interesting = interesting;
}
delete piece;
d->pendingPieces.remove(pieceIndex);
d->completedPieces.setBit(pieceIndex);
d->incompletePieces.clearBit(pieceIndex);
foreach (PeerWireClient *client, d->connections) {
if (client->state() == QAbstractSocket::ConnectedState
&& !client->availablePieces().testBit(pieceIndex)) {
client->sendPieceNotification(pieceIndex);
}
}
int completed = d->completedPieces.count(true);
if (completed == d->pieceCount) {
if (d->state != Seeding) {
d->setState(Seeding);
d->trackerClient.startSeeding();
}
} else {
if (completed == 1)
d->setState(Downloading);
else if (d->incompletePieces.count(true) < 5 && d->pendingPieces.size() > d->incompletePieces.count(true))
d->setState(Endgame);
d->callScheduler();
}
updateProgress();
}
void TorrentClient::handleFileError()
{
if (d->state == Paused)
return;
setPaused(true);
emit error(FileError);
}
void TorrentClient::connectToPeers()
{
d->connectingToClients = false;
if (d->state == Stopping || d->state == Idle || d->state == Paused)
return;
if (d->state == Searching)
d->setState(Connecting);
QList<TorrentPeer *> weighedPeers = weighedFreePeers();
while (!weighedPeers.isEmpty() && ConnectionManager::instance()->canAddConnection()
&& (qrand() % (ConnectionManager::instance()->maxConnections() / 2))) {
PeerWireClient *client = new PeerWireClient(ConnectionManager::instance()->clientId(), this);
RateController::instance()->addSocket(client);
ConnectionManager::instance()->addConnection(client);
initializeConnection(client);
d->connections << client;
TorrentPeer *peer = weighedPeers.takeAt(qrand() % weighedPeers.size());
weighedPeers.removeAll(peer);
peer->connectStart = QDateTime::currentDateTime().toTime_t();
peer->lastVisited = peer->connectStart;
client->setPeer(peer);
client->connectToHost(peer->address, peer->port);
}
}
QList<TorrentPeer *> TorrentClient::weighedFreePeers() const
{
QList<TorrentPeer *> weighedPeers;
uint now = QDateTime::currentDateTime().toTime_t();
QList<TorrentPeer *> freePeers;
QMap<QString, int> connectionsPerPeer;
foreach (TorrentPeer *peer, d->peers) {
bool busy = false;
foreach (PeerWireClient *client, d->connections) {
if (client->state() == PeerWireClient::ConnectedState
&& client->peerAddress() == peer->address
&& client->peerPort() == peer->port) {
if (++connectionsPerPeer[peer->address.toString()] >= MaxConnectionPerPeer) {
busy = true;
break;
}
}
}
if (!busy && (now - peer->lastVisited) > uint(MinimumTimeBeforeRevisit))
freePeers << peer;
}
if (freePeers.isEmpty())
return weighedPeers;
QList<QPair<int, TorrentPeer *> > points;
foreach (TorrentPeer *peer, freePeers) {
int tmp = 0;
if (peer->interesting) {
tmp += peer->numCompletedPieces;
if (d->state == Seeding)
tmp = d->pieceCount - tmp;
if (!peer->connectStart)
tmp += d->pieceCount;
if (peer->connectTime < 5)
tmp += (d->pieceCount / 10) * (5 - peer->connectTime);
}
points << QPair<int, TorrentPeer *>(tmp, peer);
}
qSort(points);
typedef QPair<int,TorrentPeer*> PointPair;
QMultiMap<int, TorrentPeer *> pointMap;
int lowestScore = 0;
int lastIndex = 0;
foreach (PointPair point, points) {
if (point.first > lowestScore) {
lowestScore = point.first;
++lastIndex;
}
pointMap.insert(lastIndex, point.second);
}
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)
{
initializeConnection(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());
foreach (TorrentPeer *peer, d->peers) {
if (peer->port == client->peerPort() && peer->address == client->peerAddress()) {
peer->connectTime = peer->lastVisited - peer->connectStart;
break;
}
}
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(QBitArray)),
this, SLOT(peerPiecesAvailable(QBitArray)));
connect(client, SIGNAL(blockRequested(int,int,int)),
this, SLOT(peerRequestsBlock(int,int,int)));
connect(client, SIGNAL(blockReceived(int,int,QByteArray)),
this, SLOT(blockReceived(int,int,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());
if (client->peer() && client->error() == QAbstractSocket::ConnectionRefusedError)
d->peers.removeAll(client->peer());
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);
}
QMapIterator<int, PeerWireClient *> it2(d->readIds);
while (it2.findNext(client))
d->readIds.remove(it2.key());
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());
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 (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;
}
}
if (peer) {
peer->pieces = pieces;
peer->numCompletedPieces = pieces.count(true);
}
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());
if (client->peerWireState() & PeerWireClient::ChokingPeer)
return;
if (!d->completedPieces.testBit(pieceIndex))
return;
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;
}
int blockBit = begin / BlockSize;
TorrentPiece *piece = d->pendingPieces.value(pieceIndex);
if (!piece || piece->completedBlocks.testBit(blockBit)) {
requestMore(client);
return;
}
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);
d->fileManager.write(pieceIndex, begin, data);
piece->completedBlocks.setBit(blockBit);
piece->requestedBlocks.clearBit(blockBit);
if (blocksLeftForPiece(piece) == 0) {
d->fileManager.verifyPiece(piece->index);
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;
}
}
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()
{
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);
}
}
}
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 ((qrand() % 10) == 0)
client->abort();
else
client->chokePeer();
allClients.removeAll(client);
}
if (!interested)
allClients.removeAll(client);
}
if (!allClients.isEmpty()) {
PeerWireClient *client = allClients[qrand() % 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;
foreach (PeerWireClient *client, d->connections)
schedulePieceForClient(client);
}
void TorrentClient::schedulePieceForClient(PeerWireClient *client)
{
if (client->state() != QTcpSocket::ConnectedState)
return;
if (client->peerWireState() & PeerWireClient::ChokedByPeer)
return;
QList<int> currentPieces;
bool somePiecesAreNotInProgress = false;
TorrentPiece *lastPendingPiece = 0;
QMultiMap<PeerWireClient *, TorrentPiece *>::Iterator it = d->payloads.find(client);
while (it != d->payloads.end() && it.key() == client) {
lastPendingPiece = it.value();
if (lastPendingPiece->inProgress) {
currentPieces << lastPendingPiece->index;
} else {
somePiecesAreNotInProgress = true;
}
++it;
}
if (client->incomingBlocks().size() >= ((d->state == Endgame || d->state == WarmingUp)
? MaxBlocksInMultiMode : MaxBlocksInProgress))
return;
if (!somePiecesAreNotInProgress || client->incomingBlocks().size() > 0)
lastPendingPiece = 0;
TorrentPiece *piece = lastPendingPiece;
if (d->state == WarmingUp && d->pendingPieces.size() >= 4) {
piece = d->payloads.value(client);
if (!piece) {
QList<TorrentPiece *> values = d->pendingPieces.values();
piece = values.value(qrand() % values.size());
piece->inProgress = true;
d->payloads.insert(client, piece);
}
if (piece->completedBlocks.count(false) == client->incomingBlocks().size())
return;
}
if (!piece) {
QBitArray incompletePiecesAvailableToClient = d->incompletePieces;
if ((d->state == Endgame && client->uploadSpeed() < 1024) || d->state != WarmingUp) {
QMap<int, TorrentPiece *>::ConstIterator it = d->pendingPieces.constBegin();
while (it != d->pendingPieces.constEnd()) {
if (it.value()->inProgress)
incompletePiecesAvailableToClient.clearBit(it.key());
++it;
}
}
incompletePiecesAvailableToClient &= client->availablePieces();
foreach (int i, currentPieces)
incompletePiecesAvailableToClient.clearBit(i);
if (incompletePiecesAvailableToClient.count(true) == 0)
return;
QList<TorrentPiece *> partialPieces;
QMap<int, TorrentPiece *>::ConstIterator it = d->pendingPieces.constBegin();
while (it != d->pendingPieces.constEnd()) {
TorrentPiece *tmp = it.value();
if (incompletePiecesAvailableToClient.testBit(it.key())) {
if (!tmp->inProgress || d->state == WarmingUp || d->state == Endgame) {
partialPieces << tmp;
break;
}
}
++it;
}
if (!partialPieces.isEmpty())
piece = partialPieces.value(qrand() % partialPieces.size());
if (!piece) {
int pieceIndex = 0;
if (d->state == WarmingUp || (qrand() & 4) == 0) {
int *occurrances = new int[d->pieceCount];
memset(occurrances, 0, d->pieceCount * sizeof(int));
foreach (PeerWireClient *peer, d->connections) {
QBitArray peerPieces = peer->availablePieces();
int peerPiecesSize = peerPieces.size();
for (int i = 0; i < peerPiecesSize; ++i) {
if (peerPieces.testBit(i))
++occurrances[i];
}
}
int numOccurrances = d->state == WarmingUp ? 0 : 99999;
QList<int> piecesReadyForDownload;
for (int i = 0; i < d->pieceCount; ++i) {
if (d->state == WarmingUp) {
if (occurrances[i] >= numOccurrances
&& incompletePiecesAvailableToClient.testBit(i)) {
if (occurrances[i] > numOccurrances)
piecesReadyForDownload.clear();
piecesReadyForDownload.append(i);
numOccurrances = occurrances[i];
}
} else {
if (occurrances[i] <= numOccurrances
&& incompletePiecesAvailableToClient.testBit(i)) {
if (occurrances[i] < numOccurrances)
piecesReadyForDownload.clear();
piecesReadyForDownload.append(i);
numOccurrances = occurrances[i];
}
}
}
pieceIndex = piecesReadyForDownload.at(qrand() % piecesReadyForDownload.size());
delete [] occurrances;
} else {
QList<int> values;
int incompletePiecesAvailableToClientSize = incompletePiecesAvailableToClient.size();
for (int i = 0; i < incompletePiecesAvailableToClientSize; ++i) {
if (incompletePiecesAvailableToClient.testBit(i))
values << i;
}
pieceIndex = values.at(qrand() % values.size());
}
piece = new TorrentPiece;
piece->index = pieceIndex;
piece->length = d->fileManager.pieceLengthAt(pieceIndex);
int numBlocks = piece->length / BlockSize;
if (piece->length % BlockSize)
++numBlocks;
piece->completedBlocks.resize(numBlocks);
piece->requestedBlocks.resize(numBlocks);
d->pendingPieces.insert(pieceIndex, piece);
}
piece->inProgress = true;
d->payloads.insert(client, piece);
}
requestMore(client);
}
void TorrentClient::requestMore(PeerWireClient *client)
{
QMultiMap<PeerWireClient *, TorrentPiece *>::Iterator it = d->payloads.find(client);
int numBlocksInProgress = client->incomingBlocks().size();
QList<TorrentPiece *> piecesInProgress;
while (it != d->payloads.end() && it.key() == client) {
TorrentPiece *piece = it.value();
if (piece->inProgress || (d->state == WarmingUp || d->state == Endgame))
piecesInProgress << piece;
++it;
}
if (piecesInProgress.isEmpty() && d->incompletePieces.count(true)) {
d->callScheduler();
return;
}
int maxInProgress = ((d->state == Endgame || d->state == WarmingUp)
? MaxBlocksInMultiMode : MaxBlocksInProgress);
if (numBlocksInProgress == maxInProgress)
return;
foreach (TorrentPiece *piece, piecesInProgress) {
numBlocksInProgress += requestBlocks(client, piece, maxInProgress - numBlocksInProgress);
if (numBlocksInProgress == maxInProgress)
break;
}
if (numBlocksInProgress < maxInProgress && d->state != WarmingUp)
d->callScheduler();
}
int TorrentClient::requestBlocks(PeerWireClient *client, TorrentPiece *piece, int maxBlocks)
{
QVector<int> bits;
int completedBlocksSize = piece->completedBlocks.size();
for (int i = 0; i < completedBlocksSize; ++i) {
if (!piece->completedBlocks.testBit(i) && !piece->requestedBlocks.testBit(i))
bits << i;
}
if (bits.size() == 0) {
if (d->state != WarmingUp && d->state != Endgame)
return 0;
bits.clear();
for (int i = 0; i < completedBlocksSize; ++i) {
if (!piece->completedBlocks.testBit(i))
bits << i;
}
}
if (d->state == WarmingUp || d->state == Endgame) {
for (int i = 0; i < bits.size(); ++i) {
int a = qrand() % bits.size();
int b = qrand() % bits.size();
int tmp = bits[a];
bits[a] = bits[b];
bits[b] = tmp;
}
}
int blocksToRequest = qMin(maxBlocks, bits.size());
for (int i = 0; i < blocksToRequest; ++i) {
int blockSize = BlockSize;
if ((piece->length % BlockSize) && bits.at(i) == completedBlocksSize - 1)
blockSize = piece->length % BlockSize;
client->requestBlock(piece->index, bits.at(i) * BlockSize, blockSize);
piece->requestedBlocks.setBit(bits.at(i));
}
return blocksToRequest;
}
void TorrentClient::peerChoked()
{
PeerWireClient *client = qobject_cast<PeerWireClient *>(sender());
if (!client)
return;
QMultiMap<PeerWireClient *, TorrentPiece *>::Iterator it = d->payloads.find(client);
while (it != d->payloads.end() && it.key() == client) {
it.value()->inProgress = false;
it.value()->requestedBlocks.fill(false);
it = d->payloads.erase(it);
}
}
void TorrentClient::peerUnchoked()
{
PeerWireClient *client = qobject_cast<PeerWireClient *>(sender());
if (!client)
return;
if (d->state != Seeding)
d->callScheduler();
}
void TorrentClient::addToPeerList(const QList<TorrentPeer> &peerList)
{
QList<QHostAddress> addresses = QNetworkInterface::allAddresses();
foreach (TorrentPeer peer, peerList) {
if (addresses.contains(peer.address)
&& peer.port == TorrentServer::instance()->serverPort()) {
continue;
}
bool known = false;
foreach (TorrentPeer *knownPeer, d->peers) {
if (knownPeer->port == peer.port
&& knownPeer->address == peer.address) {
known = true;
break;
}
}
if (!known) {
TorrentPeer *newPeer = new TorrentPeer;
*newPeer = peer;
newPeer->interesting = true;
newPeer->seed = false;
newPeer->lastVisited = 0;
newPeer->connectStart = 0;
newPeer->connectTime = 999999;
newPeer->pieces.resize(d->pieceCount);
newPeer->numCompletedPieces = 0;
d->peers << newPeer;
}
}
int maxPeers = ConnectionManager::instance()->maxConnections() * 3;
if (d->peers.size() > maxPeers) {
QSet<TorrentPeer *> activePeers;
foreach (TorrentPeer *peer, d->peers) {
foreach (PeerWireClient *client, d->connections) {
if (client->peer() == peer && (client->downloadSpeed() + client->uploadSpeed()) > 1024)
activePeers << peer;
}
}
QList<int> toRemove;
for (int i = 0; i < d->peers.size() && (d->peers.size() - toRemove.size()) > maxPeers; ++i) {
if (!activePeers.contains(d->peers.at(i)))
toRemove << i;
}
QListIterator<int> toRemoveIterator(toRemove);
toRemoveIterator.toBack();
while (toRemoveIterator.hasPrevious())
d->peers.removeAt(toRemoveIterator.previous());
while (d->peers.size() > maxPeers)
d->peers.takeFirst();
}
if (d->state != Paused && d->state != Stopping && d->state != Idle) {
if (d->state == Searching || d->state == WarmingUp)
connectToPeers();
else
d->callPeerConnector();
}
}
void TorrentClient::trackerStopped()
{
d->setState(Idle);
emit stopped();
}
void TorrentClient::updateProgress(int progress)
{
if (progress == -1 && d->pieceCount > 0) {
int newProgress = (d->completedPieces.count(true) * 100) / d->pieceCount;
if (d->lastProgressValue != newProgress) {
d->lastProgressValue = newProgress;
emit progressUpdated(newProgress);
}
} else if (d->lastProgressValue != progress) {
d->lastProgressValue = progress;
emit progressUpdated(progress);
}
}
|
|
Best Of
Actualités les plus lues
Le blog Digia au hasard
Le blog Digia est l'endroit privilégié pour la communication sur l'édition commerciale de Qt, où des réponses publiques sont apportées aux questions les plus posées au support. Lire l'article.
Communauté
Ressources
Liens utiles
Contact
- Vous souhaitez rejoindre la rédaction ou proposer un tutoriel, une traduction, une question... ? Postez dans le forum Contribuez ou contactez-nous par MP ou par email (voir en bas de page).
Qt dans le magazine
|