14 #include <sys/types.h>
18 #include <arpa/inet.h>
32 using namespace zypp::base;
34 #undef CURLVERSION_AT_LEAST
35 #define CURLVERSION_AT_LEAST(M,N,O) LIBCURL_VERSION_NUM >= ((((M)<<8)+(N))<<8)+(O)
44 class multifetchrequest;
58 bool recheckChecksum();
59 void disableCompetition();
62 void adddnsfd(fd_set &rset,
int &maxfd);
63 void dnsevent(fd_set &rset);
87 size_t writefunction(
void *ptr,
size_t size);
88 static size_t _writefunction(
void *ptr,
size_t size,
size_t nmemb,
void *stream);
90 size_t headerfunction(
char *ptr,
size_t size);
91 static size_t _headerfunction(
void *ptr,
size_t size,
size_t nmemb,
void *stream);
104 #define WORKER_STARTING 0
105 #define WORKER_LOOKUP 1
106 #define WORKER_FETCH 2
107 #define WORKER_DISCARD 3
108 #define WORKER_DONE 4
109 #define WORKER_SLEEP 5
110 #define WORKER_BROKEN 6
119 void run(std::vector<Url> &urllist);
164 #define BLKSIZE 131072
174 if (gettimeofday(&tv, NULL))
176 return tv.tv_sec + tv.tv_usec / 1000000.;
180 multifetchworker::writefunction(
void *ptr,
size_t size)
188 len = size > _size ? _size : size;
195 if (_blkstart && _off == _blkstart)
200 (void)curl_easy_getinfo(_curl, CURLINFO_EFFECTIVE_URL, &effurl);
201 if (effurl && !strncasecmp(effurl,
"http", 4))
204 (void)curl_easy_getinfo(_curl, CURLINFO_RESPONSE_CODE, &statuscode);
205 if (statuscode != 206)
213 _request->_lastprogress = now;
219 if (_request->_blklist)
220 _dig.update((
const char *)ptr, len);
225 if (fseeko(_request->_fp, _off, SEEK_SET))
227 cnt = fwrite(ptr, 1, len, _request->_fp);
230 _request->_fetchedsize += cnt;
231 if (_request->_blklist)
232 _dig.update((
const char *)ptr, cnt);
242 multifetchworker::_writefunction(
void *ptr,
size_t size,
size_t nmemb,
void *stream)
249 multifetchworker::headerfunction(
char *p,
size_t size)
252 if (l > 9 && !strncasecmp(p,
"Location:", 9))
254 string line(p + 9, l - 9);
255 if (line[l - 10] ==
'\r')
256 line.erase(l - 10, 1);
257 XXX <<
"#" << _workerno <<
": redirecting to" << line << endl;
260 if (l <= 14 || l >= 128 || strncasecmp(p,
"Content-Range:", 14) != 0)
264 while (l && (*p ==
' ' || *p ==
'\t'))
266 if (l < 6 || strncasecmp(p,
"bytes", 5))
273 unsigned long long start, off, filesize;
274 if (sscanf(buf,
"%llu-%llu/%llu", &start, &off, &filesize) != 3)
276 if (_request->_filesize == (off_t)-1)
278 WAR <<
"#" << _workerno <<
": setting request filesize to " << filesize << endl;
279 _request->_filesize = filesize;
280 if (_request->_totalsize == 0 && !_request->_blklist)
281 _request->_totalsize = filesize;
283 if (_request->_filesize != (off_t)filesize)
285 XXX <<
"#" << _workerno <<
": filesize mismatch" << endl;
287 strncpy(_curlError,
"filesize mismatch", CURL_ERROR_SIZE);
293 multifetchworker::_headerfunction(
void *ptr,
size_t size,
size_t nmemb,
void *stream)
324 XXX <<
"reused worker from pool" << endl;
328 strncpy(
_curlError,
"curl_easy_init failed", CURL_ERROR_SIZE);
337 curl_easy_cleanup(
_curl);
340 strncpy(
_curlError,
"curl_easy_setopt failed", CURL_ERROR_SIZE);
343 curl_easy_setopt(
_curl, CURLOPT_PRIVATE,
this);
346 curl_easy_setopt(
_curl, CURLOPT_WRITEDATA,
this);
350 curl_easy_setopt(
_curl, CURLOPT_HEADERDATA,
this);
364 if (use_auth.empty())
365 use_auth =
"digest,basic";
367 if( auth != CURLAUTH_NONE)
369 XXX <<
"#" <<
_workerno <<
": Enabling HTTP authentication methods: " << use_auth
370 <<
" (CURLOPT_HTTPAUTH=" << auth <<
")" << std::endl;
371 curl_easy_setopt(
_curl, CURLOPT_HTTPAUTH, auth);
386 #if CURLVERSION_AT_LEAST(7,15,5)
387 curl_easy_setopt(
_curl, CURLOPT_MAX_RECV_SPEED_LARGE, (curl_off_t)0);
389 curl_easy_setopt(
_curl, CURLOPT_PRIVATE, (
void *)0);
390 curl_easy_setopt(
_curl, CURLOPT_WRITEFUNCTION, (
void *)0);
391 curl_easy_setopt(
_curl, CURLOPT_WRITEDATA, (
void *)0);
392 curl_easy_setopt(
_curl, CURLOPT_HEADERFUNCTION, (
void *)0);
393 curl_easy_setopt(
_curl, CURLOPT_HEADERDATA, (
void *)0);
397 curl_easy_cleanup(
_curl);
404 while (waitpid(
_pid, &status, 0) == -1)
421 const char *s = getenv(name.c_str());
422 return s && *s ?
true :
false;
438 if (inet_pton(AF_INET, host.c_str(), addrbuf) == 1)
440 if (inet_pton(AF_INET6, host.c_str(), addrbuf) == 1)
451 if (schemeproxy !=
"http_proxy")
453 std::transform(schemeproxy.begin(), schemeproxy.end(), schemeproxy.begin(), ::toupper);
458 XXX <<
"checking DNS lookup of " << host << endl;
463 strncpy(
_curlError,
"DNS pipe creation failed", CURL_ERROR_SIZE);
467 if (
_pid == pid_t(-1))
473 strncpy(
_curlError,
"DNS checker fork failed", CURL_ERROR_SIZE);
480 struct addrinfo *ai, aihints;
481 memset(&aihints, 0,
sizeof(aihints));
482 aihints.ai_family = PF_UNSPEC;
483 int tstsock = socket(PF_INET6, SOCK_DGRAM | SOCK_CLOEXEC, 0);
485 aihints.ai_family = PF_INET;
488 aihints.ai_socktype = SOCK_STREAM;
489 aihints.ai_flags = AI_CANONNAME;
492 alarm(connecttimeout);
493 signal(SIGALRM, SIG_DFL);
494 if (getaddrinfo(host.c_str(), NULL, &aihints, &ai))
520 while (waitpid(
_pid, &status, 0) == -1)
531 if (!WIFEXITED(status))
534 strncpy(
_curlError,
"DNS lookup failed", CURL_ERROR_SIZE);
538 int exitcode = WEXITSTATUS(status);
539 XXX <<
"#" <<
_workerno <<
": DNS lookup returned " << exitcode << endl;
543 strncpy(
_curlError,
"DNS lookup failed", CURL_ERROR_SIZE);
573 size_t cnt = l >
sizeof(buf) ?
sizeof(buf) : l;
588 XXX <<
"start stealing!" << endl;
592 std::list<multifetchworker *>::iterator workeriter =
_request->
_workers.begin();
599 if (worker->
_pass == -1)
658 XXX <<
"#" <<
_workerno <<
": going to sleep for " << sl * 1000 <<
" ms" << endl;
679 std::list<multifetchworker *>::iterator workeriter =
_request->
_workers.begin();
752 sprintf(rangebuf,
"%llu-", (
unsigned long long)
_blkstart);
754 sprintf(rangebuf,
"%llu-%llu", (
unsigned long long)_blkstart, (
unsigned long long)_blkstart +
_blksize - 1);
756 if (curl_easy_setopt(
_curl, CURLOPT_RANGE, !
_noendrange || _blkstart != 0 ? rangebuf : (
char *)0) != CURLE_OK)
760 strncpy(
_curlError,
"curl_easy_setopt range failed", CURL_ERROR_SIZE);
767 strncpy(
_curlError,
"curl_multi_add_handle failed", CURL_ERROR_SIZE);
817 for (
size_t blkno = 0; blkno < blklist->
numBlocks(); blkno++)
823 else if (filesize != off_t(-1))
829 for (std::list<multifetchworker *>::iterator workeriter =
_workers.begin(); workeriter !=
_workers.end(); ++workeriter)
842 std::vector<Url>::iterator urliter = urllist.begin();
845 fd_set rset, wset, xset;
850 XXX <<
"finished!" << endl;
874 WAR <<
"No more active workers!" << endl;
876 for (std::list<multifetchworker *>::iterator workeriter =
_workers.begin(); workeriter !=
_workers.end(); ++workeriter)
889 curl_multi_fdset(
_multi, &rset, &wset, &xset, &maxfd);
892 for (std::list<multifetchworker *>::iterator workeriter =
_workers.begin(); workeriter !=
_workers.end(); ++workeriter)
893 (*workeriter)->adddnsfd(rset, maxfd);
904 for (std::list<multifetchworker *>::iterator workeriter =
_workers.begin(); workeriter !=
_workers.end(); ++workeriter)
920 tv.tv_usec = sl * 1000000;
922 int r = select(maxfd + 1, &rset, &wset, &xset, &tv);
923 if (r == -1 && errno != EINTR)
926 for (std::list<multifetchworker *>::iterator workeriter =
_workers.begin(); workeriter !=
_workers.end(); ++workeriter)
942 mcode = curl_multi_perform(
_multi, &tasks);
943 if (mcode == CURLM_CALL_MULTI_PERFORM)
945 if (mcode != CURLM_OK)
966 for (std::list<multifetchworker *>::iterator workeriter =
_workers.begin(); workeriter !=
_workers.end(); ++workeriter)
975 XXX <<
"#" << worker->
_workerno <<
": sleep done, wake up" << endl;
984 while ((msg = curl_multi_info_read(
_multi, &nqueue)) != 0)
986 if (msg->msg != CURLMSG_DONE)
988 CURL *easy = msg->easy_handle;
989 CURLcode cc = msg->data.result;
991 if (curl_easy_getinfo(easy, CURLINFO_PRIVATE, &worker) != CURLE_OK)
1001 curl_multi_remove_handle(
_multi, easy);
1002 if (cc == CURLE_HTTP_RETURNED_ERROR)
1004 long statuscode = 0;
1005 (void)curl_easy_getinfo(easy, CURLINFO_RESPONSE_CODE, &statuscode);
1006 XXX <<
"HTTP status " << statuscode << endl;
1007 if (statuscode == 416 && !
_blklist)
1013 XXX <<
"#" << worker->
_workerno <<
": retrying with no end range" << endl;
1033 WAR <<
"#" << worker->
_workerno <<
": checksum error, disable worker" << endl;
1035 strncpy(worker->
_curlError,
"checksum error", CURL_ERROR_SIZE);
1049 XXX <<
"#" << worker->
_workerno <<
": recheck checksum error, refetch block" << endl;
1062 int maxworkerno = 0;
1064 for (std::list<multifetchworker *>::iterator workeriter =
_workers.begin(); workeriter !=
_workers.end(); ++workeriter)
1079 double ratio = worker->
_avgspeed / maxavg;
1082 ratio = ratio * ratio;
1085 XXX <<
"#" << worker->
_workerno <<
": too slow ("<< ratio <<
", " << worker->
_avgspeed <<
", #" << maxworkerno <<
": " << maxavg <<
"), going to sleep for " << ratio * 1000 <<
" ms" << endl;
1106 #if CURLVERSION_AT_LEAST(7,15,5)
1107 curl_easy_setopt(worker->
_curl, CURLOPT_MAX_RECV_SPEED_LARGE, (curl_off_t)(avg));
1149 WAR <<
"overall result" << endl;
1150 for (std::list<multifetchworker *>::iterator workeriter =
_workers.begin(); workeriter !=
_workers.end(); ++workeriter)
1164 MIL <<
"MediaMultiCurl::MediaMultiCurl(" << url_r <<
", " << attach_point_hint_r <<
")" << endl;
1178 curl_multi_cleanup(
_multi);
1181 std::map<std::string, CURL *>::iterator it;
1184 CURL *easy = it->second;
1187 curl_easy_cleanup(easy);
1203 for (; sl; sl = sl->next)
1212 while ((l = pread(fd, buf,
sizeof(buf) - 1, (off_t)0)) == -1 && errno == EINTR)
1218 while (*p ==
' ' || *p ==
'\t' || *p ==
'\r' || *p ==
'\n')
1220 if (!strncasecmp(p,
"<?xml", 5))
1222 while (*p && *p !=
'>')
1226 while (*p ==
' ' || *p ==
'\t' || *p ==
'\r' || *p ==
'\n')
1229 bool ret = !strncasecmp(p,
"<metalink", 9) ?
true :
false;
1236 if ((fd = open(file.asString().c_str(), O_RDONLY|O_CLOEXEC)) == -1)
1240 DBG <<
"looks_like_metalink(" << file <<
"): " << ret << endl;
1256 if ( curl_easy_getinfo( _curl, CURLINFO_PRIVATE, &fp ) != CURLE_OK || !fp )
1258 if ( ftell( fp ) == 0 )
1263 long httpReturnCode = 0;
1264 if (curl_easy_getinfo(_curl, CURLINFO_RESPONSE_CODE, &httpReturnCode ) != CURLE_OK || httpReturnCode == 0)
1268 bool ismetalink =
false;
1269 if (curl_easy_getinfo(_curl, CURLINFO_CONTENT_TYPE, &ptr) == CURLE_OK && ptr)
1271 string ct = string(ptr);
1272 if (ct.find(
"application/metalink+xml") == 0 || ct.find(
"application/metalink4+xml") == 0)
1275 if (!ismetalink && dlnow < 256)
1284 DBG <<
"looks_like_metalink_fd: " << ismetalink << endl;
1300 Pathname dest = target.absolutename();
1303 DBG <<
"assert_dir " << dest.dirname() <<
" failed" << endl;
1307 ManagedFile destNew { target.extend(
".new.zypp.XXXXXX" ) };
1313 ERR <<
"out of memory for temp file name" << endl;
1317 AutoFD tmp_fd { ::mkostemp( buf, O_CLOEXEC ) };
1320 ERR <<
"mkstemp failed for file '" << destNew <<
"'" << endl;
1325 file = ::fdopen( tmp_fd,
"we" );
1328 ERR <<
"fopen failed for file '" << destNew <<
"'" << endl;
1331 tmp_fd.resetDispose();
1334 DBG <<
"dest: " << dest << endl;
1335 DBG <<
"temp: " << destNew << endl;
1340 curl_easy_setopt(
_curl, CURLOPT_TIMECONDITION, CURL_TIMECOND_IFMODSINCE);
1341 curl_easy_setopt(
_curl, CURLOPT_TIMEVALUE, (
long)PathInfo(target).mtime());
1345 curl_easy_setopt(
_curl, CURLOPT_TIMECONDITION, CURL_TIMECOND_NONE);
1346 curl_easy_setopt(
_curl, CURLOPT_TIMEVALUE, 0L);
1352 curl_easy_setopt(
_curl, CURLOPT_PRIVATE, (*file) );
1359 curl_easy_setopt(
_curl, CURLOPT_TIMECONDITION, CURL_TIMECOND_NONE);
1360 curl_easy_setopt(
_curl, CURLOPT_TIMEVALUE, 0L);
1362 curl_easy_setopt(
_curl, CURLOPT_PRIVATE, (
void *)0);
1365 curl_easy_setopt(
_curl, CURLOPT_TIMECONDITION, CURL_TIMECOND_NONE);
1366 curl_easy_setopt(
_curl, CURLOPT_TIMEVALUE, 0L);
1368 curl_easy_setopt(
_curl, CURLOPT_PRIVATE, (
void *)0);
1369 long httpReturnCode = 0;
1370 CURLcode infoRet = curl_easy_getinfo(
_curl, CURLINFO_RESPONSE_CODE, &httpReturnCode);
1371 if (infoRet == CURLE_OK)
1374 if ( httpReturnCode == 304
1377 DBG <<
"not modified: " << PathInfo(dest) << endl;
1383 WAR <<
"Could not get the reponse code." << endl;
1386 bool ismetalink =
false;
1389 if (curl_easy_getinfo(
_curl, CURLINFO_CONTENT_TYPE, &ptr) == CURLE_OK && ptr)
1391 string ct = string(ptr);
1392 if (ct.find(
"application/metalink+xml") == 0 || ct.find(
"application/metalink4+xml") == 0)
1407 bool userabort =
false;
1415 vector<Url> urls = mlp.
getUrls();
1417 file = fopen((*destNew).c_str(),
"w+e");
1420 if (PathInfo(target).isExist())
1422 XXX <<
"reusing blocks from file " << target << endl;
1426 if (bl.
haveChecksum(1) && PathInfo(failedFile).isExist())
1428 XXX <<
"reusing blocks from file " << failedFile << endl;
1436 XXX <<
"reusing blocks from file " << df << endl;
1442 multifetch(filename, file, &urls, &report, &bl, expectedFileSize_r);
1446 userabort = ex.
errstr() ==
"User abort";
1457 if (PathInfo(destNew).size() >= 63336)
1459 ::unlink(failedFile.asString().c_str());
1466 file = fopen((*destNew).c_str(),
"w+e");
1475 ERR <<
"Failed to chmod file " << destNew << endl;
1482 ERR <<
"Fclose failed for file '" << destNew <<
"'" << endl;
1486 if (
rename( destNew, dest ) != 0 )
1488 ERR <<
"Rename failed" << endl;
1491 destNew.resetDispose();
1493 DBG <<
"done: " << PathInfo(dest) << endl;
1499 inline Url propagateQueryParams(
Url url_r,
const Url & template_r )
1501 for ( std::string param : {
"proxy",
"proxyport",
"proxyuser",
"proxypass"} )
1503 const std::string & value( template_r.
getQueryParam( param ) );
1504 if ( ! value.empty() )
1515 if (blklist && filesize == off_t(-1) && blklist->
haveFilesize())
1517 if (blklist && !blklist->
haveBlocks() && filesize != 0)
1519 if (blklist && (filesize == 0 || !blklist->
numBlocks()))
1528 _multi = curl_multi_init();
1542 std::vector<Url> myurllist;
1543 for (std::vector<Url>::iterator urliter = urllist->begin(); urliter != urllist->end(); ++urliter)
1547 string scheme = urliter->getScheme();
1548 if (scheme ==
"http" || scheme ==
"https" || scheme ==
"ftp" || scheme ==
"tftp")
1551 myurllist.push_back(propagateQueryParams(*urliter,
_url));
1558 if (!myurllist.size())
1559 myurllist.push_back(baseurl);
1568 if (fseeko(fp, off_t(0), SEEK_SET))
1574 while ((l = fread(buf, 1,
sizeof(buf), fp)) > 0)
1582 return _dnsok.find(host) ==
_dnsok.end() ?
false :
true;
1604 curl_easy_cleanup(oldeasy);
int assert_dir(const Pathname &path, unsigned mode)
Like 'mkdir -p'.
void setQueryParam(const std::string ¶m, const std::string &value)
Set or add value for the specified query parameter.
#define ZYPP_THROW(EXCPT)
Drops a logline and throws the Exception.
static ZConfig & instance()
Singleton ctor.
Compute Message Digests (MD5, SHA1 etc)
Store and operate with byte count.
std::string getHost(EEncoding eflag=zypp::url::E_DECODED) const
Returns the hostname or IP from the URL authority.
Pathname repoCachePath() const
Path where the caches are kept (/var/cache/zypp)
AutoDispose< const Pathname > ManagedFile
A Pathname plus associated cleanup code to be executed when path is no longer needed.
static const Unit MB
1000^2 Byte
AutoDispose<int> calling ::close
std::string asString() const
Returns a default string representation of the Url object.
#define ZYPP_RETHROW(EXCPT)
Drops a logline and rethrows, updating the CodeLocation.
int unlink(const Pathname &path)
Like 'unlink'.
int rename(const Pathname &oldpath, const Pathname &newpath)
Like 'rename'.
int hardlinkCopy(const Pathname &oldpath, const Pathname &newpath)
Create newpath as hardlink or copy of oldpath.
std::string getQueryParam(const std::string ¶m, EEncoding eflag=zypp::url::E_DECODED) const
Return the value for the specified query parameter.
std::string numstring(char n, int w=0)
void resetDispose()
Set no dispose function.
Base class for Exception.
Reference counted access to a Tp object calling a custom Dispose function when the last AutoDispose h...
AutoDispose<FILE*> calling ::fclose
AutoDispose< void * > _state
mode_t applyUmaskTo(mode_t mode_r)
Modify mode_r according to the current umask ( mode_r & ~getUmask() ).
std::string getScheme() const
Returns the scheme name of the URL.
bool update(const char *bytes, size_t len)
feed data into digest computation algorithm
ByteCount df(const Pathname &path_r)
Report free disk space on a mounted file system.