/* */ #include "AbstractCommand.h" #include #include "Request.h" #include "DownloadEngine.h" #include "Option.h" #include "PeerStat.h" #include "SegmentMan.h" #include "Logger.h" #include "Segment.h" #include "DlAbortEx.h" #include "DlRetryEx.h" #include "DownloadFailureException.h" #include "CreateRequestCommand.h" #include "InitiateConnectionCommandFactory.h" #include "SleepCommand.h" #ifdef ENABLE_ASYNC_DNS #include "AsyncNameResolver.h" #endif // ENABLE_ASYNC_DNS #include "StreamCheckIntegrityEntry.h" #include "PieceStorage.h" #include "Socket.h" #include "message.h" #include "prefs.h" #include "StringFormat.h" #include "ServerStat.h" #include "RequestGroupMan.h" #include "A2STR.h" #include "util.h" #include "LogFactory.h" #include "DownloadContext.h" #include "wallclock.h" #include "NameResolver.h" namespace aria2 { AbstractCommand::AbstractCommand(cuid_t cuid, const SharedHandle& req, const SharedHandle& fileEntry, RequestGroup* requestGroup, DownloadEngine* e, const SocketHandle& s): Command(cuid), checkPoint(global::wallclock), _requestGroup(requestGroup), req(req), _fileEntry(fileEntry), e(e), socket(s), checkSocketIsReadable(false), checkSocketIsWritable(false), nameResolverCheck(false) { if(!socket.isNull() && socket->isOpen()) { setReadCheckSocket(socket); } timeout = _requestGroup->getTimeout(); _requestGroup->increaseStreamConnection(); _requestGroup->increaseNumCommand(); } AbstractCommand::~AbstractCommand() { disableReadCheckSocket(); disableWriteCheckSocket(); #ifdef ENABLE_ASYNC_DNS disableNameResolverCheck(_asyncNameResolver); #endif // ENABLE_ASYNC_DNS _requestGroup->decreaseNumCommand(); _requestGroup->decreaseStreamConnection(); } bool AbstractCommand::execute() { if(logger->debug()) { logger->debug("CUID#%s - socket: read:%d, write:%d, hup:%d, err:%d", util::itos(cuid).c_str(), _readEvent, _writeEvent, _hupEvent, _errorEvent); } try { if(_requestGroup->downloadFinished() || _requestGroup->isHaltRequested()) { return true; } if(!req.isNull() && req->removalRequested()) { if(logger->debug()) { logger->debug ("CUID#%s - Discard original URI=%s because it is requested.", util::itos(cuid).c_str(), req->getUri().c_str()); } return prepareForRetry(0); } // TODO it is not needed to check other PeerStats every time. // Find faster Request when no segment is available. if(!req.isNull() && _fileEntry->countPooledRequest() > 0 && !_requestGroup->getPieceStorage()->hasMissingUnusedPiece()) { SharedHandle fasterRequest = _fileEntry->findFasterRequest(req); if(!fasterRequest.isNull()) { if(logger->info()) { logger->info("CUID#%s - Use faster Request hostname=%s, port=%u", util::itos(cuid).c_str(), fasterRequest->getHost().c_str(), fasterRequest->getPort()); } // Cancel current Request object and use faster one. _fileEntry->removeRequest(req); Command* command = InitiateConnectionCommandFactory::createInitiateConnectionCommand (cuid, fasterRequest, _fileEntry, _requestGroup, e); e->setNoWait(true); e->commands.push_back(command); return true; } } if((checkSocketIsReadable && _readEvent) || (checkSocketIsWritable && _writeEvent) || _hupEvent || #ifdef ENABLE_ASYNC_DNS (nameResolverCheck && nameResolveFinished()) || #endif // ENABLE_ASYNC_DNS (!checkSocketIsReadable && !checkSocketIsWritable && !nameResolverCheck)) { checkPoint = global::wallclock; if(!_requestGroup->getPieceStorage().isNull()) { _segments.clear(); _requestGroup->getSegmentMan()->getInFlightSegment(_segments, cuid); if(!req.isNull() && _segments.empty()) { // This command previously has assigned segments, but it is // canceled. So discard current request chain. if(logger->debug()) { logger->debug("CUID#%s - It seems previously assigned segments are" " canceled. Restart.", util::itos(cuid).c_str()); } return prepareForRetry(0); } if(req.isNull() || req->getMaxPipelinedRequest() == 1 || _requestGroup->getDownloadContext()->getFileEntries().size() == 1) { if(_segments.empty()) { SharedHandle segment = _requestGroup->getSegmentMan()->getSegment(cuid); if(!segment.isNull()) { _segments.push_back(segment); } } if(_segments.empty()) { // TODO socket could be pooled here if pipelining is enabled... if(logger->info()) { logger->info(MSG_NO_SEGMENT_AVAILABLE, util::itos(cuid).c_str()); } // When all segments are ignored in SegmentMan, there are // no URIs available, so don't retry. if(_requestGroup->getSegmentMan()->allSegmentsIgnored()) { if(logger->debug()) { logger->debug("All segments are ignored."); } return true; } else { return prepareForRetry(1); } } } else { size_t maxSegments = req->getMaxPipelinedRequest(); if(_segments.size() < maxSegments) { _requestGroup->getSegmentMan()->getSegment (_segments, cuid, _fileEntry, maxSegments); } if(_segments.empty()) { return prepareForRetry(0); } } } return executeInternal(); } else if(_errorEvent) { throw DL_RETRY_EX (StringFormat(MSG_NETWORK_PROBLEM, socket->getSocketError().c_str()).str()); } else { if(checkPoint.difference(global::wallclock) >= timeout) { // timeout triggers ServerStat error state. SharedHandle ss = e->_requestGroupMan->getOrCreateServerStat(req->getHost(), req->getProtocol()); ss->setError(); throw DL_RETRY_EX2(EX_TIME_OUT, downloadresultcode::TIME_OUT); } e->commands.push_back(this); return false; } } catch(DlAbortEx& err) { if(req.isNull()) { if(logger->debug()) { logger->debug(EX_EXCEPTION_CAUGHT, err); } } else { logger->error(MSG_DOWNLOAD_ABORTED, DL_ABORT_EX2(StringFormat ("URI=%s", req->getCurrentUri().c_str()).str(),err), util::itos(cuid).c_str(), req->getUri().c_str()); _fileEntry->addURIResult(req->getUri(), err.getCode()); _requestGroup->setLastUriResult(req->getUri(), err.getCode()); if(err.getCode() == downloadresultcode::CANNOT_RESUME) { _requestGroup->increaseResumeFailureCount(); } } onAbort(); tryReserved(); return true; } catch(DlRetryEx& err) { assert(!req.isNull()); if(logger->info()) { logger->info(MSG_RESTARTING_DOWNLOAD, DL_RETRY_EX2(StringFormat ("URI=%s", req->getCurrentUri().c_str()).str(), err), util::itos(cuid).c_str(), req->getUri().c_str()); } req->addTryCount(); req->resetRedirectCount(); const unsigned int maxTries = getOption()->getAsInt(PREF_MAX_TRIES); bool isAbort = maxTries != 0 && req->getTryCount() >= maxTries; if(isAbort) { if(logger->info()) { logger->info(MSG_MAX_TRY, util::itos(cuid).c_str(), req->getTryCount()); } logger->error(MSG_DOWNLOAD_ABORTED, err, util::itos(cuid).c_str(), req->getUri().c_str()); _fileEntry->addURIResult(req->getUri(), err.getCode()); _requestGroup->setLastUriResult(req->getUri(), err.getCode()); if(err.getCode() == downloadresultcode::CANNOT_RESUME) { _requestGroup->increaseResumeFailureCount(); } onAbort(); tryReserved(); return true; } else { return prepareForRetry(0); } } catch(DownloadFailureException& err) { logger->error(EX_EXCEPTION_CAUGHT, err); if(!req.isNull()) { _fileEntry->addURIResult(req->getUri(), err.getCode()); _requestGroup->setLastUriResult(req->getUri(), err.getCode()); } _requestGroup->setHaltRequested(true); return true; } } void AbstractCommand::tryReserved() { if(_requestGroup->getDownloadContext()->getFileEntries().size() == 1) { const SharedHandle& entry = _requestGroup->getDownloadContext()->getFirstFileEntry(); // Don't create new command if currently file length is unknown // and there are no URI left. Because file length is unknown, we // can assume that there are no in-flight request object. if(entry->getLength() == 0 && entry->getRemainingUris().empty()) { if(logger->debug()) { logger->debug("CUID#%s - Not trying next request." " No reserved/pooled request is remaining and" " total length is still unknown.", util::itos(cuid).c_str()); } return; } } if(logger->debug()) { logger->debug("CUID#%s - Trying reserved/pooled request.", util::itos(cuid).c_str()); } std::vector commands; _requestGroup->createNextCommand(commands, e, 1); e->setNoWait(true); e->addCommand(commands); } bool AbstractCommand::prepareForRetry(time_t wait) { if(!_requestGroup->getPieceStorage().isNull()) { _requestGroup->getSegmentMan()->cancelSegment(cuid); } if(!req.isNull()) { _fileEntry->poolRequest(req); if(logger->debug()) { logger->debug("CUID#%s - Pooling request URI=%s", util::itos(cuid).c_str(), req->getUri().c_str()); } if(!_requestGroup->getSegmentMan().isNull()) { _requestGroup->getSegmentMan()->recognizeSegmentFor(_fileEntry); } } Command* command = new CreateRequestCommand(cuid, _requestGroup, e); if(wait == 0) { e->setNoWait(true); e->commands.push_back(command); } else { SleepCommand* scom = new SleepCommand(cuid, e, _requestGroup, command, wait); e->commands.push_back(scom); } return true; } void AbstractCommand::onAbort() { if(!req.isNull()) { // TODO This might be a problem if the failure is caused by proxy. e->_requestGroupMan->getOrCreateServerStat(req->getHost(), req->getProtocol())->setError(); _fileEntry->removeIdenticalURI(req->getUri()); _fileEntry->removeRequest(req); } if(logger->debug()) { logger->debug("CUID#%s - Aborting download", util::itos(cuid).c_str()); } if(!_requestGroup->getPieceStorage().isNull()) { SharedHandle segmentMan = _requestGroup->getSegmentMan(); segmentMan->cancelSegment(cuid); // Don't do following process if BitTorrent is involved or files // in DownloadContext is more than 1. The latter condition is // limitation of current implementation. if(!getOption()->getAsBool(PREF_ALWAYS_RESUME) && !_fileEntry.isNull() && segmentMan->calculateSessionDownloadLength() == 0 && !_requestGroup->p2pInvolved() && _requestGroup->getDownloadContext()->getFileEntries().size() == 1) { const int maxTries = getOption()->getAsInt(PREF_MAX_RESUME_FAILURE_TRIES); if((maxTries > 0 && _requestGroup->getResumeFailureCount() >= maxTries)|| _fileEntry->emptyRequestUri()) { // Local file exists, but given servers(or at least contacted // ones) doesn't support resume. Let's restart download from // scratch. logger->notice("CUID#%s - Failed to resume download." " Download from scratch.", util::itos(cuid).c_str()); if(logger->debug()) { logger->debug("CUID#%s - Gathering URIs that has CANNOT_RESUME error", util::itos(cuid).c_str()); } // Set PREF_ALWAYS_RESUME to V_TRUE to avoid repeating this // process. getOption()->put(PREF_ALWAYS_RESUME, V_TRUE); std::deque res; _fileEntry->extractURIResult(res, downloadresultcode::CANNOT_RESUME); if(!res.empty()) { segmentMan->cancelAllSegments(); segmentMan->eraseSegmentWrittenLengthMemo(); _requestGroup->getPieceStorage()->markPiecesDone(0); std::vector uris; uris.reserve(res.size()); std::transform(res.begin(), res.end(), std::back_inserter(uris), std::mem_fun_ref(&URIResult::getURI)); if(logger->debug()) { logger->debug("CUID#%s - %lu URIs found.", util::itos(cuid).c_str(), static_cast(uris.size())); } _fileEntry->addUris(uris.begin(), uris.end()); segmentMan->recognizeSegmentFor(_fileEntry); } } } } } void AbstractCommand::disableReadCheckSocket() { if(checkSocketIsReadable) { e->deleteSocketForReadCheck(readCheckTarget, this); checkSocketIsReadable = false; readCheckTarget = SocketHandle(); } } void AbstractCommand::setReadCheckSocket(const SocketHandle& socket) { if(!socket->isOpen()) { disableReadCheckSocket(); } else { if(checkSocketIsReadable) { if(readCheckTarget != socket) { e->deleteSocketForReadCheck(readCheckTarget, this); e->addSocketForReadCheck(socket, this); readCheckTarget = socket; } } else { e->addSocketForReadCheck(socket, this); checkSocketIsReadable = true; readCheckTarget = socket; } } } void AbstractCommand::setReadCheckSocketIf (const SharedHandle& socket, bool pred) { if(pred) { setReadCheckSocket(socket); } else { disableReadCheckSocket(); } } void AbstractCommand::disableWriteCheckSocket() { if(checkSocketIsWritable) { e->deleteSocketForWriteCheck(writeCheckTarget, this); checkSocketIsWritable = false; writeCheckTarget = SocketHandle(); } } void AbstractCommand::setWriteCheckSocket(const SocketHandle& socket) { if(!socket->isOpen()) { disableWriteCheckSocket(); } else { if(checkSocketIsWritable) { if(writeCheckTarget != socket) { e->deleteSocketForWriteCheck(writeCheckTarget, this); e->addSocketForWriteCheck(socket, this); writeCheckTarget = socket; } } else { e->addSocketForWriteCheck(socket, this); checkSocketIsWritable = true; writeCheckTarget = socket; } } } void AbstractCommand::setWriteCheckSocketIf (const SharedHandle& socket, bool pred) { if(pred) { setWriteCheckSocket(socket); } else { disableWriteCheckSocket(); } } // Returns proxy option value for the given protocol. static const std::string& getProxyOptionFor (const std::string& proxyPref, const SharedHandle