#pragma once #include "tcp_connect.h" #include "pe...

#pragma once
#include "tcp_connect.h"
#include "peer.h"
#include "torrent_file.h"
#include "piece_storage.h"
#include <iostream>
#include <string>
#include <utility>
#include <cstring>
#include <arpa/inet.h>
#include "byte_tools.h"
#include "message.h"
/*

• Структура, хранящая информацию о доступности частей скачиваемого файла у данного пира
  /
  class PeerPiecesAvailability {
  public:
  PeerPiecesAvailability()
  : bitfield_() {}
  /

  • bitfield – массив байтов, в котором i-й бит означает наличие или отсутствие i-й части файла у пира
  • https://wiki.theory.org/BitTorrentSpecification#bitfield:_.3Clen.3D0001.2BX.3E.3Cid.3D5.3E.3Cbitfield.3E
    /
    explicit PeerPiecesAvailability(std::string bitfield)
    : bitfield_(std::move(bitfield)) {}
    /
  • Если ли часть под номером pieceIndex у пира?
    /
    bool IsPieceAvailable(size_t pieceIndex) const {
    size_t ind = pieceIndex / 8;
    if (ind >= bitfield_.size()) {
    return false;
    }
    uint8_t m = 1 << (7 - (pieceIndex % 8));
    return (static_cast<uint8_t>(bitfield_[ind]) & m) != 0;
    }
    /
  • Пометить часть под номером pieceIndex как доступную
    /
    void SetPieceAvailability(size_t pieceIndex) {
    size_t ind = pieceIndex / 8;
    if (ind >= bitfield_.size()) {
    bitfield_.resize(ind + 1, '\0');
    }
    uint8_t m = 1 << (7 - (pieceIndex % 8));
    bitfield_[ind] = static_cast<char>(static_cast<uint8_t>(bitfield_[ind]) | m);
    }
    void SetAllPiecesAvailable(size_t piecesCount) {
    if (piecesCount == 0) {
    bitfield_.clear();
    return;
    }
    bitfield_.assign((piecesCount + 7) / 8, static_cast<char>(0xff));
    }
    /
  • Сколько бит хранится в bitfield'е
    /
    size_t Size() const {
    return bitfield_.size() * 8;
    }
    private:
    std::string bitfield_;
    };
    /

• Класс, представляющий соединение с одним пиром.

• С помощью него можно подключиться к пиру и обмениваться с ним сообщениями
  /
  class PeerConnect {
  public:
  PeerConnect(const Peer& peer, const TorrentFile& tf, std::string selfPeerId, PieceStorage& pieceStorage)
  : tf_(tf)
  , socket_(peer.ip, peer.port, std::chrono::milliseconds(500), std::chrono::milliseconds(500))
  , selfPeerId_(std::move(selfPeerId))
  , terminated_(false)
  , choked_(true)
  , pieceStorage_(pieceStorage)
  , pieceInProgress_(nullptr)
  , pendingBlock_(false){
  }
  ~PeerConnect() {
  Terminate();
  socket_.CloseConnection();
  }
  /

  • Основная функция, в которой будет происходить цикл общения с пиром.
  • https://wiki.theory.org/BitTorrentSpecification#Messages
    /
    void Run();
    void Terminate() {
    terminated_ = true;
    }
    protected:
    const TorrentFile& tf_;
    TcpConnect socket_;
    const std::string selfPeerId_;
    std::string peerId_;
    PeerPiecesAvailability piecesAvailability_;
    bool terminated_;
    bool choked_;
    PiecePtr pieceInProgress_;
    PieceStorage& pieceStorage_;
    bool pendingBlock_;
    /
  • Функция производит handshake.
  • • Подключиться к пиру по протоколу TCP
  • • Отправить пиру сообщение handshake
  • • Проверить правильность ответа пира
  • https://wiki.theory.org/BitTorrentSpecification#Handshake
    /
    void PerformHandshake() {
    socket_.EstablishConnection();
    std::string handshake;
    handshake.push_back(19);
    handshake += "BitTorrent protocol";
    handshake.append(8, '\0');
    handshake += tf_.infoHash;
    handshake += selfPeerId_;
    socket_.SendData(handshake);
    std::string response = socket_.ReceiveData(68);
    if (response.size() != 68) {
    throw std::runtime_error("Wrong format");
    }
    if (static_cast<unsigned char>(response[0]) != 19) {
    throw std::runtime_error("Wrong protocol length");
    }
    std::string protocol = response.substr(1, 19);
    if (protocol != "BitTorrent protocol") {
    throw std::runtime_error("Invalid protocol name");
    }
    std::string infoHash = response.substr(28, 20);
    if (infoHash != tf_.infoHash) {
    throw std::runtime_error("Invalid info hash");
    }
    peerId_ = response.substr(48, 20);
    }
    /
  • • Провести handshake
  • • Получить bitfield с информацией о наличии у пира различных частей файла
  • • Сообщить пиру, что мы готовы получать от него данные (отправить interested)
  • Возвращает true, если все три этапа прошли без ошибок
    /
    bool EstablishConnection() {
    try {
    PerformHandshake();
    piecesAvailability_.SetAllPiecesAvailable(tf_.pieceHashes.size());
    ReceiveBitfield();
    SendInterested();
    return true;
    } catch (...) {
    socket_.CloseConnection();
    return false;
    }
    }
    /
  • Функция читает из сокета bitfield с информацией о наличии у пира различных частей файла.
  • Полученную информацию надо сохранить в поле piecesAvailability_.
  • Также надо учесть, что сообщение тип Bitfield является опциональным, то есть пиры необязательно будут слать его.
  • Вместо этого они могут сразу прислать сообщение Unchoke, поэтому надо быть готовым обработать его в этой функции.
  • Обработка сообщения Unchoke заключается в выставлении флага choked_ в значение false
    /
    void ReceiveBitfield() {
    try {
    std::string messageString = socket_.ReceiveData();
    Message message = Message::Parse(messageString);
    if (message.id == MessageId::BitField) {
    piecesAvailability_ = PeerPiecesAvailability(message.payload);
    } else if (message.id == MessageId::Unchoke) {
    choked_ = false;
    } else if (message.id == MessageId::Have) {
    if (message.payload.size() == 4) {
    piecesAvailability_.SetPieceAvailability(BytesToInt(message.payload));
    }
    } else if (message.id == MessageId::Choke) {
    choked_ = true;
    pendingBlock_ = false;
    if (pieceInProgress_ != nullptr) {
    pieceInProgress_->Reset();
    pieceInProgress_ = nullptr;
    }
    } else if (message.id == MessageId::Piece) {
    pendingBlock_ = false;
    if (pieceInProgress_ == nullptr || message.payload.size() < 8) {
    return;
    }
    uint32_t pieceIndex = BytesToInt(message.payload.substr(0, 4)), offset = BytesToInt(message.payload.substr(4, 4));
    std::string data = message.payload.substr(8);
    if (pieceIndex != pieceInProgress_->GetIndex()) {
    return;
    }
    pieceInProgress_->SaveBlock(offset, data);
    if (pieceInProgress_->AllBlocksRetrieved()) {
    if (pieceInProgress_->HashMatches()) {
    pieceStorage_.PieceProcessed(pieceInProgress_);
    pieceInProgress_ = nullptr;
    } else {
    pieceInProgress_->Reset();
    }
    }
    }
    } catch (...) {
    return;
    }
    }
    /
  • Функция посылает пиру сообщение типа interested
    /
    void SendInterested() {
    Message m = Message::Init(MessageId::Interested, "");
    socket_.SendData(m.ToString());
    }
    /
  • Функция отправляет пиру сообщение типа request. Это сообщение обозначает запрос части файла у пира.
  • За одно сообщение запрашивается не часть целиком, а блок данных размером 2^14 байт или меньше.
  • Если в данный момент мы не знаем, какую часть файла надо запросить у пира, то надо получить эту информацию у
  • PieceStorage
    /
    void RequestPiece() {
    if (pieceStorage_.QueueIsEmpty()) {
    Terminate();
    return;
    }
    if (pieceInProgress_ == nullptr) {
    pieceInProgress_ = pieceStorage_.GetNextPieceToDownload();
    }
    if (!piecesAvailability_.IsPieceAvailable(pieceInProgress_->GetIndex())) {
    Terminate();
    return;
    }
    Block block = pieceInProgress_->FirstMissingBlock();
    block->status = Block::Status::Pending;
    char m[17];
    uint32_t net_len = htonl(13);
    memcpy(m, &net_len, 4);
    m[4] = 6;
    uint32_t ind = static_cast<uint32_t>(pieceInProgress_->GetIndex());
    uint32_t piece = htonl(ind);
    memcpy(m + 5, &piece, 4);
    uint32_t offset = htonl(block->offset);
    memcpy(m + 9, &offset, 4);
    uint32_t len = htonl(block->length);
    memcpy(m + 13, &len, 4);
    socket_.SendData(std::string(m, 17));
    pendingBlock_ = true;
    }

  /*

  • Основной цикл общения с пиром. Здесь мы ждем следующее сообщение от пира и обрабатываем его.
  • Также, если мы не ждем в данный момент от пира содержимого части файла, то надо отправить соответствующий запрос
    /
    void MainLoop() {
    while (!terminated_ && !pieceStorage_.QueueIsEmpty()) {
    ReceiveBitfield();
    if (!choked_ && !pendingBlock_ && !pieceStorage_.QueueIsEmpty()) {
    RequestPiece();
    }
    }
    }
    };
    #pragma once
    #include "torrent_file.h"
    #include "piece.h"
    #include <queue>
    #include <string>
    #include <unordered_set>
    #include <mutex>
    #include <stdexcept>
    #include <fstream>
    /

• Хранилище информации о частях скачиваемого файла.

• В этом классе отслеживается информация о том, какие части файла осталось скачать
  /
  class PieceStorage {
  public:
  explicit PieceStorage(const TorrentFile& tf) : totalPieces_(tf.pieceHashes.size()),
  outputFile_(tf.name, std::ios::binary) {
  for (int i = 0; i < tf.pieceHashes.size(); ++i) {
  size_t length = tf.pieceLength;
  if (i == tf.pieceHashes.size() - 1 && tf.length % tf.pieceLength != 0) {
  length = tf.length % tf.pieceLength;
  }
  auto piece = std::make_shared<Piece>(i, length, tf.pieceHashes[i]);
  remainPieces_.push(piece);
  }
  }
  /

  • Отдает указатель на следующую часть файла, которую надо скачать
    /
    PiecePtr GetNextPieceToDownload() {
    std::unique_lock lock(mt_);
    if (remainPieces_.empty()) {
    throw std::runtime_error("Queue is empty");
    }
    auto piece = remainPieces_.front();
    remainPieces_.pop();
    ++pieceInProgress_;
    return piece;
    }
    /
  • Эта функция вызывается из PeerConnect, когда скачивание одной части файла завершено.
  • В рамках данного задания требуется очистить очередь частей для скачивания как только хотя бы одна часть будет успешно скачана.
    /
    void PieceProcessed(const PiecePtr& piece) {
    std::unique_lock lock(mt_);
    --pieceInProgress_;
    SavePieceToDisk(piece);
    pieces_.push_back(piece->GetIndex());
    while (!remainPieces_.empty()) {
    remainPieces_.pop();
    }
    }
    /
  • Остались ли нескачанные части файла?
    /
    bool QueueIsEmpty() const {
    std::unique_lock lock(mt_);
    return remainPieces_.empty();
    }
    /
  • Сколько частей файла всего
    /
    size_t TotalPiecesCount() const {
    std::unique_lock lock(mt_);
    return totalPieces_;
    }
    /
  • Сколько частей файла было сохранено на диск
    */
    size_t PiecesSavedToDiscCount() const {
    std::unique_lock lock(mt_);
    return pieces_.size();
    }

  /*

  • Закрыть поток вывода в файл
    */
    void CloseOutputFile() {
    std::unique_lock lock(mt_);
    if (outputFile_.is_open()) {
    outputFile_.close();
    }
    }

  /*

  • Отдает список номеров частей файла, которые были сохранены на диск
    */
    std::vector<size_t> GetPiecesSavedToDiscIndices() const {
    std::unique_lock lock(mt_);
    return pieces_;
    }

  /*

  • Сколько частей файла в данный момент скачивается
    /
    size_t PiecesInProgressCount() const {
    std::unique_lock lock(mt_);
    return pieceInProgress_;
    }
    protected:
    std::queue<PiecePtr> remainPieces_;
    std::vector <size_t> pieces_;
    virtual void SavePieceToDisk(PiecePtr piece) {
    outputFile_.seekp(piece->GetIndex() * (1 << 14));
    const auto& data = piece->GetData();
    outputFile_.write(reinterpret_cast<const char>(data.data()), data.size());
    }
    long long totalPieces_ = 0, pieceInProgress_ = 0;
    mutable std::mutex mt_;
    std::ofstream outputFile_;

};
#pragma once
#include <string>
#include <vector>
#include <optional>
#include <memory>
#include <stdexcept>
#include "byte_tools.h"
/*

• Части файла скачиваются не за одно сообщение, а блоками размером 2^14 байт или меньше (последний блок обычно меньше)
  /
  struct Block {
  enum Status {
  Missing = 0,
  Pending,
  Retrieved,
  };
  uint32_t piece;
  uint32_t offset;
  uint32_t length;
  Status status;
  std::string data;
  };
  /
• Часть скачиваемого файла
  /
  class Piece {
  public:
  /
  • index -- номер части файла, нумерация начинается с 0
  • length -- длина части файла. Все части, кроме последней, имеют длину, равную torrentFile.pieceLength
  • hash -- хеш-сумма части файла, взятая из torrentFile.pieceHashes
    /
    Piece(size_t index, size_t length, std::string hash) : index_(index), length_(length), hash_(hash), numBlocks_(length / (1 << 14) + (length % (1 << 14) != 0)) {
    for (long long i = 0; i < length / (1 << 14); ++i) {
    blocks_.push_back(Block{static_cast<uint32_t>(index), static_cast<uint32_t>(i * (1 << 14)), static_cast<uint32_t>(1 << 14), Block::Status::Missing});
    }
    if (length % (1 << 14) != 0) {
    long long lastBlock = length / (1 << 14);
    long long lenLastBlock = length % (1 << 14);
    blocks_.push_back(Block{static_cast<uint32_t>(index), static_cast<uint32_t>(lastBlock * (1 << 14)), static_cast<uint32_t>(lenLastBlock), Block::Status::Missing});
    }
    }
    /
  • Совпадает ли хеш скачанных данных с ожидаемым
    /
    bool HashMatches() const {
    return hash_ == GetDataHash();
    }
    /
  • Дать указатель на отсутствующий (еще не скачанный и не запрошенный) блок
    /
    Block FirstMissingBlock() {
    if (AllBlocksRetrieved()) throw std::runtime_error("All blocks retrieved");
    for (long long i = lastUnsavedBlock_; i < numBlocks_; ++i) {
    if (blocks_[i].status == Block::Status::Missing) {
    lastUnsavedBlock_ = i;
    return &blocks_[i];
    }
    }
    throw std::runtime_error("No missing blocks");
    }
    /*
  • Получить порядковый номер части файла
    /
    size_t GetIndex() const {
    return index_;
    }
    /
  • Сохранить скачанные данные для какого-то блока
    /
    void SaveBlock(size_t blockOffset, std::string data) {
    long long blockIndex = blockOffset / (1 << 14);
    if (blockIndex >= numBlocks_) {
    throw std::runtime_error("Wrong block offset");
    }
    blocks_[blockIndex].data = data;
    if (blocks_[blockIndex].status != Block::Status::Retrieved) {
    ++savedBlocks_;
    }
    blocks_[blockIndex].status = Block::Status::Retrieved;
    if (blockIndex == lastUnsavedBlock_) {
    for (long long i = lastUnsavedBlock_ + 1; i < numBlocks_; ++i) {
    if (blocks_[i].status == Block::Status::Missing) {
    lastUnsavedBlock_ = i;
    return;
    }
    }
    }
    }
    /
  • Скачали ли уже все блоки
    /
    bool AllBlocksRetrieved() const {
    return savedBlocks_ == numBlocks_;
    }
    /
  • Получить скачанные данные для части файла
    /
    std::string GetData() const {
    std::string fullData(length_, '\0');
    for (const auto& block : blocks_) {
    if (block.status == Block::Status::Retrieved) {
    fullData.replace(block.offset, block.length, block.data);
    }
    }
    return fullData;
    }
    /
  • Посчитать хеш по скачанным данным
    /
    std::string GetDataHash() const {
    return CalculateSHA1(GetData());
    }
    /
  • Получить хеш для части из .torrent файла
    /
    const std::string& GetHash() const {
    return hash_;
    }
    /
  • Удалить все скачанные данные и отметить все блоки как Missing
    */
    void Reset() {
    for (long long i = 0; i < numBlocks_; ++i) {
    blocks_[i].data.clear();
    blocks_[i].status = Block::Status::Missing;
    }
    savedBlocks_ = 0;
    lastUnsavedBlock_ = 0;
    }
    private:
    const size_t index_, length_;
    const std::string hash_;
    std::vector<Block> blocks_;
    long long numBlocks_ = 0, savedBlocks_ = 0, lastUnsavedBlock_ = 0;
    };
    using PiecePtr = std::shared_ptr<Piece>;
    #include "torrent_tracker.h"
    #include "piece_storage.h"
    #include "peer_connect.h"
    #include "byte_tools.h"
    #include <cassert>
    #include <iostream>
    #include <filesystem>
    #include <random>
    #include <thread>
    #include <algorithm>

