/* */ #include "DefaultPieceStorage.h" #include #include #include "DownloadContext.h" #include "Piece.h" #include "Peer.h" #include "LogFactory.h" #include "Logger.h" #include "prefs.h" #include "DirectDiskAdaptor.h" #include "MultiDiskAdaptor.h" #include "DiskWriter.h" #include "BitfieldManFactory.h" #include "BitfieldMan.h" #include "message.h" #include "DefaultDiskWriterFactory.h" #include "FileEntry.h" #include "DlAbortEx.h" #include "Util.h" #include "a2functional.h" #include "Option.h" #include "StringFormat.h" #include "RarestPieceSelector.h" namespace aria2 { DefaultPieceStorage::DefaultPieceStorage (const DownloadContextHandle& downloadContext, const Option* option, const SharedHandle& pieceSelector): downloadContext(downloadContext), bitfieldMan(BitfieldManFactory::getFactoryInstance()-> createBitfieldMan(downloadContext->getPieceLength(), downloadContext->getTotalLength())), _diskWriterFactory(new DefaultDiskWriterFactory()), endGamePieceNum(END_GAME_PIECE_NUM), logger(LogFactory::getInstance()), option(option), _pieceSelector(pieceSelector) { if(_pieceSelector.isNull()) { _pieceSelector.reset (new RarestPieceSelector(downloadContext->getNumPieces(), true)); } } DefaultPieceStorage::~DefaultPieceStorage() { delete bitfieldMan; } bool DefaultPieceStorage::hasMissingPiece(const PeerHandle& peer) { return bitfieldMan->hasMissingPiece(peer->getBitfield(), peer->getBitfieldLength()); } bool DefaultPieceStorage::isEndGame() { return bitfieldMan->countMissingBlock() <= endGamePieceNum; } bool DefaultPieceStorage::getMissingPieceIndex(size_t& index, const unsigned char* bitfield, size_t& length) { std::deque indexes; bool r; if(isEndGame()) { r = bitfieldMan->getAllMissingIndexes(indexes, bitfield, length); } else { r = bitfieldMan->getAllMissingUnusedIndexes(indexes, bitfield, length); } if(r) { // We assume indexes is sorted using comparator less. return _pieceSelector->select(index, indexes); } else { return false; } } PieceHandle DefaultPieceStorage::checkOutPiece(size_t index) { bitfieldMan->setUseBit(index); PieceHandle piece = findUsedPiece(index); if(piece.isNull()) { piece.reset(new Piece(index, bitfieldMan->getBlockLength(index))); #ifdef ENABLE_MESSAGE_DIGEST piece->setHashAlgo(downloadContext->getPieceHashAlgo()); #endif // ENABLE_MESSAGE_DIGEST addUsedPiece(piece); return piece; } else { return piece; } } /** * Newly instantiated piece is not added to usedPieces. * Because it is waste of memory and there is no chance to use them later. */ PieceHandle DefaultPieceStorage::getPiece(size_t index) { PieceHandle piece; if(0 <= index && index <= bitfieldMan->getMaxIndex()) { piece = findUsedPiece(index); if(piece.isNull()) { piece.reset(new Piece(index, bitfieldMan->getBlockLength(index))); if(hasPiece(index)) { piece->setAllBlock(); } } } return piece; } void DefaultPieceStorage::addUsedPiece(const PieceHandle& piece) { std::deque >::iterator i = std::lower_bound(usedPieces.begin(), usedPieces.end(), piece); usedPieces.insert(i, piece); logger->debug("usedPieces.size()=%lu", static_cast(usedPieces.size())); } PieceHandle DefaultPieceStorage::findUsedPiece(size_t index) const { SharedHandle p(new Piece()); p->setIndex(index); std::deque >::const_iterator i = std::lower_bound(usedPieces.begin(), usedPieces.end(), p); if(i != usedPieces.end() && (*i) == p) { return *i; } else { p.reset(0); return p; } } SharedHandle DefaultPieceStorage::getMissingPiece (const unsigned char* bitfield, size_t length) { size_t index; if(getMissingPieceIndex(index, bitfield, length)) { return checkOutPiece(index); } else { return SharedHandle(); } } SharedHandle DefaultPieceStorage::getMissingPiece (const BitfieldMan& bitfield) { return getMissingPiece(bitfield.getBitfield(), bitfield.getBitfieldLength()); } PieceHandle DefaultPieceStorage::getMissingPiece(const SharedHandle& peer) { return getMissingPiece(peer->getBitfield(), peer->getBitfieldLength()); } void DefaultPieceStorage::createFastIndexBitfield (BitfieldMan& bitfield, const SharedHandle& peer) { for(std::deque::const_iterator itr = peer->getPeerAllowedIndexSet().begin(); itr != peer->getPeerAllowedIndexSet().end(); ++itr) { if(!bitfieldMan->isBitSet(*itr) && peer->hasPiece(*itr)) { bitfield.setBit(*itr); } } } PieceHandle DefaultPieceStorage::getMissingFastPiece (const SharedHandle& peer) { if(peer->isFastExtensionEnabled() && peer->countPeerAllowedIndexSet() > 0) { BitfieldMan tempBitfield(bitfieldMan->getBlockLength(), bitfieldMan->getTotalLength()); createFastIndexBitfield(tempBitfield, peer); return getMissingPiece(tempBitfield); } else { return SharedHandle(); } } static void unsetExcludedIndexes(BitfieldMan& bitfield, const std::deque& excludedIndexes) { for(std::deque::const_iterator i = excludedIndexes.begin(); i != excludedIndexes.end(); ++i) { bitfield.unsetBit(*i); } } SharedHandle DefaultPieceStorage::getMissingPiece (const SharedHandle& peer, const std::deque& excludedIndexes) { BitfieldMan tempBitfield(bitfieldMan->getBlockLength(), bitfieldMan->getTotalLength()); tempBitfield.setBitfield(peer->getBitfield(), peer->getBitfieldLength()); unsetExcludedIndexes(tempBitfield, excludedIndexes); return getMissingPiece(tempBitfield); } SharedHandle DefaultPieceStorage::getMissingFastPiece (const SharedHandle& peer, const std::deque& excludedIndexes) { if(peer->isFastExtensionEnabled() && peer->countPeerAllowedIndexSet() > 0) { BitfieldMan tempBitfield(bitfieldMan->getBlockLength(), bitfieldMan->getTotalLength()); createFastIndexBitfield(tempBitfield, peer); unsetExcludedIndexes(tempBitfield, excludedIndexes); return getMissingPiece(tempBitfield); } else { return SharedHandle(); } } PieceHandle DefaultPieceStorage::getMissingPiece() { size_t index; if(bitfieldMan->getSparseMissingUnusedIndex(index)) { return checkOutPiece(index); } else { return SharedHandle(); } } PieceHandle DefaultPieceStorage::getMissingPiece(size_t index) { if(hasPiece(index) || isPieceUsed(index)) { return SharedHandle(); } else { return checkOutPiece(index); } } void DefaultPieceStorage::deleteUsedPiece(const PieceHandle& piece) { if(piece.isNull()) { return; } std::deque >::iterator i = std::lower_bound(usedPieces.begin(), usedPieces.end(), piece); if(i != usedPieces.end() && (*i) == piece) { usedPieces.erase(i); } } // void DefaultPieceStorage::reduceUsedPieces(size_t upperBound) // { // size_t usedPiecesSize = usedPieces.size(); // if(usedPiecesSize <= upperBound) { // return; // } // size_t delNum = usedPiecesSize-upperBound; // int fillRate = 10; // while(delNum && fillRate <= 15) { // delNum -= deleteUsedPiecesByFillRate(fillRate, delNum); // fillRate += 5; // } // } // size_t DefaultPieceStorage::deleteUsedPiecesByFillRate(int fillRate, // size_t delNum) // { // size_t deleted = 0; // for(Pieces::iterator itr = usedPieces.begin(); // itr != usedPieces.end() && deleted < delNum;) { // PieceHandle& piece = *itr; // if(!bitfieldMan->isUseBitSet(piece->getIndex()) && // piece->countCompleteBlock() <= piece->countBlock()*(fillRate/100.0)) { // logger->info(MSG_DELETING_USED_PIECE, // piece->getIndex(), // (piece->countCompleteBlock()*100)/piece->countBlock(), // fillRate); // itr = usedPieces.erase(itr); // ++deleted; // } else { // ++itr; // } // } // return deleted; // } void DefaultPieceStorage::completePiece(const PieceHandle& piece) { if(piece.isNull()) { return; } deleteUsedPiece(piece); // if(!isEndGame()) { // reduceUsedPieces(100); // } if(allDownloadFinished()) { return; } bitfieldMan->setBit(piece->getIndex()); bitfieldMan->unsetUseBit(piece->getIndex()); addPieceStats(piece->getIndex()); if(downloadFinished()) { downloadContext->resetDownloadStopTime(); diskAdaptor->onDownloadComplete(); if(isSelectiveDownloadingMode()) { logger->notice(MSG_SELECTIVE_DOWNLOAD_COMPLETED); // following line was commented out in order to stop sending request // message after user-specified files were downloaded. //finishSelectiveDownloadingMode(); } else { logger->info(MSG_DOWNLOAD_COMPLETED); } } } bool DefaultPieceStorage::isSelectiveDownloadingMode() { return bitfieldMan->isFilterEnabled(); } void DefaultPieceStorage::finishSelectiveDownloadingMode() { bitfieldMan->clearFilter(); diskAdaptor->addAllDownloadEntry(); } // not unittested void DefaultPieceStorage::cancelPiece(const PieceHandle& piece) { if(piece.isNull()) { return; } bitfieldMan->unsetUseBit(piece->getIndex()); if(!isEndGame()) { if(piece->getCompletedLength() == 0) { deleteUsedPiece(piece); } } } bool DefaultPieceStorage::hasPiece(size_t index) { return bitfieldMan->isBitSet(index); } bool DefaultPieceStorage::isPieceUsed(size_t index) { return bitfieldMan->isUseBitSet(index); } uint64_t DefaultPieceStorage::getTotalLength() { return bitfieldMan->getTotalLength(); } uint64_t DefaultPieceStorage::getFilteredTotalLength() { return bitfieldMan->getFilteredTotalLength(); } uint64_t DefaultPieceStorage::getCompletedLength() { uint64_t completedLength = bitfieldMan->getCompletedLength()+getInFlightPieceCompletedLength(); uint64_t totalLength = getTotalLength(); if(completedLength > totalLength) { completedLength = totalLength; } return completedLength; } uint64_t DefaultPieceStorage::getFilteredCompletedLength() { return bitfieldMan->getFilteredCompletedLength()+getInFlightPieceCompletedLength(); } size_t DefaultPieceStorage::getInFlightPieceCompletedLength() const { return std::accumulate(usedPieces.begin(), usedPieces.end(), 0, adopt2nd(std::plus(), mem_fun_sh(&Piece::getCompletedLength))); } // not unittested void DefaultPieceStorage::setFileFilter(const std::deque& filePaths) { if(downloadContext->getFileMode() != DownloadContext::MULTI || filePaths.empty()) { return; } diskAdaptor->removeAllDownloadEntry(); for(std::deque::const_iterator pitr = filePaths.begin(); pitr != filePaths.end(); pitr++) { if(!diskAdaptor->addDownloadEntry(*pitr)) { throw DlAbortEx(StringFormat(EX_NO_SUCH_FILE_ENTRY, (*pitr).c_str()).str()); } FileEntryHandle fileEntry = diskAdaptor->getFileEntryFromPath(*pitr); bitfieldMan->addFilter(fileEntry->getOffset(), fileEntry->getLength()); } bitfieldMan->enableFilter(); } void DefaultPieceStorage::setFileFilter(IntSequence seq) { std::deque fileIndexes = seq.flush(); // TODO Is sorting necessary? std::sort(fileIndexes.begin(), fileIndexes.end()); fileIndexes.erase(std::unique(fileIndexes.begin(), fileIndexes.end()), fileIndexes.end()); std::deque filePaths; const FileEntries& entries = diskAdaptor->getFileEntries(); int32_t entriesSize = entries.size(); for(int32_t i = 0; i < entriesSize; i++) { if(std::find(fileIndexes.begin(), fileIndexes.end(), i+1) != fileIndexes.end()) { logger->debug("index=%d is %s", i+1, entries[i]->getPath().c_str()); filePaths.push_back(entries[i]->getPath()); } } setFileFilter(filePaths); } // not unittested void DefaultPieceStorage::clearFileFilter() { bitfieldMan->clearFilter(); diskAdaptor->addAllDownloadEntry(); } // not unittested bool DefaultPieceStorage::downloadFinished() { // TODO iterate all requested FileEntry and Call bitfieldMan->isBitSetOffsetRange() return bitfieldMan->isFilteredAllBitSet(); } // not unittested bool DefaultPieceStorage::allDownloadFinished() { return bitfieldMan->isAllBitSet(); } // not unittested void DefaultPieceStorage::initStorage() { if(downloadContext->getFileMode() == DownloadContext::SINGLE) { logger->debug("Instantiating DirectDiskAdaptor"); DiskWriterHandle writer = _diskWriterFactory->newDiskWriter(); writer->setDirectIOAllowed(option->getAsBool(PREF_ENABLE_DIRECT_IO)); DirectDiskAdaptorHandle directDiskAdaptor(new DirectDiskAdaptor()); directDiskAdaptor->setDiskWriter(writer); directDiskAdaptor->setTotalLength(downloadContext->getTotalLength()); this->diskAdaptor = directDiskAdaptor; } else { // file mode == DownloadContext::MULTI logger->debug("Instantiating MultiDiskAdaptor"); MultiDiskAdaptorHandle multiDiskAdaptor(new MultiDiskAdaptor()); multiDiskAdaptor->setDirectIOAllowed(option->getAsBool(PREF_ENABLE_DIRECT_IO)); multiDiskAdaptor->setPieceLength(downloadContext->getPieceLength()); multiDiskAdaptor->setMaxOpenFiles(option->getAsInt(PREF_BT_MAX_OPEN_FILES)); this->diskAdaptor = multiDiskAdaptor; } diskAdaptor->setStoreDir(downloadContext->getDir()); diskAdaptor->setFileEntries(downloadContext->getFileEntries()); #ifdef HAVE_POSIX_FALLOCATE if(option->get(PREF_FILE_ALLOCATION) == V_FALLOC) { diskAdaptor->enableFallocate(); } #endif // HAVE_POSIX_FALLOCATE } void DefaultPieceStorage::setBitfield(const unsigned char* bitfield, size_t bitfieldLength) { bitfieldMan->setBitfield(bitfield, bitfieldLength); addPieceStats(bitfield, bitfieldLength); } size_t DefaultPieceStorage::getBitfieldLength() { return bitfieldMan->getBitfieldLength(); } const unsigned char* DefaultPieceStorage::getBitfield() { return bitfieldMan->getBitfield(); } DiskAdaptorHandle DefaultPieceStorage::getDiskAdaptor() { return diskAdaptor; } size_t DefaultPieceStorage::getPieceLength(size_t index) { return bitfieldMan->getBlockLength(index); } void DefaultPieceStorage::advertisePiece(int32_t cuid, size_t index) { HaveEntry entry(cuid, index); haves.push_front(entry); } void DefaultPieceStorage::getAdvertisedPieceIndexes(std::deque& indexes, int32_t myCuid, const Time& lastCheckTime) { for(Haves::const_iterator itr = haves.begin(); itr != haves.end(); itr++) { const Haves::value_type& have = *itr; if(have.getCuid() == myCuid) { continue; } if(lastCheckTime.isNewer(have.getRegisteredTime())) { break; } indexes.push_back(have.getIndex()); } } class FindElapsedHave { private: time_t elapsed; public: FindElapsedHave(time_t elapsed):elapsed(elapsed) {} bool operator()(const HaveEntry& have) { if(have.getRegisteredTime().elapsed(elapsed)) { return true; } else { return false; } } }; void DefaultPieceStorage::removeAdvertisedPiece(time_t elapsed) { Haves::iterator itr = std::find_if(haves.begin(), haves.end(), FindElapsedHave(elapsed)); if(itr != haves.end()) { logger->debug(MSG_REMOVED_HAVE_ENTRY, haves.end()-itr); haves.erase(itr, haves.end()); } } void DefaultPieceStorage::markAllPiecesDone() { bitfieldMan->setAllBit(); } void DefaultPieceStorage::markPiecesDone(uint64_t length) { if(length == bitfieldMan->getTotalLength()) { bitfieldMan->setAllBit(); } else { size_t numPiece = length/bitfieldMan->getBlockLength(); if(numPiece > 0) { bitfieldMan->setBitRange(0, numPiece-1); } size_t r = (length%bitfieldMan->getBlockLength())/Piece::BLOCK_LENGTH; if(r > 0) { PieceHandle p(new Piece(numPiece, bitfieldMan->getBlockLength(numPiece))); for(size_t i = 0; i < r; ++i) { p->completeBlock(i); } #ifdef ENABLE_MESSAGE_DIGEST p->setHashAlgo(downloadContext->getPieceHashAlgo()); #endif // ENABLE_MESSAGE_DIGEST addUsedPiece(p); } } } void DefaultPieceStorage::markPieceMissing(size_t index) { bitfieldMan->unsetBit(index); } void DefaultPieceStorage::addInFlightPiece(const Pieces& pieces) { usedPieces.insert(usedPieces.end(), pieces.begin(), pieces.end()); std::sort(usedPieces.begin(), usedPieces.end()); } size_t DefaultPieceStorage::countInFlightPiece() { return usedPieces.size(); } void DefaultPieceStorage::getInFlightPieces(std::deque >& pieces) { pieces.insert(pieces.end(), usedPieces.begin(), usedPieces.end()); } void DefaultPieceStorage::setDiskWriterFactory(const DiskWriterFactoryHandle& diskWriterFactory) { _diskWriterFactory = diskWriterFactory; } void DefaultPieceStorage::addPieceStats(const unsigned char* bitfield, size_t bitfieldLength) { _pieceSelector->addPieceStats(bitfield, bitfieldLength); } void DefaultPieceStorage::subtractPieceStats(const unsigned char* bitfield, size_t bitfieldLength) { _pieceSelector->subtractPieceStats(bitfield, bitfieldLength); } void DefaultPieceStorage::updatePieceStats(const unsigned char* newBitfield, size_t newBitfieldLength, const unsigned char* oldBitfield) { _pieceSelector->updatePieceStats(newBitfield, newBitfieldLength, oldBitfield); } void DefaultPieceStorage::addPieceStats(size_t index) { _pieceSelector->addPieceStats(index); } } // namespace aria2