zypp/media/MediaMultiCurl.cc

   1 /*---------------------------------------------------------------------\
   2 |                          ____ _   __ __ ___                          |
   3 |                         |__  / \ / / . \ . \                         |
   4 |                           / / \ V /|  _/  _/                         |
   5 |                          / /__ | | | | | |                           |
   6 |                         /_____||_| |_| |_|                           |
   7 |                                                                      |
   8 \---------------------------------------------------------------------*/
   9 /** \file zypp/media/MediaMultiCurl.cc
  10  *
  11 */
  12
  13 #include <ctype.h>
  14 #include <sys/types.h>
  15 #include <signal.h>
  16 #include <sys/wait.h>
  17 #include <netdb.h>
  18 #include <arpa/inet.h>
  19
  20 #include <vector>
  21 #include <iostream>
  22 #include <algorithm>
  23
  24
  25 #include "zypp/base/Logger.h"
  26 #include "zypp/media/MediaMultiCurl.h"
  27 #include "zypp/media/MetaLinkParser.h"
  28
  29 using namespace std;
  30 using namespace zypp::base;
  31
  32 namespace zypp {
  33   namespace media {
  34
  35
  36 //////////////////////////////////////////////////////////////////////
  37
  38
  39 class multifetchrequest;
  40
  41 // Hack: we derive from MediaCurl just to get the storage space for
  42 // settings, url, curlerrors and the like
  43
  44 class multifetchworker : MediaCurl {
  45   friend class multifetchrequest;
  46
  47 public:
  48   multifetchworker(int no, multifetchrequest &request, const Url &url);
  49   ~multifetchworker();
  50   void nextjob();
  51   void run();
  52   bool checkChecksum();
  53   bool recheckChecksum();
  54   void disableCompetition();
  55
  56   void checkdns();
  57   void adddnsfd(fd_set &rset, int &maxfd);
  58   void dnsevent(fd_set &rset);
  59
  60   int _workerno;
  61
  62   int _state;
  63   bool _competing;
  64
  65   size_t _blkno;
  66   off_t _blkstart;
  67   size_t _blksize;
  68   bool _noendrange;
  69
  70   double _blkstarttime;
  71   size_t _blkreceived;
  72   off_t  _received;
  73
  74   double _avgspeed;
  75   double _maxspeed;
  76
  77   double _sleepuntil;
  78
  79 private:
  80   void stealjob();
  81
  82   size_t writefunction(void *ptr, size_t size);
  83   static size_t _writefunction(void *ptr, size_t size, size_t nmemb, void *stream);
  84
  85   size_t headerfunction(char *ptr, size_t size);
  86   static size_t _headerfunction(void *ptr, size_t size, size_t nmemb, void *stream);
  87
  88   multifetchrequest *_request;
  89   int _pass;
  90   string _urlbuf;
  91   off_t _off;
  92   size_t _size;
  93   Digest _dig;
  94
  95   pid_t _pid;
  96   int _dnspipe;
  97 };
  98
  99 #define WORKER_STARTING 0
 100 #define WORKER_LOOKUP   1
 101 #define WORKER_FETCH    2
 102 #define WORKER_DISCARD  3
 103 #define WORKER_DONE     4
 104 #define WORKER_SLEEP    5
 105 #define WORKER_BROKEN   6
 106
 107
 108
 109 class multifetchrequest {
 110 public:
 111   multifetchrequest(const MediaMultiCurl *context, const Pathname &filename, const Url &baseurl, CURLM *multi, FILE *fp, callback::SendReport<DownloadProgressReport> *report, MediaBlockList *blklist, off_t filesize);
 112   ~multifetchrequest();
 113
 114   void run(std::vector<Url> &urllist);
 115
 116 protected:
 117   friend class multifetchworker;
 118
 119   const MediaMultiCurl *_context;
 120   const Pathname _filename;
 121   Url _baseurl;
 122
 123   FILE *_fp;
 124   callback::SendReport<DownloadProgressReport> *_report;
 125   MediaBlockList *_blklist;
 126   off_t _filesize;
 127
 128   CURLM *_multi;
 129
 130   std::list<multifetchworker *> _workers;
 131   bool _stealing;
 132   bool _havenewjob;
 133
 134   size_t _blkno;
 135   off_t _blkoff;
 136   size_t _activeworkers;
 137   size_t _lookupworkers;
 138   size_t _sleepworkers;
 139   double _minsleepuntil;
 140   bool _finished;
 141   off_t _totalsize;
 142   off_t _fetchedsize;
 143   off_t _fetchedgoodsize;
 144
 145   double _starttime;
 146   double _lastprogress;
 147
 148   double _lastperiodstart;
 149   double _lastperiodfetched;
 150   double _periodavg;
 151
 152 public:
 153   double _timeout;
 154   double _connect_timeout;
 155   double _maxspeed;
 156 };
 157
 158 #define BLKSIZE         131072
 159 #define MAXWORKERS      5
 160 #define MAXURLS         10
 161
 162
 163 //////////////////////////////////////////////////////////////////////
 164
 165 static double
 166 currentTime()
 167 {
 168   struct timeval tv;
 169   if (gettimeofday(&tv, NULL))
 170     return 0;
 171   return tv.tv_sec + tv.tv_usec / 1000000.;
 172 }
 173
 174 size_t
 175 multifetchworker::writefunction(void *ptr, size_t size)
 176 {
 177   size_t len, cnt;
 178   if (_state == WORKER_BROKEN)
 179     return size ? 0 : 1;
 180
 181   double now = currentTime();
 182
 183   len = size > _size ? _size : size;
 184   if (!len)
 185     {
 186       // kill this job?
 187       return size;
 188     }
 189
 190   _blkreceived += len;
 191   _received += len;
 192
 193   _request->_lastprogress = now;
 194
 195   if (_state == WORKER_DISCARD || !_request->_fp)
 196     {
 197       // block is no longer needed
 198       // still calculate the checksum so that we can throw out bad servers
 199       if (_request->_blklist)
 200         _dig.update((const char *)ptr, len);
 201       _off += len;
 202       _size -= len;
 203       return size;
 204     }
 205   if (fseeko(_request->_fp, _off, SEEK_SET))
 206     return size ? 0 : 1;
 207   cnt = fwrite(ptr, 1, len, _request->_fp);
 208   if (cnt > 0)
 209     {
 210       _request->_fetchedsize += cnt;
 211       if (_request->_blklist)
 212         _dig.update((const char *)ptr, cnt);
 213       _off += cnt;
 214       _size -= cnt;
 215       if (cnt == len)
 216         return size;
 217     }
 218   return cnt;
 219 }
 220
 221 size_t
 222 multifetchworker::_writefunction(void *ptr, size_t size, size_t nmemb, void *stream)
 223 {
 224   multifetchworker *me = reinterpret_cast<multifetchworker *>(stream);
 225   return me->writefunction(ptr, size * nmemb);
 226 }
 227
 228 size_t
 229 multifetchworker::headerfunction(char *p, size_t size)
 230 {
 231   size_t l = size;
 232   if (l > 9 && !strncasecmp(p, "Location:", 9))
 233     {
 234       string line(p + 9, l - 9);
 235       if (line[l - 10] == '\r')
 236         line.erase(l - 10, 1);
 237       DBG << "#" << _workerno << ": redirecting to" << line << endl;
 238       return size;
 239     }
 240   if (l <= 14 || l >= 128 || strncasecmp(p, "Content-Range:", 14) != 0)
 241     return size;
 242   p += 14;
 243   l -= 14;
 244   while (l && (*p == ' ' || *p == '\t'))
 245     p++, l--;
 246   if (l < 6 || strncasecmp(p, "bytes", 5))
 247     return size;
 248   p += 5;
 249   l -= 5;
 250   char buf[128];
 251   memcpy(buf, p, l);
 252   buf[l] = 0;
 253   unsigned long long start, off, filesize;
 254   if (sscanf(buf, "%llu-%llu/%llu", &start, &off, &filesize) != 3)
 255     return size;
 256   if (_request->_filesize == (off_t)-1)
 257     {
 258       WAR << "#" << _workerno << ": setting request filesize to " << filesize << endl;
 259       _request->_filesize = filesize;
 260     }
 261   if (_request->_filesize != (off_t)filesize)
 262     {
 263       DBG << "#" << _workerno << ": filesize mismatch" << endl;
 264       _state = WORKER_BROKEN;
 265       strncpy(_curlError, "filesize mismatch", CURL_ERROR_SIZE);
 266     }
 267   return size;
 268 }
 269
 270 size_t
 271 multifetchworker::_headerfunction(void *ptr, size_t size, size_t nmemb, void *stream)
 272 {
 273   multifetchworker *me = reinterpret_cast<multifetchworker *>(stream);
 274   return me->headerfunction((char *)ptr, size * nmemb);
 275 }
 276
 277 multifetchworker::multifetchworker(int no, multifetchrequest &request, const Url &url)
 278 : MediaCurl(url, Pathname())
 279 {
 280   _workerno = no;
 281   _request = &request;
 282   _state = WORKER_STARTING;
 283   _competing = false;
 284   _off = _blkstart = 0;
 285   _size = _blksize = 0;
 286   _pass = 0;
 287   _blkno = 0;
 288   _pid = 0;
 289   _dnspipe = -1;
 290   _blkreceived = 0;
 291   _received = 0;
 292   _blkstarttime = 0;
 293   _avgspeed = 0;
 294   _sleepuntil = 0;
 295   _maxspeed = _request->_maxspeed;
 296   _noendrange = false;
 297
 298   Url curlUrl(url);
 299   curlUrl.setUsername( "" );
 300   curlUrl.setPassword( "" );
 301   curlUrl.setPathParams( "" );
 302   curlUrl.setQueryString( "" );
 303   curlUrl.setFragment( "" );
 304   _urlbuf = curlUrl.asString();
 305   _curl = _request->_context->fromEasyPool(_url.getHost());
 306   if (_curl)
 307     DBG << "reused worker from pool" << endl;
 308   if (!_curl && !(_curl = curl_easy_init()))
 309     {
 310       _state = WORKER_BROKEN;
 311       strncpy(_curlError, "curl_easy_init failed", CURL_ERROR_SIZE);
 312       return;
 313     }
 314   try
 315     {
 316       setupEasy();
 317     }
 318   catch (Exception &ex)
 319     {
 320       curl_easy_cleanup(_curl);
 321       _curl = 0;
 322       _state = WORKER_BROKEN;
 323       strncpy(_curlError, "curl_easy_setopt failed", CURL_ERROR_SIZE);
 324       return;
 325     }
 326   curl_easy_setopt(_curl, CURLOPT_PRIVATE, this);
 327   curl_easy_setopt(_curl, CURLOPT_URL, _urlbuf.c_str());
 328   curl_easy_setopt(_curl, CURLOPT_WRITEFUNCTION, &_writefunction);
 329   curl_easy_setopt(_curl, CURLOPT_WRITEDATA, this);
 330   if (_request->_filesize == off_t(-1) || !_request->_blklist || !_request->_blklist->haveChecksum(0))
 331     {
 332       curl_easy_setopt(_curl, CURLOPT_HEADERFUNCTION, &_headerfunction);
 333       curl_easy_setopt(_curl, CURLOPT_HEADERDATA, this);
 334     }
 335   // if this is the same host copy authorization
 336   // (the host check is also what curl does when doing a redirect)
 337   // (note also that unauthorized exceptions are thrown with the request host)
 338   if (url.getHost() == _request->_context->_url.getHost())
 339     {
 340       _settings.setUsername(_request->_context->_settings.username());
 341       _settings.setPassword(_request->_context->_settings.password());
 342       _settings.setAuthType(_request->_context->_settings.authType());
 343       if ( _settings.userPassword().size() )
 344         {
 345           curl_easy_setopt(_curl, CURLOPT_USERPWD, _settings.userPassword().c_str());
 346           string use_auth = _settings.authType();
 347           if (use_auth.empty())
 348             use_auth = "digest,basic";        // our default
 349           long auth = CurlAuthData::auth_type_str2long(use_auth);
 350           if( auth != CURLAUTH_NONE)
 351           {
 352             DBG << "#" << _workerno << ": Enabling HTTP authentication methods: " << use_auth
 353                 << " (CURLOPT_HTTPAUTH=" << auth << ")" << std::endl;
 354             curl_easy_setopt(_curl, CURLOPT_HTTPAUTH, auth);
 355           }
 356         }
 357     }
 358   checkdns();
 359 }
 360
 361 multifetchworker::~multifetchworker()
 362 {
 363   if (_curl)
 364     {
 365       if (_state == WORKER_FETCH || _state == WORKER_DISCARD)
 366         curl_multi_remove_handle(_request->_multi, _curl);
 367       if (_state == WORKER_DONE || _state == WORKER_SLEEP)
 368         {
 369           curl_easy_setopt(_curl, CURLOPT_MAX_RECV_SPEED_LARGE, (curl_off_t)0);
 370           curl_easy_setopt(_curl, CURLOPT_PRIVATE, (void *)0);
 371           curl_easy_setopt(_curl, CURLOPT_WRITEFUNCTION, (void *)0);
 372           curl_easy_setopt(_curl, CURLOPT_WRITEDATA, (void *)0);
 373           curl_easy_setopt(_curl, CURLOPT_HEADERFUNCTION, (void *)0);
 374           curl_easy_setopt(_curl, CURLOPT_HEADERDATA, (void *)0);
 375           _request->_context->toEasyPool(_url.getHost(), _curl);
 376         }
 377       else
 378         curl_easy_cleanup(_curl);
 379       _curl = 0;
 380     }
 381   if (_pid)
 382     {
 383       kill(_pid, SIGKILL);
 384       int status;
 385       while (waitpid(_pid, &status, 0) == -1)
 386         if (errno != EINTR)
 387           break;
 388       _pid = 0;
 389     }
 390   if (_dnspipe != -1)
 391     {
 392       close(_dnspipe);
 393       _dnspipe = -1;
 394     }
 395   // the destructor in MediaCurl doesn't call disconnect() if
 396   // the media is not attached, so we do it here manually
 397   disconnectFrom();
 398 }
 399
 400 static inline bool env_isset(string name)
 401 {
 402   const char *s = getenv(name.c_str());
 403   return s && *s ? true : false;
 404 }
 405
 406 void
 407 multifetchworker::checkdns()
 408 {
 409   string host = _url.getHost();
 410
 411   if (host.empty())
 412     return;
 413
 414   if (_request->_context->isDNSok(host))
 415     return;
 416
 417   // no need to do dns checking for numeric hosts
 418   char addrbuf[128];
 419   if (inet_pton(AF_INET, host.c_str(), addrbuf) == 1)
 420     return;
 421   if (inet_pton(AF_INET6, host.c_str(), addrbuf) == 1)
 422     return;
 423
 424   // no need to do dns checking if we use a proxy
 425   if (!_settings.proxy().empty())
 426     return;
 427   if (env_isset("all_proxy") || env_isset("ALL_PROXY"))
 428     return;
 429   string schemeproxy = _url.getScheme() + "_proxy";
 430   if (env_isset(schemeproxy))
 431     return;
 432   if (schemeproxy != "http_proxy")
 433     {
 434       std::transform(schemeproxy.begin(), schemeproxy.end(), schemeproxy.begin(), ::toupper);
 435       if (env_isset(schemeproxy))
 436         return;
 437     }
 438
 439   DBG << "checking DNS lookup of " << host << endl;
 440   int pipefds[2];
 441   if (pipe(pipefds))
 442     {
 443       _state = WORKER_BROKEN;
 444       strncpy(_curlError, "DNS pipe creation failed", CURL_ERROR_SIZE);
 445       return;
 446     }
 447   _pid = fork();
 448   if (_pid == pid_t(-1))
 449     {
 450       close(pipefds[0]);
 451       close(pipefds[1]);
 452       _pid = 0;
 453       _state = WORKER_BROKEN;
 454       strncpy(_curlError, "DNS checker fork failed", CURL_ERROR_SIZE);
 455       return;
 456     }
 457   else if (_pid == 0)
 458     {
 459       close(pipefds[0]);
 460       // XXX: close all other file descriptors
 461       struct addrinfo *ai, aihints;
 462       memset(&aihints, 0, sizeof(aihints));
 463       aihints.ai_family = PF_UNSPEC;
 464       int tstsock = socket(PF_INET6, SOCK_DGRAM, 0);
 465       if (tstsock == -1)
 466         aihints.ai_family = PF_INET;
 467       else
 468         close(tstsock);
 469       aihints.ai_socktype = SOCK_STREAM;
 470       aihints.ai_flags = AI_CANONNAME;
 471       unsigned int connecttimeout = _request->_connect_timeout;
 472       if (connecttimeout)
 473         alarm(connecttimeout);
 474       signal(SIGALRM, SIG_DFL);
 475       if (getaddrinfo(host.c_str(), NULL, &aihints, &ai))
 476         _exit(1);
 477       _exit(0);
 478     }
 479   close(pipefds[1]);
 480   _dnspipe = pipefds[0];
 481   _state = WORKER_LOOKUP;
 482 }
 483
 484 void
 485 multifetchworker::adddnsfd(fd_set &rset, int &maxfd)
 486 {
 487   if (_state != WORKER_LOOKUP)
 488     return;
 489   FD_SET(_dnspipe, &rset);
 490   if (maxfd < _dnspipe)
 491     maxfd = _dnspipe;
 492 }
 493
 494 void
 495 multifetchworker::dnsevent(fd_set &rset)
 496 {
 497
 498   if (_state != WORKER_LOOKUP || !FD_ISSET(_dnspipe, &rset))
 499     return;
 500   int status;
 501   while (waitpid(_pid, &status, 0) == -1)
 502     {
 503       if (errno != EINTR)
 504         return;
 505     }
 506   _pid = 0;
 507   if (_dnspipe != -1)
 508     {
 509       close(_dnspipe);
 510       _dnspipe = -1;
 511     }
 512   if (!WIFEXITED(status))
 513     {
 514       _state = WORKER_BROKEN;
 515       strncpy(_curlError, "DNS lookup failed", CURL_ERROR_SIZE);
 516       _request->_activeworkers--;
 517       return;
 518     }
 519   int exitcode = WEXITSTATUS(status);
 520   DBG << "#" << _workerno << ": DNS lookup returned " << exitcode << endl;
 521   if (exitcode != 0)
 522     {
 523       _state = WORKER_BROKEN;
 524       strncpy(_curlError, "DNS lookup failed", CURL_ERROR_SIZE);
 525       _request->_activeworkers--;
 526       return;
 527     }
 528   _request->_context->setDNSok(_url.getHost());
 529   nextjob();
 530 }
 531
 532 bool
 533 multifetchworker::checkChecksum()
 534 {
 535   // DBG << "checkChecksum block " << _blkno << endl;
 536   if (!_blksize || !_request->_blklist)
 537     return true;
 538   return _request->_blklist->verifyDigest(_blkno, _dig);
 539 }
 540
 541 bool
 542 multifetchworker::recheckChecksum()
 543 {
 544   // DBG << "recheckChecksum block " << _blkno << endl;
 545   if (!_request->_fp || !_blksize || !_request->_blklist)
 546     return true;
 547   if (fseeko(_request->_fp, _blkstart, SEEK_SET))
 548     return false;
 549   char buf[4096];
 550   size_t l = _blksize;
 551   _request->_blklist->createDigest(_dig);       // resets digest
 552   while (l)
 553     {
 554       size_t cnt = l > sizeof(buf) ? sizeof(buf) : l;
 555       if (fread(buf, cnt, 1, _request->_fp) != 1)
 556         return false;
 557       _dig.update(buf, cnt);
 558       l -= cnt;
 559     }
 560   return _request->_blklist->verifyDigest(_blkno, _dig);
 561 }
 562
 563
 564 void
 565 multifetchworker::stealjob()
 566 {
 567   if (!_request->_stealing)
 568     {
 569       DBG << "start stealing!" << endl;
 570       _request->_stealing = true;
 571     }
 572   multifetchworker *best = 0;
 573   std::list<multifetchworker *>::iterator workeriter = _request->_workers.begin();
 574   double now = 0;
 575   for (; workeriter != _request->_workers.end(); ++workeriter)
 576     {
 577       multifetchworker *worker = *workeriter;
 578       if (worker == this)
 579         continue;
 580       if (worker->_pass == -1)
 581         continue;       // do not steal!
 582       if (worker->_state == WORKER_DISCARD || worker->_state == WORKER_DONE || worker->_state == WORKER_SLEEP || !worker->_blksize)
 583         continue;       // do not steal finished jobs
 584       if (!worker->_avgspeed && worker->_blkreceived)
 585         {
 586           if (!now)
 587             now = currentTime();
 588           if (now > worker->_blkstarttime)
 589             worker->_avgspeed = worker->_blkreceived / (now - worker->_blkstarttime);
 590         }
 591       if (!best || best->_pass > worker->_pass)
 592         {
 593           best = worker;
 594           continue;
 595         }
 596       if (best->_pass < worker->_pass)
 597         continue;
 598       // if it is the same block, we want to know the best worker, otherwise the worst
 599       if (worker->_blkstart == best->_blkstart)
 600         {
 601           if ((worker->_blksize - worker->_blkreceived) * best->_avgspeed < (best->_blksize - best->_blkreceived) * worker->_avgspeed)
 602             best = worker;
 603         }
 604       else
 605         {
 606           if ((worker->_blksize - worker->_blkreceived) * best->_avgspeed > (best->_blksize - best->_blkreceived) * worker->_avgspeed)
 607             best = worker;
 608         }
 609     }
 610   if (!best)
 611     {
 612       _state = WORKER_DONE;
 613       _request->_activeworkers--;
 614       _request->_finished = true;
 615       return;
 616     }
 617   // do not sleep twice
 618   if (_state != WORKER_SLEEP)
 619     {
 620       if (!_avgspeed && _blkreceived)
 621         {
 622           if (!now)
 623             now = currentTime();
 624           if (now > _blkstarttime)
 625             _avgspeed = _blkreceived / (now - _blkstarttime);
 626         }
 627
 628       // lets see if we should sleep a bit
 629       DBG << "me #" << _workerno << ": " << _avgspeed << ", size " << best->_blksize << endl;
 630       DBG << "best #" << best->_workerno << ": " << best->_avgspeed << ", size " << (best->_blksize - best->_blkreceived) << endl;
 631       if (_avgspeed && best->_avgspeed && (best->_blksize - best->_blkreceived) * _avgspeed < best->_blksize * best->_avgspeed)
 632         {
 633           if (!now)
 634             now = currentTime();
 635           double sl = (best->_blksize - best->_blkreceived) / best->_avgspeed * 2;
 636           if (sl > 1)
 637             sl = 1;
 638           DBG << "#" << _workerno << ": going to sleep for " << sl * 1000 << " ms" << endl;
 639           _sleepuntil = now + sl;
 640           _state = WORKER_SLEEP;
 641           _request->_sleepworkers++;
 642           return;
 643         }
 644     }
 645
 646   _competing = true;
 647   best->_competing = true;
 648   _blkstart = best->_blkstart;
 649   _blksize = best->_blksize;
 650   best->_pass++;
 651   _pass = best->_pass;
 652   _blkno = best->_blkno;
 653   run();
 654 }
 655
 656 void
 657 multifetchworker::disableCompetition()
 658 {
 659   std::list<multifetchworker *>::iterator workeriter = _request->_workers.begin();
 660   for (; workeriter != _request->_workers.end(); ++workeriter)
 661     {
 662       multifetchworker *worker = *workeriter;
 663       if (worker == this)
 664         continue;
 665       if (worker->_blkstart == _blkstart)
 666         {
 667           if (worker->_state == WORKER_FETCH)
 668             worker->_state = WORKER_DISCARD;
 669           worker->_pass = -1;   /* do not steal this one, we already have it */
 670         }
 671     }
 672 }
 673
 674
 675 void
 676 multifetchworker::nextjob()
 677 {
 678   _noendrange = false;
 679   if (_request->_stealing)
 680     {
 681       stealjob();
 682       return;
 683     }
 684
 685   MediaBlockList *blklist = _request->_blklist;
 686   if (!blklist)
 687     {
 688       _blksize = BLKSIZE;
 689       if (_request->_filesize != off_t(-1))
 690         {
 691           if (_request->_blkoff >= _request->_filesize)
 692             {
 693               stealjob();
 694               return;
 695             }
 696           _blksize = _request->_filesize - _request->_blkoff;
 697           if (_blksize > BLKSIZE)
 698             _blksize = BLKSIZE;
 699         }
 700     }
 701   else
 702     {
 703       MediaBlock blk = blklist->getBlock(_request->_blkno);
 704       while (_request->_blkoff >= blk.off + blk.size)
 705         {
 706           if (++_request->_blkno == blklist->numBlocks())
 707             {
 708               stealjob();
 709               return;
 710             }
 711           blk = blklist->getBlock(_request->_blkno);
 712           _request->_blkoff = blk.off;
 713         }
 714       _blksize = blk.off + blk.size - _request->_blkoff;
 715       if (_blksize > BLKSIZE && !blklist->haveChecksum(_request->_blkno))
 716         _blksize = BLKSIZE;
 717     }
 718   _blkno = _request->_blkno;
 719   _blkstart = _request->_blkoff;
 720   _request->_blkoff += _blksize;
 721   run();
 722 }
 723
 724 void
 725 multifetchworker::run()
 726 {
 727   char rangebuf[128];
 728
 729   if (_state == WORKER_BROKEN || _state == WORKER_DONE)
 730      return;    // just in case...
 731   if (_noendrange)
 732     sprintf(rangebuf, "%llu-", (unsigned long long)_blkstart);
 733   else
 734     sprintf(rangebuf, "%llu-%llu", (unsigned long long)_blkstart, (unsigned long long)_blkstart + _blksize - 1);
 735   DBG << "#" << _workerno << ": BLK " << _blkno << ":" << rangebuf << " " << _url << endl;
 736   if (curl_easy_setopt(_curl, CURLOPT_RANGE, !_noendrange || _blkstart != 0 ? rangebuf : (char *)0) != CURLE_OK)
 737     {
 738       _request->_activeworkers--;
 739       _state = WORKER_BROKEN;
 740       strncpy(_curlError, "curl_easy_setopt range failed", CURL_ERROR_SIZE);
 741       return;
 742     }
 743   if (curl_multi_add_handle(_request->_multi, _curl) != CURLM_OK)
 744     {
 745       _request->_activeworkers--;
 746       _state = WORKER_BROKEN;
 747       strncpy(_curlError, "curl_multi_add_handle failed", CURL_ERROR_SIZE);
 748       return;
 749     }
 750   _request->_havenewjob = true;
 751   _off = _blkstart;
 752   _size = _blksize;
 753   if (_request->_blklist)
 754     _request->_blklist->createDigest(_dig);     // resets digest
 755   _state = WORKER_FETCH;
 756
 757   double now = currentTime();
 758   _blkstarttime = now;
 759   _blkreceived = 0;
 760 }
 761
 762
 763 //////////////////////////////////////////////////////////////////////
 764
 765
 766 multifetchrequest::multifetchrequest(const MediaMultiCurl *context, const Pathname &filename, const Url &baseurl, CURLM *multi, FILE *fp, callback::SendReport<DownloadProgressReport> *report, MediaBlockList *blklist, off_t filesize) : _context(context), _filename(filename), _baseurl(baseurl)
 767 {
 768   _fp = fp;
 769   _report = report;
 770   _blklist = blklist;
 771   _filesize = filesize;
 772   _multi = multi;
 773   _stealing = false;
 774   _havenewjob = false;
 775   _blkno = 0;
 776   if (_blklist)
 777     _blkoff = _blklist->getBlock(0).off;
 778   else
 779     _blkoff = 0;
 780   _activeworkers = 0;
 781   _lookupworkers = 0;
 782   _sleepworkers = 0;
 783   _minsleepuntil = 0;
 784   _finished = false;
 785   _fetchedsize = 0;
 786   _fetchedgoodsize = 0;
 787   _totalsize = 0;
 788   _lastperiodstart = _lastprogress = _starttime = currentTime();
 789   _lastperiodfetched = 0;
 790   _periodavg = 0;
 791   _timeout = 0;
 792   _connect_timeout = 0;
 793   _maxspeed = 0;
 794   if (blklist)
 795     {
 796       for (size_t blkno = 0; blkno < blklist->numBlocks(); blkno++)
 797         {
 798           MediaBlock blk = blklist->getBlock(blkno);
 799           _totalsize += blk.size;
 800         }
 801     }
 802   else if (filesize != off_t(-1))
 803     _totalsize = filesize;
 804 }
 805
 806 multifetchrequest::~multifetchrequest()
 807 {
 808   for (std::list<multifetchworker *>::iterator workeriter = _workers.begin(); workeriter != _workers.end(); ++workeriter)
 809     {
 810       multifetchworker *worker = *workeriter;
 811       *workeriter = NULL;
 812       delete worker;
 813     }
 814   _workers.clear();
 815 }
 816
 817 void
 818 multifetchrequest::run(std::vector<Url> &urllist)
 819 {
 820   int workerno = 0;
 821   std::vector<Url>::iterator urliter = urllist.begin();
 822   for (;;)
 823     {
 824       fd_set rset, wset, xset;
 825       int maxfd, nqueue;
 826
 827       if (_finished)
 828         {
 829           DBG << "finished!" << endl;
 830           break;
 831         }
 832
 833       if (_activeworkers < MAXWORKERS && urliter != urllist.end() && _workers.size() < MAXURLS)
 834         {
 835           // spawn another worker!
 836           multifetchworker *worker = new multifetchworker(workerno++, *this, *urliter);
 837           _workers.push_back(worker);
 838           if (worker->_state != WORKER_BROKEN)
 839             {
 840               _activeworkers++;
 841               if (worker->_state != WORKER_LOOKUP)
 842                 {
 843                   worker->nextjob();
 844                 }
 845               else
 846                 _lookupworkers++;
 847             }
 848           ++urliter;
 849           continue;
 850         }
 851       if (!_activeworkers)
 852         {
 853           WAR << "No more active workers!" << endl;
 854           // show the first worker error we find
 855           for (std::list<multifetchworker *>::iterator workeriter = _workers.begin(); workeriter != _workers.end(); ++workeriter)
 856             {
 857               if ((*workeriter)->_state != WORKER_BROKEN)
 858                 continue;
 859               ZYPP_THROW(MediaCurlException(_baseurl, "Server error", (*workeriter)->_curlError));
 860             }
 861           break;
 862         }
 863
 864       FD_ZERO(&rset);
 865       FD_ZERO(&wset);
 866       FD_ZERO(&xset);
 867
 868       curl_multi_fdset(_multi, &rset, &wset, &xset, &maxfd);
 869
 870       if (_lookupworkers)
 871         for (std::list<multifetchworker *>::iterator workeriter = _workers.begin(); workeriter != _workers.end(); ++workeriter)
 872           (*workeriter)->adddnsfd(rset, maxfd);
 873
 874       timeval tv;
 875       // if we added a new job we have to call multi_perform once
 876       // to make it show up in the fd set. do not sleep in this case.
 877       tv.tv_sec = 0;
 878       tv.tv_usec = _havenewjob ? 0 : 200000;
 879       if (_sleepworkers && !_havenewjob)
 880         {
 881           if (_minsleepuntil == 0)
 882             {
 883               for (std::list<multifetchworker *>::iterator workeriter = _workers.begin(); workeriter != _workers.end(); ++workeriter)
 884                 {
 885                   multifetchworker *worker = *workeriter;
 886                   if (worker->_state != WORKER_SLEEP)
 887                     continue;
 888                   if (!_minsleepuntil || _minsleepuntil > worker->_sleepuntil)
 889                     _minsleepuntil = worker->_sleepuntil;
 890                 }
 891             }
 892           double sl = _minsleepuntil - currentTime();
 893           if (sl < 0)
 894             {
 895               sl = 0;
 896               _minsleepuntil = 0;
 897             }
 898           if (sl < .2)
 899             tv.tv_usec = sl * 1000000;
 900         }
 901       int r = select(maxfd + 1, &rset, &wset, &xset, &tv);
 902       if (r == -1 && errno != EINTR)
 903         ZYPP_THROW(MediaCurlException(_baseurl, "select() failed", "unknown error"));
 904       if (r != 0 && _lookupworkers)
 905         for (std::list<multifetchworker *>::iterator workeriter = _workers.begin(); workeriter != _workers.end(); ++workeriter)
 906           {
 907             multifetchworker *worker = *workeriter;
 908             if (worker->_state != WORKER_LOOKUP)
 909               continue;
 910             (*workeriter)->dnsevent(rset);
 911             if (worker->_state != WORKER_LOOKUP)
 912               _lookupworkers--;
 913           }
 914       _havenewjob = false;
 915
 916       // run curl
 917       for (;;)
 918         {
 919           CURLMcode mcode;
 920           int tasks;
 921           mcode = curl_multi_perform(_multi, &tasks);
 922           if (mcode == CURLM_CALL_MULTI_PERFORM)
 923             continue;
 924           if (mcode != CURLM_OK)
 925             ZYPP_THROW(MediaCurlException(_baseurl, "curl_multi_perform", "unknown error"));
 926           break;
 927         }
 928
 929       double now = currentTime();
 930
 931       // update periodavg
 932       if (now > _lastperiodstart + .5)
 933         {
 934           if (!_periodavg)
 935             _periodavg = (_fetchedsize - _lastperiodfetched) / (now - _lastperiodstart);
 936           else
 937             _periodavg = (_periodavg + (_fetchedsize - _lastperiodfetched) / (now - _lastperiodstart)) / 2;
 938           _lastperiodfetched = _fetchedsize;
 939           _lastperiodstart = now;
 940         }
 941
 942       // wake up sleepers
 943       if (_sleepworkers)
 944         {
 945           for (std::list<multifetchworker *>::iterator workeriter = _workers.begin(); workeriter != _workers.end(); ++workeriter)
 946             {
 947               multifetchworker *worker = *workeriter;
 948               if (worker->_state != WORKER_SLEEP)
 949                 continue;
 950               if (worker->_sleepuntil > now)
 951                 continue;
 952               if (_minsleepuntil == worker->_sleepuntil)
 953                 _minsleepuntil = 0;
 954               DBG << "#" << worker->_workerno << ": sleep done, wake up" << endl;
 955               _sleepworkers--;
 956               // nextjob chnages the state
 957               worker->nextjob();
 958             }
 959         }
 960
 961       // collect all curl results, reschedule new jobs
 962       CURLMsg *msg;
 963       while ((msg = curl_multi_info_read(_multi, &nqueue)) != 0)
 964         {
 965           if (msg->msg != CURLMSG_DONE)
 966             continue;
 967           multifetchworker *worker;
 968           if (curl_easy_getinfo(msg->easy_handle, CURLINFO_PRIVATE, &worker) != CURLE_OK)
 969             ZYPP_THROW(MediaCurlException(_baseurl, "curl_easy_getinfo", "unknown error"));
 970           CURLcode cc = msg->data.result;
 971           if (worker->_blkreceived && now > worker->_blkstarttime)
 972             {
 973               if (worker->_avgspeed)
 974                 worker->_avgspeed = (worker->_avgspeed + worker->_blkreceived / (now - worker->_blkstarttime)) / 2;
 975               else
 976                 worker->_avgspeed = worker->_blkreceived / (now - worker->_blkstarttime);
 977             }
 978           DBG << "#" << worker->_workerno << ": BLK " << worker->_blkno << " done code " << cc << " speed " << worker->_avgspeed << endl;
 979           curl_multi_remove_handle(_multi, msg->easy_handle);
 980           if (cc == CURLE_HTTP_RETURNED_ERROR)
 981             {
 982               long statuscode = 0;
 983               (void)curl_easy_getinfo(msg->easy_handle, CURLINFO_RESPONSE_CODE, &statuscode);
 984               DBG << "HTTP status " << statuscode << endl;
 985               if (statuscode == 416 && !_blklist)       /* Range error */
 986                 {
 987                   if (_filesize == off_t(-1))
 988                     {
 989                       if (!worker->_noendrange)
 990                         {
 991                           DBG << "#" << worker->_workerno << ": retrying with no end range" << endl;
 992                           worker->_noendrange = true;
 993                           worker->run();
 994                           continue;
 995                         }
 996                       worker->_noendrange = false;
 997                       worker->stealjob();
 998                       continue;
 999                     }
1000                   if (worker->_blkstart >= _filesize)
1001                     {
1002                       worker->nextjob();
1003                       continue;
1004                     }
1005                 }
1006             }
1007           if (cc == 0)
1008             {
1009               if (!worker->checkChecksum())
1010                 {
1011                   WAR << "#" << worker->_workerno << ": checksum error, disable worker" << endl;
1012                   worker->_state = WORKER_BROKEN;
1013                   strncpy(worker->_curlError, "checksum error", CURL_ERROR_SIZE);
1014                   _activeworkers--;
1015                   continue;
1016                 }
1017               if (worker->_state == WORKER_FETCH)
1018                 {
1019                   if (worker->_competing)
1020                     {
1021                       worker->disableCompetition();
1022                       // multiple workers wrote into this block. We already know that our
1023                       // data was correct, but maybe some other worker overwrote our data
1024                       // with something broken. Thus we have to re-check the block.
1025                       if (!worker->recheckChecksum())
1026                         {
1027                           DBG << "#" << worker->_workerno << ": recheck checksum error, refetch block" << endl;
1028                           // re-fetch! No need to worry about the bad workers,
1029                           // they will now be set to DISCARD. At the end of their block
1030                           // they will notice that they wrote bad data and go into BROKEN.
1031                           worker->run();
1032                           continue;
1033                         }
1034                     }
1035                   _fetchedgoodsize += worker->_blksize;
1036                 }
1037
1038               // make bad workers sleep a little
1039               double maxavg = 0;
1040               int maxworkerno = 0;
1041               int numbetter = 0;
1042               for (std::list<multifetchworker *>::iterator workeriter = _workers.begin(); workeriter != _workers.end(); ++workeriter)
1043                 {
1044                   multifetchworker *oworker = *workeriter;
1045                   if (oworker->_state == WORKER_BROKEN)
1046                     continue;
1047                   if (oworker->_avgspeed > maxavg)
1048                     {
1049                       maxavg = oworker->_avgspeed;
1050                       maxworkerno = oworker->_workerno;
1051                     }
1052                   if (oworker->_avgspeed > worker->_avgspeed)
1053                     numbetter++;
1054                 }
1055               if (maxavg && !_stealing)
1056                 {
1057                   double ratio = worker->_avgspeed / maxavg;
1058                   ratio = 1 - ratio;
1059                   if (numbetter < 3)    // don't sleep that much if we're in the top two
1060                     ratio = ratio * ratio;
1061                   if (ratio > .01)
1062                     {
1063                       DBG << "#" << worker->_workerno << ": too slow ("<< ratio << ", " << worker->_avgspeed << ", #" << maxworkerno << ": " << maxavg << "), going to sleep for " << ratio * 1000 << " ms" << endl;
1064                       worker->_sleepuntil = now + ratio;
1065                       worker->_state = WORKER_SLEEP;
1066                       _sleepworkers++;
1067                       continue;
1068                     }
1069                 }
1070
1071               // do rate control (if requested)
1072               // should use periodavg, but that's not what libcurl does
1073               if (_maxspeed && now > _starttime)
1074                 {
1075                   double avg = _fetchedsize / (now - _starttime);
1076                   avg = worker->_maxspeed * _maxspeed / avg;
1077                   if (avg < _maxspeed / MAXWORKERS)
1078                     avg = _maxspeed / MAXWORKERS;
1079                   if (avg > _maxspeed)
1080                     avg = _maxspeed;
1081                   if (avg < 1024)
1082                     avg = 1024;
1083                   worker->_maxspeed = avg;
1084                   curl_easy_setopt(worker->_curl, CURLOPT_MAX_RECV_SPEED_LARGE, (curl_off_t)(avg));
1085                 }
1086
1087               worker->nextjob();
1088             }
1089           else
1090             {
1091               worker->_state = WORKER_BROKEN;
1092               _activeworkers--;
1093               if (!_activeworkers && !(urliter != urllist.end() && _workers.size() < MAXURLS))
1094                 {
1095                   // end of workers reached! goodbye!
1096                   worker->evaluateCurlCode(Pathname(), cc, false);
1097                 }
1098             }
1099         }
1100
1101       // send report
1102       if (_report)
1103         {
1104           int percent = _totalsize ? (100 * (_fetchedgoodsize + _fetchedsize)) / (_totalsize + _fetchedsize) : 0;
1105           double avg = 0;
1106           if (now > _starttime)
1107             avg = _fetchedsize / (now - _starttime);
1108           if (!(*(_report))->progress(percent, _baseurl, avg, _lastperiodstart == _starttime ? avg : _periodavg))
1109             ZYPP_THROW(MediaCurlException(_baseurl, "User abort", "cancelled"));
1110         }
1111
1112       if (_timeout && now - _lastprogress > _timeout)
1113         break;
1114     }
1115
1116   if (!_finished)
1117     ZYPP_THROW(MediaTimeoutException(_baseurl));
1118
1119   // print some download stats
1120   WAR << "overall result" << endl;
1121   for (std::list<multifetchworker *>::iterator workeriter = _workers.begin(); workeriter != _workers.end(); ++workeriter)
1122     {
1123       multifetchworker *worker = *workeriter;
1124       WAR << "#" << worker->_workerno << ": state: " << worker->_state << " received: " << worker->_received << " url: " << worker->_url << endl;
1125     }
1126 }
1127
1128
1129 //////////////////////////////////////////////////////////////////////
1130
1131
1132 MediaMultiCurl::MediaMultiCurl(const Url &url_r, const Pathname & attach_point_hint_r)
1133     : MediaCurl(url_r, attach_point_hint_r)
1134 {
1135   MIL << "MediaMultiCurl::MediaMultiCurl(" << url_r << ", " << attach_point_hint_r << ")" << endl;
1136   _multi = 0;
1137   _customHeadersMetalink = 0;
1138 }
1139
1140 MediaMultiCurl::~MediaMultiCurl()
1141 {
1142   if (_customHeadersMetalink)
1143     {
1144       curl_slist_free_all(_customHeadersMetalink);
1145       _customHeadersMetalink = 0;
1146     }
1147   if (_multi)
1148     {
1149       curl_multi_cleanup(_multi);
1150       _multi = 0;
1151     }
1152   std::map<std::string, CURL *>::iterator it;
1153   for (it = _easypool.begin(); it != _easypool.end(); it++)
1154     {
1155       CURL *easy = it->second;
1156       if (easy)
1157         {
1158           curl_easy_cleanup(easy);
1159           it->second = NULL;
1160         }
1161     }
1162 }
1163
1164 void MediaMultiCurl::setupEasy()
1165 {
1166   MediaCurl::setupEasy();
1167
1168   if (_customHeadersMetalink)
1169     {
1170       curl_slist_free_all(_customHeadersMetalink);
1171       _customHeadersMetalink = 0;
1172     }
1173   struct curl_slist *sl = _customHeaders;
1174   for (; sl; sl = sl->next)
1175     _customHeadersMetalink = curl_slist_append(_customHeadersMetalink, sl->data);
1176   _customHeadersMetalink = curl_slist_append(_customHeadersMetalink, "Accept: */*, application/metalink+xml, application/metalink4+xml");
1177 }
1178
1179
1180 void MediaMultiCurl::doGetFileCopy( const Pathname & filename , const Pathname & target, callback::SendReport<DownloadProgressReport> & report, RequestOptions options ) const
1181 {
1182   Pathname dest = target.absolutename();
1183   if( assert_dir( dest.dirname() ) )
1184   {
1185     DBG << "assert_dir " << dest.dirname() << " failed" << endl;
1186     Url url(getFileUrl(filename));
1187     ZYPP_THROW( MediaSystemException(url, "System error on " + dest.dirname().asString()) );
1188   }
1189   string destNew = target.asString() + ".new.zypp.XXXXXX";
1190   char *buf = ::strdup( destNew.c_str());
1191   if( !buf)
1192   {
1193     ERR << "out of memory for temp file name" << endl;
1194     Url url(getFileUrl(filename));
1195     ZYPP_THROW(MediaSystemException(url, "out of memory for temp file name"));
1196   }
1197
1198   int tmp_fd = ::mkstemp( buf );
1199   if( tmp_fd == -1)
1200   {
1201     free( buf);
1202     ERR << "mkstemp failed for file '" << destNew << "'" << endl;
1203     ZYPP_THROW(MediaWriteException(destNew));
1204   }
1205   destNew = buf;
1206   free( buf);
1207
1208   FILE *file = ::fdopen( tmp_fd, "w" );
1209   if ( !file ) {
1210     ::close( tmp_fd);
1211     filesystem::unlink( destNew );
1212     ERR << "fopen failed for file '" << destNew << "'" << endl;
1213     ZYPP_THROW(MediaWriteException(destNew));
1214   }
1215   DBG << "dest: " << dest << endl;
1216   DBG << "temp: " << destNew << endl;
1217
1218   // set IFMODSINCE time condition (no download if not modified)
1219   if( PathInfo(target).isExist() && !(options & OPTION_NO_IFMODSINCE) )
1220   {
1221     curl_easy_setopt(_curl, CURLOPT_TIMECONDITION, CURL_TIMECOND_IFMODSINCE);
1222     curl_easy_setopt(_curl, CURLOPT_TIMEVALUE, (long)PathInfo(target).mtime());
1223   }
1224   else
1225   {
1226     curl_easy_setopt(_curl, CURLOPT_TIMECONDITION, CURL_TIMECOND_NONE);
1227     curl_easy_setopt(_curl, CURLOPT_TIMEVALUE, 0L);
1228   }
1229   // change header to include Accept: metalink
1230   curl_easy_setopt(_curl, CURLOPT_HTTPHEADER, _customHeadersMetalink);
1231   try
1232     {
1233       MediaCurl::doGetFileCopyFile(filename, dest, file, report, options);
1234     }
1235   catch (Exception &ex)
1236     {
1237       ::fclose(file);
1238       filesystem::unlink(destNew);
1239       curl_easy_setopt(_curl, CURLOPT_TIMECONDITION, CURL_TIMECOND_NONE);
1240       curl_easy_setopt(_curl, CURLOPT_TIMEVALUE, 0L);
1241       curl_easy_setopt(_curl, CURLOPT_HTTPHEADER, _customHeaders);
1242       ZYPP_RETHROW(ex);
1243     }
1244   curl_easy_setopt(_curl, CURLOPT_TIMECONDITION, CURL_TIMECOND_NONE);
1245   curl_easy_setopt(_curl, CURLOPT_TIMEVALUE, 0L);
1246   curl_easy_setopt(_curl, CURLOPT_HTTPHEADER, _customHeaders);
1247   long httpReturnCode = 0;
1248   CURLcode infoRet = curl_easy_getinfo(_curl, CURLINFO_RESPONSE_CODE, &httpReturnCode);
1249   if (infoRet == CURLE_OK)
1250   {
1251     DBG << "HTTP response: " + str::numstring(httpReturnCode) << endl;
1252     if ( httpReturnCode == 304
1253          || ( httpReturnCode == 213 && _url.getScheme() == "ftp" ) ) // not modified
1254     {
1255       DBG << "not modified: " << PathInfo(dest) << endl;
1256       return;
1257     }
1258   }
1259   else
1260   {
1261     WAR << "Could not get the reponse code." << endl;
1262   }
1263   char *ptr = NULL;
1264   if (curl_easy_getinfo(_curl, CURLINFO_CONTENT_TYPE, &ptr) == CURLE_OK && ptr)
1265     {
1266       string ct = string(ptr);
1267       if (ct.find("application/metalink+xml") == 0 || ct.find("application/metalink4+xml") == 0)
1268         {
1269           bool userabort = false;
1270           fclose(file);
1271           file = NULL;
1272           try
1273             {
1274               MetaLinkParser mlp;
1275               mlp.parse(destNew);
1276               MediaBlockList bl = mlp.getBlockList();
1277               vector<Url> urls = mlp.getUrls();
1278               DBG << bl << endl;
1279               file = fopen(destNew.c_str(), "w+");
1280               if (!file)
1281                 ZYPP_THROW(MediaWriteException(destNew));
1282               if (PathInfo(target).isExist())
1283                 {
1284                   DBG << "reusing blocks from file " << target << endl;
1285                   bl.reuseBlocks(file, target.asString());
1286                   DBG << bl << endl;
1287                 }
1288               Pathname df = deltafile();
1289               if (!df.empty())
1290                 {
1291                   DBG << "reusing blocks from file " << df << endl;
1292                   bl.reuseBlocks(file, df.asString());
1293                   DBG << bl << endl;
1294                 }
1295               try
1296                 {
1297                   multifetch(filename, file, &urls, &report, &bl);
1298                 }
1299               catch (MediaCurlException &ex)
1300                 {
1301                   fclose(file);
1302                   userabort = ex.errstr() == "User abort";
1303                   ZYPP_RETHROW(ex);
1304                 }
1305             }
1306           catch (Exception &ex)
1307             {
1308               // something went wrong. fall back to normal download
1309               if (file)
1310                 fclose(file);
1311               filesystem::unlink(destNew);
1312               if (userabort)
1313                 ZYPP_RETHROW(ex);
1314               MediaCurl::doGetFileCopyFile(filename, dest, file, report, options | OPTION_NO_REPORT_START);
1315             }
1316         }
1317     }
1318   if (::fchmod( ::fileno(file), filesystem::applyUmaskTo( 0644 )))
1319     {
1320       ERR << "Failed to chmod file " << destNew << endl;
1321     }
1322   if (::fclose(file))
1323     {
1324       filesystem::unlink(destNew);
1325       ERR << "Fclose failed for file '" << destNew << "'" << endl;
1326       ZYPP_THROW(MediaWriteException(destNew));
1327     }
1328   if ( rename( destNew, dest ) != 0 )
1329     {
1330       ERR << "Rename failed" << endl;
1331       ZYPP_THROW(MediaWriteException(dest));
1332     }
1333   DBG << "done: " << PathInfo(dest) << endl;
1334 }
1335
1336 void MediaMultiCurl::multifetch(const Pathname & filename, FILE *fp, std::vector<Url> *urllist, callback::SendReport<DownloadProgressReport> *report, MediaBlockList *blklist, off_t filesize) const
1337 {
1338   Url baseurl(getFileUrl(filename));
1339   if (blklist && filesize == off_t(-1) && blklist->haveFilesize())
1340     filesize = blklist->getFilesize();
1341   if (blklist && !blklist->haveBlocks() && filesize != 0)
1342     blklist = 0;
1343   if (blklist && (filesize == 0 || !blklist->numBlocks()))
1344     {
1345       checkFileDigest(baseurl, fp, blklist);
1346       return;
1347     }
1348   if (filesize == 0)
1349     return;
1350   if (!_multi)
1351     {
1352       _multi = curl_multi_init();
1353       if (!_multi)
1354         ZYPP_THROW(MediaCurlInitException(baseurl));
1355     }
1356   multifetchrequest req(this, filename, baseurl, _multi, fp, report, blklist, filesize);
1357   req._timeout = _settings.timeout();
1358   req._connect_timeout = _settings.connectTimeout();
1359   req._maxspeed = _settings.maxDownloadSpeed();
1360   std::vector<Url> myurllist;
1361   for (std::vector<Url>::iterator urliter = urllist->begin(); urliter != urllist->end(); ++urliter)
1362     {
1363       try
1364         {
1365           string scheme = urliter->getScheme();
1366           if (scheme == "http" || scheme == "https" || scheme == "ftp")
1367             {
1368               checkProtocol(*urliter);
1369               myurllist.push_back(*urliter);
1370             }
1371         }
1372       catch (...)
1373         {
1374         }
1375     }
1376   if (!myurllist.size())
1377     myurllist.push_back(baseurl);
1378   req.run(myurllist);
1379   checkFileDigest(baseurl, fp, blklist);
1380 }
1381
1382 void MediaMultiCurl::checkFileDigest(Url &url, FILE *fp, MediaBlockList *blklist) const
1383 {
1384   if (!blklist || !blklist->haveFileChecksum())
1385     return;
1386   if (fseeko(fp, off_t(0), SEEK_SET))
1387     ZYPP_THROW(MediaCurlException(url, "fseeko", "seek error"));
1388   Digest dig;
1389   blklist->createFileDigest(dig);
1390   char buf[4096];
1391   size_t l;
1392   while ((l = fread(buf, 1, sizeof(buf), fp)) > 0)
1393     dig.update(buf, l);
1394   if (!blklist->verifyFileDigest(dig))
1395     ZYPP_THROW(MediaCurlException(url, "file verification failed", "checksum error"));
1396 }
1397
1398 bool MediaMultiCurl::isDNSok(const string &host) const
1399 {
1400   return _dnsok.find(host) == _dnsok.end() ? false : true;
1401 }
1402
1403 void MediaMultiCurl::setDNSok(const string &host) const
1404 {
1405   _dnsok.insert(host);
1406 }
1407
1408 CURL *MediaMultiCurl::fromEasyPool(const string &host) const
1409 {
1410   if (_easypool.find(host) == _easypool.end())
1411     return 0;
1412   CURL *ret = _easypool[host];
1413   _easypool.erase(host);
1414   return ret;
1415 }
1416
1417 void MediaMultiCurl::toEasyPool(const std::string &host, CURL *easy) const
1418 {
1419   CURL *oldeasy = _easypool[host];
1420   _easypool[host] = easy;
1421   if (oldeasy)
1422     curl_easy_cleanup(oldeasy);
1423 }
1424
1425   } // namespace media
1426 } // namespace zypp
1427