namespace fs = std::filesystem;

std::mutex cerrMutex, coutMutex;

std::string RandomString(size_t length) {
std::random_device random;
std::string result;
result.reserve(length);
for (size_t i = 0; i < length; ++i) {
result.push_back(random() % ('Z' - 'A') + 'A');
}
return result;
}

const std::string PeerId = "TESTAPPDONTWORRY" + RandomString(4);
constexpr size_t PiecesToDownload = 20;

void CheckDownloadedPiecesIntegrity(const std::filesystem::path& outputFilename, const TorrentFile& tf, PieceStorage& pieces) {
pieces.CloseOutputFile();

text
if (std::filesystem::file_size(outputFilename) != tf.length) {
    throw std::runtime_error("Output file has wrong size");
}

if (pieces.GetPiecesSavedToDiscIndices().size() != pieces.PiecesSavedToDiscCount()) {
    throw std::runtime_error("Cannot determine real amount of saved pieces");
}

if (pieces.PiecesSavedToDiscCount() < PiecesToDownload) {
    throw std::runtime_error("Downloaded pieces amount is not enough");
}

if (pieces.TotalPiecesCount() != tf.pieceHashes.size() || pieces.TotalPiecesCount() < 200) {
    throw std::runtime_error("Wrong amount of pieces");
}

std::vector<size_t> pieceIndices = pieces.GetPiecesSavedToDiscIndices();
std::sort(pieceIndices.begin(), pieceIndices.end());

std::ifstream file(outputFilename, std::ios_base::binary);
for (size_t pieceIndex : pieceIndices) {
    const std::streamoff positionInFile = pieceIndex * tf.pieceLength;
    file.seekg(positionInFile);
    if (!file.good()) {
        throw std::runtime_error("Cannot read from file");
    }
    std::string pieceDataFromFile(tf.pieceLength, '\0');
    file.read(pieceDataFromFile.data(), tf.pieceLength);
    const size_t readBytesCount = file.gcount();
    pieceDataFromFile.resize(readBytesCount);
    const std::string realHash = CalculateSHA1(pieceDataFromFile);

    if (realHash != tf.pieceHashes[pieceIndex]) {
        std::cerr << "File piece with index " << pieceIndex << " started at position " << positionInFile <<
                  " with length " << pieceDataFromFile.length() << " has wrong hash " << HexEncode(realHash) <<
                  ". Expected hash is " << HexEncode(tf.pieceHashes[pieceIndex]) << std::endl;
        throw std::runtime_error("Wrong piece hash");
    }
}

}

