github
/
aria2
mirror of https://github.com/aria2/aria2
1
0
Fork 0
Code Issues Releases Wiki Activity

Revert "Cleanup AbstractCommand" 

This reverts commit 0465aa5455.
pull/235/head
Nils Maier 2014-05-29 18:21:49 +02:00
parent 0465aa5455
commit 6240345fd5
5 changed files with 479 additions and 566 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

538
src/AbstractCommand.cc
View File

@ -74,12 +74,10 @@
namespace aria2 {
AbstractCommand::AbstractCommand(
cuid_t cuid,
AbstractCommand::AbstractCommand(cuid_t cuid,
const std::shared_ptr<Request>& req,
const std::shared_ptr<FileEntry>& fileEntry,
RequestGroup *requestGroup,
DownloadEngine *e,
RequestGroup* requestGroup, DownloadEngine* e,
const std::shared_ptr<SocketCore>& s,
const std::shared_ptr<SocketRecvBuffer>& socketRecvBuffer,
bool incNumConnection)
@ -113,8 +111,7 @@ AbstractCommand::AbstractCommand(
#endif // ENABLE_ASYNC_DNS
}
AbstractCommand::~AbstractCommand()
{
AbstractCommand::~AbstractCommand() {
disableReadCheckSocket();
disableWriteCheckSocket();
#ifdef ENABLE_ASYNC_DNS
@ -127,8 +124,8 @@ AbstractCommand::~AbstractCommand()
}
}
void
AbstractCommand::useFasterRequest(const std::shared_ptr<Request> &fasterRequest)
void AbstractCommand::useFasterRequest
(const std::shared_ptr<Request>& fasterRequest)
{
A2_LOG_INFO(fmt("CUID#%" PRId64 " - Use faster Request hostname=%s, port=%u",
getCuid(),
@ -137,45 +134,13 @@ AbstractCommand::useFasterRequest(const std::shared_ptr<Request> &fasterRequest)
// Cancel current Request object and use faster one.
fileEntry_->removeRequest(req_);
e_->setNoWait(true);
e_->addCommand(
InitiateConnectionCommandFactory::createInitiateConnectionCommand(
getCuid(), fasterRequest, fileEntry_, requestGroup_, e_));
e_->addCommand(InitiateConnectionCommandFactory::
createInitiateConnectionCommand
(getCuid(), fasterRequest, fileEntry_, requestGroup_, e_));
}
bool AbstractCommand::shouldCheck() const
{
if (checkSocketIsReadable_) {
if (readEventEnabled()) {
return true;
}
if (socketRecvBuffer_ && !socketRecvBuffer_->bufferEmpty()) {
return true;
}
}
if (checkSocketIsWritable_ && writeEventEnabled()) {
return true;
}
#ifdef ENABLE_ASYNC_DNS
const auto resolverChecked = asyncNameResolverMan_->resolverChecked();
if (resolverChecked && asyncNameResolverMan_->getStatus() != 0) {
return true;
}
if (!checkSocketIsReadable_ && !checkSocketIsWritable_ && !resolverChecked) {
return true;
}
#endif // ENABLE_ASYNC_DNS
return noCheck();
}
bool AbstractCommand::execute()
{
A2_LOG_DEBUG(fmt("CUID#%" PRId64
" - socket: read:%d, write:%d, hup:%d, err:%d",
bool AbstractCommand::execute() {
A2_LOG_DEBUG(fmt("CUID#%" PRId64 " - socket: read:%d, write:%d, hup:%d, err:%d",
getCuid(),
readEventEnabled(),
writeEventEnabled(),
@ -185,28 +150,20 @@ bool AbstractCommand::execute()
if(requestGroup_->downloadFinished() || requestGroup_->isHaltRequested()) {
return true;
}
if(req_ && req_->removalRequested()) {
A2_LOG_DEBUG(fmt("CUID#%" PRId64
" - Discard original URI=%s because it is"
A2_LOG_DEBUG(fmt("CUID#%" PRId64 " - Discard original URI=%s because it is"
" requested.",
getCuid(),
req_->getUri().c_str()));
getCuid(), req_->getUri().c_str()));
return prepareForRetry(0);
}
auto sm = getSegmentMan();
if(getPieceStorage()) {
segments_.clear();
sm->getInFlightSegment(segments_, getCuid());
getSegmentMan()->getInFlightSegment(segments_, getCuid());
if(req_ && segments_.empty()) {
// This command previously has assigned segments, but it is
// canceled. So discard current request chain. Plus, if no
// segment is available when http pipelining is used.
A2_LOG_DEBUG(fmt("CUID#%" PRId64
" - It seems previously assigned segments"
A2_LOG_DEBUG(fmt("CUID#%" PRId64 " - It seems previously assigned segments"
" are canceled. Restart.",
getCuid()));
// Request::isPipeliningEnabled() == true means aria2
@ -217,12 +174,12 @@ bool AbstractCommand::execute()
}
return prepareForRetry(0);
}
// TODO it is not needed to check other PeerStats every time.
// Find faster Request when no segment split is allowed.
if(req_ && fileEntry_->countPooledRequest() > 0 &&
requestGroup_->getPendingLength() < calculateMinSplitSize() * 2) {
auto fasterRequest = fileEntry_->findFasterRequest(req_);
requestGroup_->getTotalLength()-requestGroup_->getCompletedLength()
< calculateMinSplitSize()*2) {
std::shared_ptr<Request> fasterRequest = fileEntry_->findFasterRequest(req_);
if(fasterRequest) {
useFasterRequest(fasterRequest);
return true;
@ -239,22 +196,31 @@ bool AbstractCommand::execute()
if(getOption()->getAsBool(PREF_SELECT_LEAST_USED_HOST)) {
getDownloadEngine()->getRequestGroupMan()->getUsedHosts(usedHosts);
}
auto fasterRequest = fileEntry_->findFasterRequest(
req_, usedHosts, e_->getRequestGroupMan()->getServerStatMan());
std::shared_ptr<Request> fasterRequest =
fileEntry_->findFasterRequest
(req_, usedHosts, e_->getRequestGroupMan()->getServerStatMan());
if(fasterRequest) {
useFasterRequest(fasterRequest);
return true;
}
}
}
if (shouldCheck()) {
if((checkSocketIsReadable_ &&
(readEventEnabled() ||
(socketRecvBuffer_ && !socketRecvBuffer_->bufferEmpty()))) ||
(checkSocketIsWritable_ && writeEventEnabled()) ||
hupEventEnabled() ||
#ifdef ENABLE_ASYNC_DNS
(asyncNameResolverMan_->resolverChecked() &&
asyncNameResolverMan_->getStatus() != 0) ||
#endif // ENABLE_ASYNC_DNS
(!checkSocketIsReadable_ && !checkSocketIsWritable_
#ifdef ENABLE_ASYNC_DNS
&& !asyncNameResolverMan_->resolverChecked()
#endif // ENABLE_ASYNC_DNS
) || noCheck()) {
checkPoint_ = global::wallclock();
if (!getPieceStorage()) {
return executeInternal();
}
if(getPieceStorage()) {
if(!req_ || req_->getMaxPipelinedRequest() == 1 ||
// Why the following condition is necessary? That's because
// For single file download, SegmentMan::getSegment(cuid)
@ -263,55 +229,55 @@ bool AbstractCommand::execute()
size_t maxSegments = req_?req_->getMaxPipelinedRequest():1;
size_t minSplitSize = calculateMinSplitSize();
while(segments_.size() < maxSegments) {
auto segment = sm->getSegment(getCuid(), minSplitSize);
std::shared_ptr<Segment> segment =
getSegmentMan()->getSegment(getCuid(), minSplitSize);
if(!segment) {
break;
}
} else {
segments_.push_back(segment);
}
}
if(segments_.empty()) {
// TODO socket could be pooled here if pipelining is
// enabled... Hmm, I don't think if pipelining is enabled
// it does not go here.
A2_LOG_INFO(fmt(MSG_NO_SEGMENT_AVAILABLE, getCuid()));
A2_LOG_INFO(fmt(MSG_NO_SEGMENT_AVAILABLE,
getCuid()));
// When all segments are ignored in SegmentMan, there are
// no URIs available, so don't retry.
if (sm->allSegmentsIgnored()) {
if(getSegmentMan()->allSegmentsIgnored()) {
A2_LOG_DEBUG("All segments are ignored.");
// This will execute other idle Commands and let them
// finish quickly.
e_->setRefreshInterval(0);
return true;
}
} else {
return prepareForRetry(1);
}
}
else {
} else {
// For multi-file downloads
size_t minSplitSize = calculateMinSplitSize();
size_t maxSegments = req_->getMaxPipelinedRequest();
if(segments_.size() < maxSegments) {
sm->getSegment(
segments_, getCuid(), minSplitSize, fileEntry_, maxSegments);
getSegmentMan()->getSegment
(segments_, getCuid(), minSplitSize, fileEntry_, maxSegments);
}
if(segments_.empty()) {
return prepareForRetry(0);
}
}
}
return executeInternal();
}
if (errorEventEnabled()) {
throw DL_RETRY_EX(
fmt(MSG_NETWORK_PROBLEM, socket_->getSocketError().c_str()));
}
} else if(errorEventEnabled()) {
throw DL_RETRY_EX(fmt(MSG_NETWORK_PROBLEM,
socket_->getSocketError().c_str()));
} else {
if(checkPoint_.difference(global::wallclock()) >= timeout_) {
// timeout triggers ServerStat error state.
auto ss = e_->getRequestGroupMan()->getOrCreateServerStat(
req_->getHost(), req_->getProtocol());
std::shared_ptr<ServerStat> ss =
e_->getRequestGroupMan()->getOrCreateServerStat(req_->getHost(),
req_->getProtocol());
ss->setError();
// When DNS query was timeout, req_->getConnectedAddr() is
// empty.
@ -324,53 +290,56 @@ bool AbstractCommand::execute()
req_->getConnectedAddr(),
req_->getConnectedPort());
}
if (e_->findCachedIPAddress(req_->getConnectedHostname(),
req_->getConnectedPort()).empty()) {
if(e_->findCachedIPAddress
(req_->getConnectedHostname(), req_->getConnectedPort()).empty()) {
A2_LOG_DEBUG(fmt("CUID#%" PRId64 " - All IP addresses were marked bad."
" Removing Entry.",
getCuid()));
e_->removeCachedIPAddress(req_->getConnectedHostname(),
req_->getConnectedPort());
e_->removeCachedIPAddress
(req_->getConnectedHostname(), req_->getConnectedPort());
}
throw DL_RETRY_EX2(EX_TIME_OUT, error_code::TIME_OUT);
}
addCommandSelf();
return false;
}
catch (DlAbortEx &err) {
} catch(DlAbortEx& err) {
requestGroup_->setLastErrorCode(err.getErrorCode());
if(req_) {
A2_LOG_ERROR_EX(
fmt(MSG_DOWNLOAD_ABORTED, getCuid(), req_->getUri().c_str()),
DL_ABORT_EX2(fmt("URI=%s", req_->getCurrentUri().c_str()), err));
A2_LOG_ERROR_EX(fmt(MSG_DOWNLOAD_ABORTED,
getCuid(),
req_->getUri().c_str()),
DL_ABORT_EX2(fmt("URI=%s", req_->getCurrentUri().c_str()),
err));
fileEntry_->addURIResult(req_->getUri(), err.getErrorCode());
if(err.getErrorCode() == error_code::CANNOT_RESUME) {
requestGroup_->increaseResumeFailureCount();
}
}
else {
} else {
A2_LOG_DEBUG_EX(EX_EXCEPTION_CAUGHT, err);
}
onAbort();
tryReserved();
return true;
}
catch (DlRetryEx &err) {
} catch(DlRetryEx& err) {
assert(req_);
A2_LOG_INFO_EX(
fmt(MSG_RESTARTING_DOWNLOAD, getCuid(), req_->getUri().c_str()),
DL_RETRY_EX2(fmt("URI=%s", req_->getCurrentUri().c_str()), err));
A2_LOG_INFO_EX(fmt(MSG_RESTARTING_DOWNLOAD,
getCuid(), req_->getUri().c_str()),
DL_RETRY_EX2(fmt("URI=%s", req_->getCurrentUri().c_str()),
err));
req_->addTryCount();
req_->resetRedirectCount();
req_->resetUri();
const int maxTries = getOption()->getAsInt(PREF_MAX_TRIES);
bool isAbort = maxTries != 0 && req_->getTryCount() >= maxTries;
if(isAbort) {
A2_LOG_INFO(fmt(MSG_MAX_TRY, getCuid(), req_->getTryCount()));
A2_LOG_ERROR_EX(
fmt(MSG_DOWNLOAD_ABORTED, getCuid(), req_->getUri().c_str()), err);
A2_LOG_INFO(fmt(MSG_MAX_TRY,
getCuid(),
req_->getTryCount()));
A2_LOG_ERROR_EX(fmt(MSG_DOWNLOAD_ABORTED,
getCuid(),
req_->getUri().c_str()),
err);
fileEntry_->addURIResult(req_->getUri(), err.getErrorCode());
requestGroup_->setLastErrorCode(err.getErrorCode());
if(err.getErrorCode() == error_code::CANNOT_RESUME) {
@ -379,19 +348,20 @@ bool AbstractCommand::execute()
onAbort();
tryReserved();
return true;
}
} else {
Timer wakeTime(global::wallclock());
wakeTime.advance(getOption()->getAsInt(PREF_RETRY_WAIT));
req_->setWakeTime(wakeTime);
return prepareForRetry(0);
}
catch (DownloadFailureException &err) {
} catch(DownloadFailureException& err) {
requestGroup_->setLastErrorCode(err.getErrorCode());
if(req_) {
A2_LOG_ERROR_EX(
fmt(MSG_DOWNLOAD_ABORTED, getCuid(), req_->getUri().c_str()),
DL_ABORT_EX2(fmt("URI=%s", req_->getCurrentUri().c_str()), err));
A2_LOG_ERROR_EX(fmt(MSG_DOWNLOAD_ABORTED,
getCuid(),
req_->getUri().c_str()),
DL_ABORT_EX2(fmt("URI=%s", req_->getCurrentUri().c_str()),
err));
fileEntry_->addURIResult(req_->getUri(), err.getErrorCode());
} else {
A2_LOG_ERROR_EX(EX_EXCEPTION_CAUGHT, err);
@ -402,10 +372,10 @@ bool AbstractCommand::execute()
}
}
void AbstractCommand::tryReserved()
{
void AbstractCommand::tryReserved() {
if(getDownloadContext()->getFileEntries().size() == 1) {
const auto &entry = getDownloadContext()->getFirstFileEntry();
const std::shared_ptr<FileEntry>& entry =
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.
@ -417,16 +387,15 @@ void AbstractCommand::tryReserved()
return;
}
}
A2_LOG_DEBUG(
fmt("CUID#%" PRId64 " - Trying reserved/pooled request.", getCuid()));
A2_LOG_DEBUG(fmt("CUID#%" PRId64 " - Trying reserved/pooled request.",
getCuid()));
std::vector<std::unique_ptr<Command>> commands;
requestGroup_->createNextCommand(commands, e_, 1);
e_->setNoWait(true);
e_->addCommand(std::move(commands));
}
bool AbstractCommand::prepareForRetry(time_t wait)
{
bool AbstractCommand::prepareForRetry(time_t wait) {
if(getPieceStorage()) {
getSegmentMan()->cancelSegment(getCuid());
}
@ -439,19 +408,17 @@ bool AbstractCommand::prepareForRetry(time_t wait)
fileEntry_->poolRequest(req_);
A2_LOG_DEBUG(fmt("CUID#%" PRId64 " - Pooling request URI=%s",
getCuid(),
req_->getUri().c_str()));
getCuid(), req_->getUri().c_str()));
if(getSegmentMan()) {
getSegmentMan()->recognizeSegmentFor(fileEntry_);
}
}
auto command =
make_unique<CreateRequestCommand>(getCuid(), requestGroup_, e_);
auto command = make_unique<CreateRequestCommand>(getCuid(),
requestGroup_, e_);
if(wait == 0) {
e_->setNoWait(true);
}
else {
} else {
// We don't use wait so that Command can be executed by
// DownloadEngine::setRefreshInterval(0).
command->setStatus(Command::STATUS_INACTIVE);
@ -460,34 +427,26 @@ bool AbstractCommand::prepareForRetry(time_t wait)
return true;
}
void AbstractCommand::onAbort()
{
void AbstractCommand::onAbort() {
if(req_) {
fileEntry_->removeIdenticalURI(req_->getUri());
fileEntry_->removeRequest(req_);
}
A2_LOG_DEBUG(fmt("CUID#%" PRId64 " - Aborting download", getCuid()));
if (!getPieceStorage()) {
return;
}
A2_LOG_DEBUG(fmt("CUID#%" PRId64 " - Aborting download",
getCuid()));
if(getPieceStorage()) {
getSegmentMan()->cancelSegment(getCuid());
// 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_ ||
getDownloadContext()->getNetStat().getSessionDownloadLength() != 0 ||
requestGroup_->p2pInvolved() ||
getDownloadContext()->getFileEntries().size() != 1) {
return;
}
if(!getOption()->getAsBool(PREF_ALWAYS_RESUME) &&
fileEntry_ &&
getDownloadContext()->getNetStat().getSessionDownloadLength() == 0 &&
!requestGroup_->p2pInvolved() &&
getDownloadContext()->getFileEntries().size() == 1) {
const int maxTries = getOption()->getAsInt(PREF_MAX_RESUME_FAILURE_TRIES);
if (!(maxTries > 0 && requestGroup_->getResumeFailureCount() >= maxTries) &&
!fileEntry_->emptyRequestUri()) {
return;
}
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.
@ -502,18 +461,13 @@ void AbstractCommand::onAbort()
getOption()->put(PREF_ALWAYS_RESUME, A2_V_TRUE);
std::deque<URIResult> res;
fileEntry_->extractURIResult(res, error_code::CANNOT_RESUME);
if (res.empty()) {
return;
}
if(!res.empty()) {
getSegmentMan()->cancelAllSegments();
getSegmentMan()->eraseSegmentWrittenLengthMemo();
getPieceStorage()->markPiecesDone(0);
std::vector<std::string> uris;
uris.reserve(res.size());
std::transform(res.begin(),
res.end(),
std::back_inserter(uris),
std::transform(res.begin(), res.end(), std::back_inserter(uris),
std::mem_fn(&URIResult::getURI));
A2_LOG_DEBUG(fmt("CUID#%" PRId64 " - %lu URIs found.",
getCuid(),
@ -521,94 +475,84 @@ void AbstractCommand::onAbort()
fileEntry_->addUris(uris.begin(), uris.end());
getSegmentMan()->recognizeSegmentFor(fileEntry_);
}
void AbstractCommand::disableReadCheckSocket()
{
if (!checkSocketIsReadable_) {
return;
}
}
}
}
void AbstractCommand::disableReadCheckSocket() {
if(checkSocketIsReadable_) {
e_->deleteSocketForReadCheck(readCheckTarget_, this);
checkSocketIsReadable_ = false;
readCheckTarget_.reset();
}
void
AbstractCommand::setReadCheckSocket(const std::shared_ptr<SocketCore> &socket)
{
if (!socket->isOpen()) {
disableReadCheckSocket();
return;
}
void AbstractCommand::setReadCheckSocket
(const std::shared_ptr<SocketCore>& socket) {
if(!socket->isOpen()) {
disableReadCheckSocket();
} else {
if(checkSocketIsReadable_) {
if(*readCheckTarget_ != *socket) {
e_->deleteSocketForReadCheck(readCheckTarget_, this);
e_->addSocketForReadCheck(socket, this);
readCheckTarget_ = socket;
}
return;
}
} else {
e_->addSocketForReadCheck(socket, this);
checkSocketIsReadable_ = true;
readCheckTarget_ = socket;
}
}
}
void
AbstractCommand::setReadCheckSocketIf(const std::shared_ptr<SocketCore> &socket,
bool pred)
void AbstractCommand::setReadCheckSocketIf
(const std::shared_ptr<SocketCore>& socket, bool pred)
{
if(pred) {
setReadCheckSocket(socket);
return;
}
} else {
disableReadCheckSocket();
}
void AbstractCommand::disableWriteCheckSocket()
{
if (!checkSocketIsWritable_) {
return;
}
void AbstractCommand::disableWriteCheckSocket() {
if(checkSocketIsWritable_) {
e_->deleteSocketForWriteCheck(writeCheckTarget_, this);
checkSocketIsWritable_ = false;
writeCheckTarget_.reset();
}
void
AbstractCommand::setWriteCheckSocket(const std::shared_ptr<SocketCore> &socket)
{
if (!socket->isOpen()) {
disableWriteCheckSocket();
return;
}
void AbstractCommand::setWriteCheckSocket
(const std::shared_ptr<SocketCore>& socket) {
if(!socket->isOpen()) {
disableWriteCheckSocket();
} else {
if(checkSocketIsWritable_) {
if(*writeCheckTarget_ != *socket) {
e_->deleteSocketForWriteCheck(writeCheckTarget_, this);
e_->addSocketForWriteCheck(socket, this);
writeCheckTarget_ = socket;
}
return;
}
} else {
e_->addSocketForWriteCheck(socket, this);
checkSocketIsWritable_ = true;
writeCheckTarget_ = socket;
}
}
}
void AbstractCommand::setWriteCheckSocketIf(
const std::shared_ptr<SocketCore> &socket, bool pred)
void AbstractCommand::setWriteCheckSocketIf
(const std::shared_ptr<SocketCore>& socket, bool pred)
{
if(pred) {
setWriteCheckSocket(socket);
return;
}
} else {
disableWriteCheckSocket();
}
}
void AbstractCommand::swapSocket(std::shared_ptr<SocketCore>& socket)
{
@ -620,15 +564,14 @@ void AbstractCommand::swapSocket(std::shared_ptr<SocketCore> &socket)
namespace {
// Constructs proxy URI, merging username and password if they are
// defined.
std::string makeProxyUri(PrefPtr proxyPref,
std::string makeProxyUri
(PrefPtr proxyPref,
PrefPtr proxyUser,
PrefPtr proxyPasswd,
const Option* option)
{
uri::UriStruct us;
if (!uri::parse(us, option->get(proxyPref))) {
return "";
}
if(uri::parse(us, option->get(proxyPref))) {
if(option->defined(proxyUser)) {
us.username = option->get(proxyUser);
}
@ -637,55 +580,60 @@ std::string makeProxyUri(PrefPtr proxyPref,
us.hasPassword = true;
}
return uri::construct(us);
} else {
return "";
}
}
} // namespace
namespace {
// Returns proxy option value for the given protocol.
std::string getProxyOptionFor(PrefPtr proxyPref,
std::string getProxyOptionFor
(PrefPtr proxyPref,
PrefPtr proxyUser,
PrefPtr proxyPasswd,
const Option* option)
{
std::string uri = makeProxyUri(proxyPref, proxyUser, proxyPasswd, option);
if(uri.empty()) {
return makeProxyUri(
PREF_ALL_PROXY, PREF_ALL_PROXY_USER, PREF_ALL_PROXY_PASSWD, option);
}
return makeProxyUri(PREF_ALL_PROXY, PREF_ALL_PROXY_USER,
PREF_ALL_PROXY_PASSWD, option);
} else {
return uri;
}
}
} // namespace
// Returns proxy URI for given protocol. If no proxy URI is defined,
// then returns an empty string.
std::string getProxyUri(const std::string &protocol, const Option *option)
std::string getProxyUri
(const std::string& protocol, const Option* option)
{
if(protocol == "http") {
return getProxyOptionFor(
PREF_HTTP_PROXY, PREF_HTTP_PROXY_USER, PREF_HTTP_PROXY_PASSWD, option);
}
if (protocol == "https") {
return getProxyOptionFor(PREF_HTTP_PROXY,
PREF_HTTP_PROXY_USER,
PREF_HTTP_PROXY_PASSWD,
option);
} else if(protocol == "https") {
return getProxyOptionFor(PREF_HTTPS_PROXY,
PREF_HTTPS_PROXY_USER,
PREF_HTTPS_PROXY_PASSWD,
option);
}
if (protocol == "ftp") {
return getProxyOptionFor(
PREF_FTP_PROXY, PREF_FTP_PROXY_USER, PREF_FTP_PROXY_PASSWD, option);
}
} else if(protocol == "ftp") {
return getProxyOptionFor(PREF_FTP_PROXY,
PREF_FTP_PROXY_USER,
PREF_FTP_PROXY_PASSWD,
option);
} else {
return A2STR::NIL;
}
}
namespace {
// Returns true if proxy is defined for the given protocol. Otherwise
// returns false.
bool isProxyRequest(const std::string &protocol,
const std::shared_ptr<Option> &option)
bool isProxyRequest
(const std::string& protocol, const std::shared_ptr<Option>& option)
{
std::string proxyUri = getProxyUri(protocol, option.get());
return !proxyUri.empty();
@ -693,36 +641,39 @@ bool isProxyRequest(const std::string &protocol,
} // namespace
namespace {
bool inNoProxy(const std::shared_ptr<Request> &req, const std::string &noProxy)
bool inNoProxy(const std::shared_ptr<Request>& req,
const std::string& noProxy)
{
std::vector<Scip> entries;
util::splitIter(
noProxy.begin(), noProxy.end(), std::back_inserter(entries), ',', true);
util::splitIter(noProxy.begin(), noProxy.end(), std::back_inserter(entries),
',', true);
if(entries.empty()) {
return false;
}
for (const auto &e : entries) {
const auto slashpos = std::find(e.first, e.second, '/');
if (slashpos == e.second) {
if (util::noProxyDomainMatch(req->getHost(),
std::string(e.first, e.second))) {
for(std::vector<Scip>::const_iterator i = entries.begin(),
eoi = entries.end(); i != eoi; ++i) {
std::string::const_iterator slashpos =
std::find((*i).first, (*i).second, '/');
if(slashpos == (*i).second) {
if(util::noProxyDomainMatch
(req->getHost(), std::string((*i).first, (*i).second))) {
return true;
}
}
} else {
// TODO We don't resolve hostname here. More complete
// implementation is that we should first resolve
// hostname(which may result in several IP addresses) and
// evaluates against all of them
std::string ip(e.first, slashpos);
std::string ip((*i).first, slashpos);
uint32_t bits;
if (!util::parseUIntNoThrow(bits, std::string(slashpos + 1, e.second))) {
if(!util::parseUIntNoThrow(bits, std::string(slashpos+1, (*i).second))) {
continue;
}
if(util::inSameCidrBlock(ip, req->getHost(), bits)) {
return true;
}
}
}
return false;
}
} // namespace
@ -739,42 +690,30 @@ std::shared_ptr<Request> AbstractCommand::createProxyRequest() const
if(inNoProxy(req_, getOption()->get(PREF_NO_PROXY))) {
return proxyRequest;
}
std::string proxy = getProxyUri(req_->getProtocol(), getOption().get());
if(!proxy.empty()) {
proxyRequest.reset(new Request());
if(proxyRequest->setUri(proxy)) {
A2_LOG_DEBUG(fmt("CUID#%" PRId64 " - Using proxy", getCuid()));
}
else {
A2_LOG_DEBUG(
fmt("CUID#%" PRId64 " - Failed to parse proxy string", getCuid()));
} else {
A2_LOG_DEBUG(fmt("CUID#%" PRId64 " - Failed to parse proxy string",
getCuid()));
proxyRequest.reset();
}
}
return proxyRequest;
}
std::string AbstractCommand::resolveHostname(std::vector<std::string> &addrs,
const std::string &hostname,
uint16_t port)
std::string AbstractCommand::resolveHostname
(std::vector<std::string>& addrs, const std::string& hostname, uint16_t port)
{
if(util::isNumericHost(hostname)) {
addrs.push_back(hostname);
return hostname;
}
e_->findAllCachedIPAddresses(std::back_inserter(addrs), hostname, port);
if (!addrs.empty()) {
auto ipaddr = addrs.front();
A2_LOG_INFO(fmt(MSG_DNS_CACHE_HIT,
getCuid(),
hostname.c_str(),
strjoin(addrs.begin(), addrs.end(), ", ").c_str()));
return ipaddr;
}
std::string ipaddr;
if(addrs.empty()) {
#ifdef ENABLE_ASYNC_DNS
if(getOption()->getAsBool(PREF_ASYNC_DNS)) {
if(!asyncNameResolverMan_->started()) {
@ -783,31 +722,26 @@ std::string AbstractCommand::resolveHostname(std::vector<std::string> &addrs,
switch(asyncNameResolverMan_->getStatus()) {
case -1:
if(!isProxyRequest(req_->getProtocol(), getOption())) {
e_->getRequestGroupMan()
->getOrCreateServerStat(req_->getHost(), req_->getProtocol())
->setError();
e_->getRequestGroupMan()->getOrCreateServerStat
(req_->getHost(), req_->getProtocol())->setError();
}
throw DL_ABORT_EX2(fmt(MSG_NAME_RESOLUTION_FAILED,
getCuid(),
hostname.c_str(),
throw DL_ABORT_EX2
(fmt(MSG_NAME_RESOLUTION_FAILED, getCuid(), hostname.c_str(),
asyncNameResolverMan_->getLastError().c_str()),
error_code::NAME_RESOLVE_ERROR);
case 0:
return A2STR::NIL;
case 1:
asyncNameResolverMan_->getResolvedAddress(addrs);
if(addrs.empty()) {
throw DL_ABORT_EX2(fmt(MSG_NAME_RESOLUTION_FAILED,
getCuid(),
hostname.c_str(),
throw DL_ABORT_EX2
(fmt(MSG_NAME_RESOLUTION_FAILED, getCuid(), hostname.c_str(),
"No address returned"),
error_code::NAME_RESOLVE_ERROR);
}
break;
}
}
else
} else
#endif // ENABLE_ASYNC_DNS
{
NameResolver res;
@ -821,67 +755,74 @@ std::string AbstractCommand::resolveHostname(std::vector<std::string> &addrs,
getCuid(),
hostname.c_str(),
strjoin(addrs.begin(), addrs.end(), ", ").c_str()));
for (const auto &addr: addrs) {
e_->cacheIPAddress(hostname, addr, port);
for(std::vector<std::string>::const_iterator i = addrs.begin(),
eoi = addrs.end(); i != eoi; ++i) {
e_->cacheIPAddress(hostname, *i, port);
}
ipaddr = e_->findCachedIPAddress(hostname, port);
} else {
ipaddr = addrs.front();
A2_LOG_INFO(fmt(MSG_DNS_CACHE_HIT,
getCuid(),
hostname.c_str(),
strjoin(addrs.begin(), addrs.end(), ", ").c_str()));
}
return ipaddr;
}
void AbstractCommand::prepareForNextAction(
std::unique_ptr<CheckIntegrityEntry> checkEntry)
void AbstractCommand::prepareForNextAction
(std::unique_ptr<CheckIntegrityEntry> checkEntry)
{
std::vector<std::unique_ptr<Command>> commands;
requestGroup_->processCheckIntegrityEntry(
commands, std::move(checkEntry), e_);
requestGroup_->processCheckIntegrityEntry(commands, std::move(checkEntry),
e_);
e_->addCommand(std::move(commands));
e_->setNoWait(true);
}
bool AbstractCommand::checkIfConnectionEstablished(
const std::shared_ptr<SocketCore> &socket,
bool AbstractCommand::checkIfConnectionEstablished
(const std::shared_ptr<SocketCore>& socket,
const std::string& connectedHostname,
const std::string& connectedAddr,
uint16_t connectedPort)
{
std::string error = socket->getSocketError();
if (error.empty()) {
return true;
}
if(!error.empty()) {
// See also InitiateConnectionCommand::executeInternal()
e_->markBadIPAddress(connectedHostname, connectedAddr, connectedPort);
if (e_->findCachedIPAddress(connectedHostname, connectedPort).empty()) {
if(!e_->findCachedIPAddress(connectedHostname, connectedPort).empty()) {
A2_LOG_INFO(fmt(MSG_CONNECT_FAILED_AND_RETRY,
getCuid(),
connectedAddr.c_str(), connectedPort));
e_->setNoWait(true);
e_->addCommand(InitiateConnectionCommandFactory::
createInitiateConnectionCommand
(getCuid(), req_, fileEntry_, requestGroup_, e_));
return false;
}
e_->removeCachedIPAddress(connectedHostname, connectedPort);
// Don't set error if proxy server is used and its method is GET.
if(resolveProxyMethod(req_->getProtocol()) != V_GET ||
!isProxyRequest(req_->getProtocol(), getOption())) {
e_->getRequestGroupMan()
->getOrCreateServerStat(req_->getHost(), req_->getProtocol())
->setError();
e_->getRequestGroupMan()->getOrCreateServerStat
(req_->getHost(), req_->getProtocol())->setError();
}
throw DL_RETRY_EX(fmt(MSG_ESTABLISHING_CONNECTION_FAILED, error.c_str()));
throw DL_RETRY_EX
(fmt(MSG_ESTABLISHING_CONNECTION_FAILED, error.c_str()));
}
return true;
}
A2_LOG_INFO(fmt(MSG_CONNECT_FAILED_AND_RETRY,
getCuid(),
connectedAddr.c_str(),
connectedPort));
e_->setNoWait(true);
e_->addCommand(
InitiateConnectionCommandFactory::createInitiateConnectionCommand(
getCuid(), req_, fileEntry_, requestGroup_, e_));
return false;
}
const std::string&
AbstractCommand::resolveProxyMethod(const std::string &protocol) const
const std::string& AbstractCommand::resolveProxyMethod
(const std::string& protocol) const
{
if (getOption()->get(PREF_PROXY_METHOD) == V_TUNNEL || protocol == "https") {
if(getOption()->get(PREF_PROXY_METHOD) == V_TUNNEL ||
protocol == "https") {
return V_TUNNEL;
}
} else {
return V_GET;
}
}
const std::shared_ptr<Option>& AbstractCommand::getOption() const
{
@ -897,10 +838,10 @@ int32_t AbstractCommand::calculateMinSplitSize() const
{
if(req_ && req_->isPipeliningEnabled()) {
return getDownloadContext()->getPieceLength();
}
} else {
return getOption()->getAsInt(PREF_MIN_SPLIT_SIZE);
}
}
void AbstractCommand::setRequest(const std::shared_ptr<Request>& request)
{
@ -922,8 +863,7 @@ void AbstractCommand::setSocket(const std::shared_ptr<SocketCore>& s)
socket_ = s;
}
const std::shared_ptr<DownloadContext>&
AbstractCommand::getDownloadContext() const
const std::shared_ptr<DownloadContext>& AbstractCommand::getDownloadContext() const
{
return requestGroup_->getDownloadContext();
}
@ -940,13 +880,11 @@ const std::shared_ptr<PieceStorage>& AbstractCommand::getPieceStorage() const
void AbstractCommand::checkSocketRecvBuffer()
{
if (socketRecvBuffer_->bufferEmpty()) {
return;
}
if(!socketRecvBuffer_->bufferEmpty()) {
setStatus(Command::STATUS_ONESHOT_REALTIME);
e_->setNoWait(true);
}
}
void AbstractCommand::addCommandSelf()
{

50
src/AbstractCommand.h
View File

@ -43,8 +43,7 @@
#include "TimerA2.h"
namespace aria2
{
namespace aria2 {
class FileEntry;
class RequestGroup;
@ -63,8 +62,7 @@ class AsyncNameResolver;
class AsyncNameResolverMan;
#endif // ENABLE_ASYNC_DNS
class AbstractCommand : public Command
{
class AbstractCommand : public Command {
private:
std::shared_ptr<Request> req_;
std::shared_ptr<FileEntry> fileEntry_;
@ -92,11 +90,8 @@ private:
bool incNumConnection_;
int32_t calculateMinSplitSize() const;
void useFasterRequest(const std::shared_ptr<Request>& fasterRequest);
bool shouldCheck() const;
public:
RequestGroup* getRequestGroup() const
{
@ -155,32 +150,26 @@ public:
// return empty string. In this case, call this function with same
// arguments until resolved address is returned. Exception is
// thrown on error. port is used for retrieving cached addresses.
std::string resolveHostname(std::vector<std::string> &addrs,
const std::string &hostname,
uint16_t port);
std::string resolveHostname
(std::vector<std::string>& addrs, const std::string& hostname, uint16_t port);
void tryReserved();
void setReadCheckSocket(const std::shared_ptr<SocketCore>& socket);
void setWriteCheckSocket(const std::shared_ptr<SocketCore>& socket);
void disableReadCheckSocket();
void disableWriteCheckSocket();
/**
* If pred == true, calls setReadCheckSocket(socket). Otherwise, calls
* disableReadCheckSocket().
*/
void setReadCheckSocketIf(const std::shared_ptr<SocketCore> &socket,
bool pred);
void setReadCheckSocketIf(const std::shared_ptr<SocketCore>& socket, bool pred);
/**
* If pred == true, calls setWriteCheckSocket(socket). Otherwise, calls
* disableWriteCheckSocket().
*/
void setWriteCheckSocketIf(const std::shared_ptr<SocketCore> &socket,
bool pred);
void setWriteCheckSocketIf(const std::shared_ptr<SocketCore>& socket, bool pred);
// Swaps socket_ with socket. This disables current read and write
// check.
@ -191,10 +180,7 @@ public:
return timeout_;
}
void setTimeout(time_t timeout)
{
timeout_ = timeout;
}
void setTimeout(time_t timeout) { timeout_ = timeout; }
void prepareForNextAction(std::unique_ptr<CheckIntegrityEntry> checkEntry);
@ -203,7 +189,8 @@ public:
// InitiateConnectionCommandFactory and it is pushed to
// DownloadEngine and returns false. If no addresses left, DlRetryEx
// exception is thrown.
bool checkIfConnectionEstablished(const std::shared_ptr<SocketCore> &socket,
bool checkIfConnectionEstablished
(const std::shared_ptr<SocketCore>& socket,
const std::string& connectedHostname,
const std::string& connectedAddr,
uint16_t connectedPort);
@ -238,40 +225,33 @@ public:
void checkSocketRecvBuffer();
void addCommandSelf();
protected:
virtual bool prepareForRetry(time_t wait);
virtual void onAbort();
virtual bool executeInternal() = 0;
// Returns true if the derived class wants to execute
// executeInternal() unconditionally
virtual bool noCheck() const
virtual bool noCheck()
{
return false;
}
public:
AbstractCommand(cuid_t cuid,
const std::shared_ptr<Request> &req,
AbstractCommand(cuid_t cuid, const std::shared_ptr<Request>& req,
const std::shared_ptr<FileEntry>& fileEntry,
RequestGroup *requestGroup,
DownloadEngine *e,
RequestGroup* requestGroup, DownloadEngine* e,
const std::shared_ptr<SocketCore>& s = nullptr,
const std::shared_ptr<SocketRecvBuffer> &socketRecvBuffer =
nullptr,
const std::shared_ptr<SocketRecvBuffer>& socketRecvBuffer = nullptr,
bool incNumConnection = true);
virtual ~AbstractCommand();
virtual bool execute() CXX11_OVERRIDE;
};
// Returns proxy URI for given protocol. If no proxy URI is defined,
// then returns an empty string.
std::string getProxyUri(const std::string &protocol, const Option *option);
std::string getProxyUri
(const std::string& protocol, const Option* option);
} // namespace aria2

2
src/ConnectCommand.cc
View File

@ -115,7 +115,7 @@ bool ConnectCommand::executeInternal()
return true;
}
bool ConnectCommand::noCheck() const
bool ConnectCommand::noCheck()
{
return backupConnectionInfo_ && !backupConnectionInfo_->ipaddr.empty();
}

2
src/ConnectCommand.h
View File

@ -58,7 +58,7 @@ public:
const std::shared_ptr<Request>& getProxyRequest() const;
protected:
virtual bool executeInternal() CXX11_OVERRIDE;
virtual bool noCheck() const CXX11_OVERRIDE;
virtual bool noCheck() CXX11_OVERRIDE;
private:
std::shared_ptr<Request> proxyRequest_;
std::shared_ptr<BackupConnectInfo> backupConnectionInfo_;

5
src/RequestGroup.h
View File

@ -241,11 +241,6 @@ public:
int64_t getCompletedLength() const;
inline int64_t getPendingLength() const
{
return getTotalLength() - getCompletedLength();
}
/**
* Compares expected filename with specified actualFilename.
* The expected filename refers to FileEntry::getBasename() of the first