/* */ #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 "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" namespace aria2 { // TODO1.5 Remove this AbstractCommand::AbstractCommand(int32_t cuid, const SharedHandle& req, RequestGroup* requestGroup, DownloadEngine* e, const SocketHandle& s): Command(cuid), _requestGroup(requestGroup), req(req), 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(int32_t cuid, const SharedHandle& req, const SharedHandle& fileEntry, RequestGroup* requestGroup, DownloadEngine* e, const SocketHandle& s): Command(cuid), _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() { logger->debug("CUID#%d - socket: read:%d, write:%d, hup:%d, err:%d", cuid, _readEvent, _writeEvent, _hupEvent, _errorEvent); try { if(_requestGroup->downloadFinished() || _requestGroup->isHaltRequested()) { //logger->debug("CUID#%d - finished.", cuid); return true; } PeerStatHandle peerStat; if(!_requestGroup->getSegmentMan().isNull()) { peerStat = _requestGroup->getSegmentMan()->getPeerStat(cuid); } if(!peerStat.isNull()) { if(peerStat->getStatus() == PeerStat::REQUEST_IDLE) { logger->info(MSG_ABORT_REQUESTED, cuid); onAbort(); req->resetUrl(); tryReserved(); return true; } } if((checkSocketIsReadable && _readEvent) || (checkSocketIsWritable && _writeEvent) || _hupEvent || #ifdef ENABLE_ASYNC_DNS (nameResolverCheck && nameResolveFinished()) || #endif // ENABLE_ASYNC_DNS (!checkSocketIsReadable && !checkSocketIsWritable && !nameResolverCheck)) { checkPoint.reset(); if(!_requestGroup->getPieceStorage().isNull()) { _segments.clear(); _requestGroup->getSegmentMan()->getInFlightSegment(_segments, cuid); size_t maxSegments = req.isNull()?1:req->getMaxPipelinedRequest(); while(_segments.size() < maxSegments) { SegmentHandle segment = _requestGroup->getSegmentMan()->getSegment(cuid); if(segment.isNull()) { break; } _segments.push_back(segment); } if(_segments.empty()) { // TODO socket could be pooled here if pipelining is enabled... logger->info(MSG_NO_SEGMENT_AVAILABLE, cuid); return true; } } return executeInternal(); } else if(_errorEvent) { throw DL_RETRY_EX (StringFormat(MSG_NETWORK_PROBLEM, socket->getSocketError().c_str()).str()); } else { if(checkPoint.elapsed(timeout)) { // timeout triggers ServerStat error state. SharedHandle ss = e->_requestGroupMan->getOrCreateServerStat(req->getHost(), req->getProtocol()); ss->setError(); throw DL_RETRY_EX2(EX_TIME_OUT, DownloadResult::TIME_OUT); } e->commands.push_back(this); return false; } } catch(DlAbortEx& err) { if(req.isNull()) { logger->debug(EX_EXCEPTION_CAUGHT, err); } else { logger->error(MSG_DOWNLOAD_ABORTED, DL_ABORT_EX2(StringFormat ("URI=%s", req->getCurrentUrl().c_str()).str(),err), cuid, req->getUrl().c_str()); _requestGroup->addURIResult(req->getUrl(), err.getCode()); } onAbort(); // TODO Do we need this? //req->resetUrl(); tryReserved(); return true; } catch(DlRetryEx& err) { logger->info(MSG_RESTARTING_DOWNLOAD, DL_RETRY_EX2(StringFormat ("URI=%s", req->getCurrentUrl().c_str()).str(),err), cuid, req->getUrl().c_str()); req->addTryCount(); req->resetRedirectCount(); const unsigned int maxTries = getOption()->getAsInt(PREF_MAX_TRIES); bool isAbort = maxTries != 0 && req->getTryCount() >= maxTries; if(isAbort) { onAbort(); req->resetUrl(); } else { if(getOption()->getAsBool(PREF_RESET_URI)) { req->resetUrl(); } } if(isAbort) { logger->info(MSG_MAX_TRY, cuid, req->getTryCount()); logger->error(MSG_DOWNLOAD_ABORTED, err, cuid, req->getUrl().c_str()); _requestGroup->addURIResult(req->getUrl(), err.getCode()); tryReserved(); return true; } else { return prepareForRetry(getOption()->getAsInt(PREF_RETRY_WAIT)); } } catch(DownloadFailureException& err) { logger->error(EX_EXCEPTION_CAUGHT, err); _requestGroup->addURIResult(req->getUrl(), err.getCode()); _requestGroup->setHaltRequested(true); return true; } } void AbstractCommand::tryReserved() { _requestGroup->removeServerHost(cuid); 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().size() == 0) { return; } } Commands 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(!_segments.empty()) { // TODO1.5 subtract 1 from getPositionToWrite() SharedHandle fileEntry = _requestGroup->getDownloadContext()->findFileEntryByOffset(_segments.front()->getPositionToWrite()-1); logger->debug("CUID#%d - Pooling request URI=%s", cuid, req->getUrl().c_str()); _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()) { logger->debug(req->getCurrentUrl().c_str()); // TODO This might be a problem if the failure is caused by proxy. e->_requestGroupMan->getOrCreateServerStat(req->getHost(), req->getProtocol())->setError(); _requestGroup->removeIdenticalURI(req->getUrl()); _fileEntry->removeRequest(req); } logger->debug(MSG_UNREGISTER_CUID, cuid); //_segmentMan->unregisterId(cuid); if(!_requestGroup->getPieceStorage().isNull()) { _requestGroup->getSegmentMan()->cancelSegment(cuid); } // TODO1.5 Should be moved to FileEntry // _requestGroup->removeIdenticalURI(req->getUrl()); } 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(); } } static const std::string& getProxyStringFor(const std::string& proxyPref, const SharedHandle