void DeleteDownloadedFile(const std::filesystem::path& outputFilename) {
std::filesystem::remove(outputFilename);
}

std::filesystem::path PrepareDownloadDirectory(const std::string& randomString) {
std::filesystem::path outputDirectory = "/tmp/downloads";
outputDirectory /= randomString;
std::filesystem::create_directories(outputDirectory);
return outputDirectory;
}

bool RunDownloadMultithread(PieceStorage& pieces, const TorrentFile& torrentFile, const std::string& ourId, const TorrentTracker& tracker) {
using namespace std::chrono_literals;

text
std::vector<std::thread> peerThreads;
std::vector<std::shared_ptr<PeerConnect>> peerConnections;

for (const Peer& peer : tracker.GetPeers()) {
    peerConnections.emplace_back(std::make_shared<PeerConnect>(peer, torrentFile, ourId, pieces));
}

peerThreads.reserve(peerConnections.size());

for (auto& peerConnectPtr : peerConnections) {
    peerThreads.emplace_back(
            [peerConnectPtr] () {
                bool tryAgain = true;
                int attempts = 0;
                do {
                    try {
                        ++attempts;
                        peerConnectPtr->Run();
                    } catch (const std::runtime_error& e) {
                        std::lock_guard<std::mutex> cerrLock(cerrMutex);
                        std::cerr << "Runtime error: " << e.what() << std::endl;
                    } catch (const std::exception& e) {
                        std::lock_guard<std::mutex> cerrLock(cerrMutex);
                        std::cerr << "Exception: " << e.what() << std::endl;
                    } catch (...) {
                        std::lock_guard<std::mutex> cerrLock(cerrMutex);
                        std::cerr << "Unknown error" << std::endl;
                    }
                    tryAgain = peerConnectPtr->Failed() && attempts < 3;
                } while (tryAgain);
            }
    );
}

{
    std::lock_guard<std::mutex> coutLock(coutMutex);
    std::cout << "Started " << peerThreads.size() << " threads for peers" << std::endl;
}

std::this_thread::sleep_for(10s);
while (pieces.PiecesSavedToDiscCount() < PiecesToDownload) {
    if (pieces.PiecesInProgressCount() == 0) {
        {
            std::lock_guard<std::mutex> coutLock(coutMutex);
            std::cout
                    << "Want to download more pieces but all peer connections are not working. Let's request new peers"
                    << std::endl;
        }

        for (auto& peerConnectPtr : peerConnections) {
            peerConnectPtr->Terminate();
        }
        for (std::thread& thread : peerThreads) {
            thread.join();
        }
        return true;
    }
    std::this_thread::sleep_for(1s);
}

{
    std::lock_guard<std::mutex> coutLock(coutMutex);
    std::cout << "Terminating all peer connections" << std::endl;
}
for (auto& peerConnectPtr : peerConnections) {
    peerConnectPtr->Terminate();
}

for (std::thread& thread : peerThreads) {
    thread.join();
}

return false;

}

