DownloadCommand.cc 13 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369
  1. /* <!-- copyright */
  2. /*
  3. * aria2 - The high speed download utility
  4. *
  5. * Copyright (C) 2006 Tatsuhiro Tsujikawa
  6. *
  7. * This program is free software; you can redistribute it and/or modify
  8. * it under the terms of the GNU General Public License as published by
  9. * the Free Software Foundation; either version 2 of the License, or
  10. * (at your option) any later version.
  11. *
  12. * This program is distributed in the hope that it will be useful,
  13. * but WITHOUT ANY WARRANTY; without even the implied warranty of
  14. * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
  15. * GNU General Public License for more details.
  16. *
  17. * You should have received a copy of the GNU General Public License
  18. * along with this program; if not, write to the Free Software
  19. * Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301 USA
  20. *
  21. * In addition, as a special exception, the copyright holders give
  22. * permission to link the code of portions of this program with the
  23. * OpenSSL library under certain conditions as described in each
  24. * individual source file, and distribute linked combinations
  25. * including the two.
  26. * You must obey the GNU General Public License in all respects
  27. * for all of the code used other than OpenSSL. If you modify
  28. * file(s) with this exception, you may extend this exception to your
  29. * version of the file(s), but you are not obligated to do so. If you
  30. * do not wish to do so, delete this exception statement from your
  31. * version. If you delete this exception statement from all source
  32. * files in the program, then also delete it here.
  33. */
  34. /* copyright --> */
  35. #include "DownloadCommand.h"
  36. #include <cassert>
  37. #include "Request.h"
  38. #include "RequestGroup.h"
  39. #include "DownloadEngine.h"
  40. #include "PeerStat.h"
  41. #include "DlAbortEx.h"
  42. #include "DlRetryEx.h"
  43. #include "SegmentMan.h"
  44. #include "Segment.h"
  45. #include "Logger.h"
  46. #include "ChecksumCheckIntegrityEntry.h"
  47. #include "PieceStorage.h"
  48. #include "CheckIntegrityCommand.h"
  49. #include "DiskAdaptor.h"
  50. #include "DownloadContext.h"
  51. #include "Option.h"
  52. #include "util.h"
  53. #include "Socket.h"
  54. #include "message.h"
  55. #include "prefs.h"
  56. #include "StringFormat.h"
  57. #include "Decoder.h"
  58. #include "RequestGroupMan.h"
  59. #include "wallclock.h"
  60. #include "ServerStatMan.h"
  61. #include "FileAllocationEntry.h"
  62. #ifdef ENABLE_MESSAGE_DIGEST
  63. # include "MessageDigestHelper.h"
  64. #endif // ENABLE_MESSAGE_DIGEST
  65. namespace aria2 {
  66. static const size_t BUFSIZE = 16*1024;
  67. DownloadCommand::DownloadCommand(cuid_t cuid,
  68. const SharedHandle<Request>& req,
  69. const SharedHandle<FileEntry>& fileEntry,
  70. RequestGroup* requestGroup,
  71. DownloadEngine* e,
  72. const SocketHandle& s):
  73. AbstractCommand(cuid, req, fileEntry, requestGroup, e, s),
  74. _buf(new unsigned char[BUFSIZE]),
  75. _startupIdleTime(10),
  76. _lowestDownloadSpeedLimit(0)
  77. #ifdef ENABLE_MESSAGE_DIGEST
  78. , _pieceHashValidationEnabled(false)
  79. #endif // ENABLE_MESSAGE_DIGEST
  80. {
  81. #ifdef ENABLE_MESSAGE_DIGEST
  82. {
  83. if(getOption()->getAsBool(PREF_REALTIME_CHUNK_CHECKSUM)) {
  84. const std::string& algo = getDownloadContext()->getPieceHashAlgo();
  85. if(MessageDigestContext::supports(algo)) {
  86. _messageDigestContext.reset(new MessageDigestContext());
  87. _messageDigestContext->trySetAlgo(algo);
  88. _messageDigestContext->digestInit();
  89. _pieceHashValidationEnabled = true;
  90. }
  91. }
  92. }
  93. #endif // ENABLE_MESSAGE_DIGEST
  94. _peerStat = req->initPeerStat();
  95. _peerStat->downloadStart();
  96. getSegmentMan()->registerPeerStat(_peerStat);
  97. }
  98. DownloadCommand::~DownloadCommand() {
  99. _peerStat->downloadStop();
  100. getSegmentMan()->updateFastestPeerStat(_peerStat);
  101. delete [] _buf;
  102. }
  103. bool DownloadCommand::executeInternal() {
  104. if(getDownloadEngine()->getRequestGroupMan()->doesOverallDownloadSpeedExceed()
  105. || getRequestGroup()->doesDownloadSpeedExceed()) {
  106. getDownloadEngine()->addCommand(this);
  107. disableReadCheckSocket();
  108. return false;
  109. }
  110. setReadCheckSocket(getSocket());
  111. SharedHandle<Segment> segment = getSegments().front();
  112. size_t bufSize;
  113. if(segment->getLength() > 0) {
  114. if(static_cast<uint64_t>(segment->getPosition()+segment->getLength()) <=
  115. static_cast<uint64_t>(getFileEntry()->getLastOffset())) {
  116. bufSize = std::min(segment->getLength()-segment->getWrittenLength(),
  117. BUFSIZE);
  118. } else {
  119. bufSize =
  120. std::min
  121. (static_cast<size_t>
  122. (getFileEntry()->getLastOffset()-segment->getPositionToWrite()),
  123. BUFSIZE);
  124. }
  125. } else {
  126. bufSize = BUFSIZE;
  127. }
  128. getSocket()->readData(_buf, bufSize);
  129. const SharedHandle<DiskAdaptor>& diskAdaptor =
  130. getPieceStorage()->getDiskAdaptor();
  131. const unsigned char* bufFinal;
  132. size_t bufSizeFinal;
  133. std::string decoded;
  134. if(_transferEncodingDecoder.isNull()) {
  135. bufFinal = _buf;
  136. bufSizeFinal = bufSize;
  137. } else {
  138. decoded = _transferEncodingDecoder->decode(_buf, bufSize);
  139. bufFinal = reinterpret_cast<const unsigned char*>(decoded.c_str());
  140. bufSizeFinal = decoded.size();
  141. }
  142. if(_contentEncodingDecoder.isNull()) {
  143. diskAdaptor->writeData(bufFinal, bufSizeFinal,
  144. segment->getPositionToWrite());
  145. } else {
  146. std::string out = _contentEncodingDecoder->decode(bufFinal, bufSizeFinal);
  147. diskAdaptor->writeData(reinterpret_cast<const unsigned char*>(out.data()),
  148. out.size(),
  149. segment->getPositionToWrite());
  150. bufSizeFinal = out.size();
  151. }
  152. #ifdef ENABLE_MESSAGE_DIGEST
  153. if(_pieceHashValidationEnabled) {
  154. segment->updateHash(segment->getWrittenLength(), bufFinal, bufSizeFinal);
  155. }
  156. #endif // ENABLE_MESSAGE_DIGEST
  157. if(bufSizeFinal > 0) {
  158. segment->updateWrittenLength(bufSizeFinal);
  159. }
  160. _peerStat->updateDownloadLength(bufSize);
  161. getSegmentMan()->updateDownloadSpeedFor(_peerStat);
  162. bool segmentPartComplete = false;
  163. // Note that GrowSegment::complete() always returns false.
  164. if(_transferEncodingDecoder.isNull() && _contentEncodingDecoder.isNull()) {
  165. if(segment->complete() ||
  166. segment->getPositionToWrite() == getFileEntry()->getLastOffset()) {
  167. segmentPartComplete = true;
  168. } else if(segment->getLength() == 0 && bufSize == 0 &&
  169. !getSocket()->wantRead() && !getSocket()->wantWrite()) {
  170. segmentPartComplete = true;
  171. }
  172. } else if(!_transferEncodingDecoder.isNull() &&
  173. (segment->complete() ||
  174. segment->getPositionToWrite() == getFileEntry()->getLastOffset())){
  175. // In this case, transferEncodingDecoder is used and
  176. // Content-Length is known.
  177. segmentPartComplete = true;
  178. } else if((_transferEncodingDecoder.isNull() ||
  179. _transferEncodingDecoder->finished()) &&
  180. (_contentEncodingDecoder.isNull() ||
  181. _contentEncodingDecoder->finished())) {
  182. segmentPartComplete = true;
  183. }
  184. if(!segmentPartComplete && bufSize == 0 &&
  185. !getSocket()->wantRead() && !getSocket()->wantWrite()) {
  186. throw DL_RETRY_EX(EX_GOT_EOF);
  187. }
  188. if(segmentPartComplete) {
  189. if(segment->complete() || segment->getLength() == 0) {
  190. // If segment->getLength() == 0, the server doesn't provide
  191. // content length, but the client detected that download
  192. // completed.
  193. if(getLogger()->info()) {
  194. getLogger()->info(MSG_SEGMENT_DOWNLOAD_COMPLETED,
  195. util::itos(getCuid()).c_str());
  196. }
  197. #ifdef ENABLE_MESSAGE_DIGEST
  198. {
  199. const std::string& expectedPieceHash =
  200. getDownloadContext()->getPieceHash(segment->getIndex());
  201. if(_pieceHashValidationEnabled && !expectedPieceHash.empty()) {
  202. if(segment->isHashCalculated()) {
  203. if(getLogger()->debug()) {
  204. getLogger()->debug
  205. ("Hash is available! index=%lu",
  206. static_cast<unsigned long>(segment->getIndex()));
  207. }
  208. validatePieceHash
  209. (segment, expectedPieceHash, segment->getHashString());
  210. } else {
  211. _messageDigestContext->digestReset();
  212. validatePieceHash
  213. (segment, expectedPieceHash,
  214. MessageDigestHelper::digest
  215. (_messageDigestContext.get(),
  216. getPieceStorage()->getDiskAdaptor(),
  217. segment->getPosition(),
  218. segment->getLength()));
  219. }
  220. } else {
  221. getSegmentMan()->completeSegment(getCuid(), segment);
  222. }
  223. }
  224. #else // !ENABLE_MESSAGE_DIGEST
  225. getSegmentMan()->completeSegment(getCuid(), segment);
  226. #endif // !ENABLE_MESSAGE_DIGEST
  227. } else {
  228. // If segment is not canceled here, in the next pipelining
  229. // request, aria2 requests bad range
  230. // [FileEntry->getLastOffset(), FileEntry->getLastOffset())
  231. getSegmentMan()->cancelSegment(getCuid(), segment);
  232. }
  233. checkLowestDownloadSpeed();
  234. // this unit is going to download another segment.
  235. return prepareForNextSegment();
  236. } else {
  237. checkLowestDownloadSpeed();
  238. setWriteCheckSocketIf(getSocket(), getSocket()->wantWrite());
  239. getDownloadEngine()->addCommand(this);
  240. return false;
  241. }
  242. }
  243. void DownloadCommand::checkLowestDownloadSpeed() const
  244. {
  245. // calculate downloading speed
  246. if(_peerStat->getDownloadStartTime().difference(global::wallclock) >=
  247. _startupIdleTime) {
  248. unsigned int nowSpeed = _peerStat->calculateDownloadSpeed();
  249. if(_lowestDownloadSpeedLimit > 0 && nowSpeed <= _lowestDownloadSpeedLimit) {
  250. throw DL_ABORT_EX2(StringFormat(EX_TOO_SLOW_DOWNLOAD_SPEED,
  251. nowSpeed,
  252. _lowestDownloadSpeedLimit,
  253. getRequest()->getHost().c_str()).str(),
  254. downloadresultcode::TOO_SLOW_DOWNLOAD_SPEED);
  255. }
  256. }
  257. }
  258. bool DownloadCommand::prepareForNextSegment() {
  259. if(getRequestGroup()->downloadFinished()) {
  260. // Remove in-flight request here.
  261. getFileEntry()->poolRequest(getRequest());
  262. // If this is a single file download, and file size becomes known
  263. // just after downloading, set total length to FileEntry object
  264. // here.
  265. if(getDownloadContext()->getFileEntries().size() == 1) {
  266. if(getFileEntry()->getLength() == 0) {
  267. getFileEntry()->setLength(getPieceStorage()->getCompletedLength());
  268. }
  269. }
  270. #ifdef ENABLE_MESSAGE_DIGEST
  271. if(getDownloadContext()->getPieceHashAlgo().empty()) {
  272. SharedHandle<CheckIntegrityEntry> entry
  273. (new ChecksumCheckIntegrityEntry(getRequestGroup()));
  274. if(entry->isValidationReady()) {
  275. entry->initValidator();
  276. // TODO do we need cuttrailinggarbage here?
  277. getDownloadEngine()->getCheckIntegrityMan()->pushEntry(entry);
  278. }
  279. }
  280. // Following 2lines are needed for DownloadEngine to detect
  281. // completed RequestGroups without 1sec delay.
  282. getDownloadEngine()->setNoWait(true);
  283. getDownloadEngine()->setRefreshInterval(0);
  284. #endif // ENABLE_MESSAGE_DIGEST
  285. return true;
  286. } else {
  287. // The number of segments should be 1 in order to pass through the next
  288. // segment.
  289. if(getSegments().size() == 1) {
  290. SharedHandle<Segment> tempSegment = getSegments().front();
  291. if(!tempSegment->complete()) {
  292. return prepareForRetry(0);
  293. }
  294. SharedHandle<Segment> nextSegment =
  295. getSegmentMan()->getCleanSegmentIfOwnerIsIdle
  296. (getCuid(), tempSegment->getIndex()+1);
  297. if(nextSegment.isNull()) {
  298. return prepareForRetry(0);
  299. } else {
  300. getDownloadEngine()->addCommand(this);
  301. return false;
  302. }
  303. } else {
  304. return prepareForRetry(0);
  305. }
  306. }
  307. }
  308. #ifdef ENABLE_MESSAGE_DIGEST
  309. void DownloadCommand::validatePieceHash(const SharedHandle<Segment>& segment,
  310. const std::string& expectedPieceHash,
  311. const std::string& actualPieceHash)
  312. {
  313. if(actualPieceHash == expectedPieceHash) {
  314. getLogger()->info(MSG_GOOD_CHUNK_CHECKSUM, actualPieceHash.c_str());
  315. getSegmentMan()->completeSegment(getCuid(), segment);
  316. } else {
  317. getLogger()->info(EX_INVALID_CHUNK_CHECKSUM,
  318. segment->getIndex(),
  319. util::itos(segment->getPosition(), true).c_str(),
  320. expectedPieceHash.c_str(),
  321. actualPieceHash.c_str());
  322. segment->clear();
  323. getSegmentMan()->cancelSegment(getCuid());
  324. throw DL_RETRY_EX
  325. (StringFormat("Invalid checksum index=%d", segment->getIndex()).str());
  326. }
  327. }
  328. #endif // ENABLE_MESSAGE_DIGEST
  329. void DownloadCommand::setTransferEncodingDecoder
  330. (const SharedHandle<Decoder>& decoder)
  331. {
  332. this->_transferEncodingDecoder = decoder;
  333. }
  334. void DownloadCommand::setContentEncodingDecoder
  335. (const SharedHandle<Decoder>& decoder)
  336. {
  337. _contentEncodingDecoder = decoder;
  338. }
  339. } // namespace aria2