/* */ #include "BtPieceMessage.h" #include #include #include #include "bittorrent_helper.h" #include "util.h" #include "message.h" #include "DlAbortEx.h" #include "message_digest_helper.h" #include "DiskAdaptor.h" #include "Logger.h" #include "LogFactory.h" #include "Peer.h" #include "Piece.h" #include "PieceStorage.h" #include "BtMessageDispatcher.h" #include "BtMessageFactory.h" #include "BtRequestFactory.h" #include "PeerConnection.h" #include "fmt.h" #include "DownloadContext.h" namespace aria2 { const std::string BtPieceMessage::NAME("piece"); BtPieceMessage::BtPieceMessage (size_t index, uint32_t begin, size_t blockLength) : AbstractBtMessage(ID, NAME), index_(index), begin_(begin), blockLength_(blockLength), block_(0), rawData_(0) { setUploading(true); } BtPieceMessage::~BtPieceMessage() { delete [] rawData_; } void BtPieceMessage::setRawMessage(unsigned char* data) { delete [] rawData_; rawData_ = data; block_ = data+9; } BtPieceMessageHandle BtPieceMessage::create (const unsigned char* data, size_t dataLength) { bittorrent::assertPayloadLengthGreater(9, dataLength, NAME); bittorrent::assertID(ID, data, NAME); BtPieceMessageHandle message(new BtPieceMessage()); message->setIndex(bittorrent::getIntParam(data, 1)); message->setBegin(bittorrent::getIntParam(data, 5)); message->setBlockLength(dataLength-9); return message; } void BtPieceMessage::doReceivedAction() { if(isMetadataGetMode()) { return; } RequestSlot slot = getBtMessageDispatcher()->getOutstandingRequest (index_, begin_, blockLength_); getPeer()->updateDownloadLength(blockLength_); if(!RequestSlot::isNull(slot)) { getPeer()->snubbing(false); SharedHandle piece = getPieceStorage()->getPiece(index_); off_t offset = (off_t)index_*downloadContext_->getPieceLength()+begin_; A2_LOG_DEBUG(fmt(MSG_PIECE_RECEIVED, getCuid(), static_cast(index_), begin_, blockLength_, static_cast(offset), static_cast(slot.getBlockIndex()))); if(piece->hasBlock(slot.getBlockIndex())) { A2_LOG_DEBUG("Already have this block."); return; } getPieceStorage()->getDiskAdaptor()->writeData (block_, blockLength_, offset); piece->completeBlock(slot.getBlockIndex()); A2_LOG_DEBUG(fmt(MSG_PIECE_BITFIELD, getCuid(), util::toHex(piece->getBitfield(), piece->getBitfieldLength()).c_str())); piece->updateHash(begin_, block_, blockLength_); getBtMessageDispatcher()->removeOutstandingRequest(slot); if(piece->pieceComplete()) { if(checkPieceHash(piece)) { onNewPiece(piece); } else { onWrongPiece(piece); throw DL_ABORT_EX("Bad piece hash."); } } } else { A2_LOG_DEBUG(fmt("CUID#%lld - RequestSlot not found, index=%lu, begin=%u", getCuid(), static_cast(index_), begin_)); } } size_t BtPieceMessage::MESSAGE_HEADER_LENGTH = 13; unsigned char* BtPieceMessage::createMessageHeader() { /** * len --- 9+blockLength, 4bytes * id --- 7, 1byte * index --- index, 4bytes * begin --- begin, 4bytes * total: 13bytes */ unsigned char* msgHeader = new unsigned char[MESSAGE_HEADER_LENGTH]; bittorrent::createPeerMessageString(msgHeader, MESSAGE_HEADER_LENGTH, 9+blockLength_, ID); bittorrent::setIntParam(&msgHeader[5], index_); bittorrent::setIntParam(&msgHeader[9], begin_); return msgHeader; } size_t BtPieceMessage::getMessageHeaderLength() { return MESSAGE_HEADER_LENGTH; } void BtPieceMessage::send() { if(isInvalidate()) { return; } size_t writtenLength; if(!isSendingInProgress()) { A2_LOG_INFO(fmt(MSG_SEND_PEER_MESSAGE, getCuid(), getPeer()->getIPAddress().c_str(), getPeer()->getPort(), toString().c_str())); unsigned char* msgHdr = createMessageHeader(); size_t msgHdrLen = getMessageHeaderLength(); A2_LOG_DEBUG(fmt("msglength = %lu bytes", static_cast(msgHdrLen+blockLength_))); getPeerConnection()->pushBytes(msgHdr, msgHdrLen); off_t pieceDataOffset = (off_t)index_*downloadContext_->getPieceLength()+begin_; pushPieceData(pieceDataOffset, blockLength_); } writtenLength = getPeerConnection()->sendPendingData(); getPeer()->updateUploadLength(writtenLength); setSendingInProgress(!getPeerConnection()->sendBufferIsEmpty()); } void BtPieceMessage::pushPieceData(off_t offset, size_t length) const { assert(length <= 16*1024); unsigned char* buf = new unsigned char[length]; ssize_t r; try { r = getPieceStorage()->getDiskAdaptor()->readData(buf, length, offset); } catch(RecoverableException& e) { delete [] buf; throw; } if(r == static_cast(length)) { getPeerConnection()->pushBytes(buf, length); } else { throw DL_ABORT_EX(EX_DATA_READ); } } std::string BtPieceMessage::toString() const { return strconcat(NAME, " index=", util::itos(index_), ", begin=", util::itos(begin_), ", length=", util::itos(blockLength_)); } bool BtPieceMessage::checkPieceHash(const SharedHandle& piece) { if(!getPieceStorage()->isEndGame() && piece->isHashCalculated()) { A2_LOG_DEBUG(fmt("Hash is available!! index=%lu", static_cast(piece->getIndex()))); return piece->getDigest() == downloadContext_->getPieceHash(piece->getIndex()); } else { off_t offset = (off_t)piece->getIndex()*downloadContext_->getPieceLength(); return message_digest::staticSHA1Digest (getPieceStorage()->getDiskAdaptor(), offset, piece->getLength()) == downloadContext_->getPieceHash(piece->getIndex()); } } void BtPieceMessage::onNewPiece(const SharedHandle& piece) { A2_LOG_INFO(fmt(MSG_GOT_NEW_PIECE, getCuid(), static_cast(piece->getIndex()))); getPieceStorage()->completePiece(piece); getPieceStorage()->advertisePiece(getCuid(), piece->getIndex()); } void BtPieceMessage::onWrongPiece(const SharedHandle& piece) { A2_LOG_INFO(fmt(MSG_GOT_WRONG_PIECE, getCuid(), static_cast(piece->getIndex()))); erasePieceOnDisk(piece); piece->clearAllBlock(); piece->destroyHashContext(); getBtRequestFactory()->removeTargetPiece(piece); } void BtPieceMessage::erasePieceOnDisk(const SharedHandle& piece) { size_t BUFSIZE = 4096; unsigned char buf[BUFSIZE]; memset(buf, 0, BUFSIZE); off_t offset = (off_t)piece->getIndex()*downloadContext_->getPieceLength(); div_t res = div(piece->getLength(), BUFSIZE); for(int i = 0; i < res.quot; ++i) { getPieceStorage()->getDiskAdaptor()->writeData(buf, BUFSIZE, offset); offset += BUFSIZE; } if(res.rem > 0) { getPieceStorage()->getDiskAdaptor()->writeData(buf, res.rem, offset); } } void BtPieceMessage::onChokingEvent(const BtChokingEvent& event) { if(!isInvalidate() && !isSendingInProgress() && !getPeer()->isInAmAllowedIndexSet(index_)) { A2_LOG_DEBUG(fmt(MSG_REJECT_PIECE_CHOKED, getCuid(), static_cast(index_), begin_, blockLength_)); if(getPeer()->isFastExtensionEnabled()) { BtMessageHandle rej = getBtMessageFactory()->createRejectMessage (index_, begin_, blockLength_); getBtMessageDispatcher()->addMessageToQueue(rej); } setInvalidate(true); } } void BtPieceMessage::onCancelSendingPieceEvent (const BtCancelSendingPieceEvent& event) { if(!isInvalidate() && !isSendingInProgress() && index_ == event.getIndex() && begin_ == event.getBegin() && blockLength_ == event.getLength()) { A2_LOG_DEBUG(fmt(MSG_REJECT_PIECE_CANCEL, getCuid(), static_cast(index_), begin_, blockLength_)); if(getPeer()->isFastExtensionEnabled()) { BtMessageHandle rej = getBtMessageFactory()->createRejectMessage (index_, begin_, blockLength_); getBtMessageDispatcher()->addMessageToQueue(rej); } setInvalidate(true); } } void BtPieceMessage::setDownloadContext (const SharedHandle& downloadContext) { downloadContext_ = downloadContext; } } // namespace aria2