void DownloadTorrentFile(const TorrentFile& torrentFile, PieceStorage& pieces, const std::string& ourId) {
std::cout << "Connecting to tracker " << torrentFile.announce << std::endl;
TorrentTracker tracker(torrentFile.announce);
bool requestMorePeers = false;
do {
tracker.UpdatePeers(torrentFile, ourId, 12345);

text
    if (tracker.GetPeers().empty()) {
        std::cerr << "No peers found. Cannot download a file" << std::endl;
    }

    std::cout << "Found " << tracker.GetPeers().size() << " peers" << std::endl;
    for (const Peer& peer : tracker.GetPeers()) {
        std::cout << "Found peer " << peer.ip << ":" << peer.port << std::endl;
    }

    requestMorePeers = RunDownloadMultithread(pieces, torrentFile, ourId, tracker);
} while (requestMorePeers);

}

void TestTorrentFile(const fs::path& file) {
TorrentFile torrentFile;
try {
torrentFile = LoadTorrentFile(file);
std::cout << "Loaded torrent file " << file << ". " << torrentFile.comment << std::endl;
} catch (const std::invalid_argument& e) {
std::cerr << e.what() << std::endl;
return;
}

text
const std::filesystem::path outputDirectory = PrepareDownloadDirectory(PeerId);
PieceStorage pieces(torrentFile, outputDirectory);

DownloadTorrentFile(torrentFile, pieces, PeerId);

CheckDownloadedPiecesIntegrity(outputDirectory / torrentFile.name, torrentFile, pieces);
DeleteDownloadedFile(outputDirectory / torrentFile.name);

}

int main() {
for (const auto& entry : fs::directory_iterator("resources")) {
TestTorrentFile(entry.path());
}
return 0;
}
найди ошибки и исправь минимально. стиль кода не меняй! это ВАжно! ТЕ ЖЕ названия переменных, функций, библиотеки, без проспуков и пустых строк.
дай точечные исправления (что на что поменять) с комментариями что не верно у меня
дай готовый верный код В МОЕМ СТИЛЕ