apt-pkg/acquire.cc

   1 // -*- mode: cpp; mode: fold -*-
   2 // Description                                                          /*{{{*/
   3 // $Id: acquire.cc,v 1.50 2004/03/17 05:17:11 mdz Exp $
   4 /* ######################################################################
   5
   6    Acquire - File Acquiration
   7
   8    The core element for the schedule system is the concept of a named
   9    queue. Each queue is unique and each queue has a name derived from the
  10    URI. The degree of paralization can be controlled by how the queue
  11    name is derived from the URI.
  12
  13    ##################################################################### */
  14                                                                         /*}}}*/
  15 // Include Files                                                        /*{{{*/
  16 #include <config.h>
  17
  18 #include <apt-pkg/acquire.h>
  19 #include <apt-pkg/acquire-item.h>
  20 #include <apt-pkg/acquire-worker.h>
  21 #include <apt-pkg/configuration.h>
  22 #include <apt-pkg/error.h>
  23 #include <apt-pkg/strutl.h>
  24 #include <apt-pkg/fileutl.h>
  25
  26 #include <algorithm>
  27 #include <string>
  28 #include <vector>
  29 #include <iostream>
  30 #include <sstream>
  31 #include <iomanip>
  32
  33 #include <stdio.h>
  34 #include <stdlib.h>
  35 #include <string.h>
  36 #include <unistd.h>
  37 #include <pwd.h>
  38 #include <grp.h>
  39 #include <dirent.h>
  40 #include <sys/time.h>
  41 #include <sys/select.h>
  42 #include <errno.h>
  43 #include <sys/stat.h>
  44
  45 #include <apti18n.h>
  46                                                                         /*}}}*/
  47
  48 using namespace std;
  49
  50 // Acquire::pkgAcquire - Constructor                                    /*{{{*/
  51 // ---------------------------------------------------------------------
  52 /* We grab some runtime state from the configuration space */
  53 pkgAcquire::pkgAcquire() : LockFD(-1), d(NULL), Queues(0), Workers(0), Configs(0), Log(NULL), ToFetch(0),
  54                            Debug(_config->FindB("Debug::pkgAcquire",false)),
  55                            Running(false)
  56 {
  57    Initialize();
  58 }
  59 pkgAcquire::pkgAcquire(pkgAcquireStatus *Progress) : LockFD(-1), d(NULL), Queues(0), Workers(0),
  60                            Configs(0), Log(NULL), ToFetch(0),
  61                            Debug(_config->FindB("Debug::pkgAcquire",false)),
  62                            Running(false)
  63 {
  64    Initialize();
  65    SetLog(Progress);
  66 }
  67 void pkgAcquire::Initialize()
  68 {
  69    string const Mode = _config->Find("Acquire::Queue-Mode","host");
  70    if (strcasecmp(Mode.c_str(),"host") == 0)
  71       QueueMode = QueueHost;
  72    if (strcasecmp(Mode.c_str(),"access") == 0)
  73       QueueMode = QueueAccess;
  74
  75    // chown the auth.conf file as it will be accessed by our methods
  76    std::string const SandboxUser = _config->Find("APT::Sandbox::User");
  77    if (getuid() == 0 && SandboxUser.empty() == false) // if we aren't root, we can't chown, so don't try it
  78    {
  79       struct passwd const * const pw = getpwnam(SandboxUser.c_str());
  80       struct group const * const gr = getgrnam("root");
  81       if (pw != NULL && gr != NULL)
  82       {
  83          std::string const AuthConf = _config->FindFile("Dir::Etc::netrc");
  84          if(AuthConf.empty() == false && RealFileExists(AuthConf) &&
  85                chown(AuthConf.c_str(), pw->pw_uid, gr->gr_gid) != 0)
  86             _error->WarningE("SetupAPTPartialDirectory", "chown to %s:root of file %s failed", SandboxUser.c_str(), AuthConf.c_str());
  87       }
  88    }
  89 }
  90                                                                         /*}}}*/
  91 // Acquire::GetLock - lock directory and prepare for action             /*{{{*/
  92 static bool SetupAPTPartialDirectory(std::string const &grand, std::string const &parent)
  93 {
  94    std::string const partial = parent + "partial";
  95    mode_t const mode = umask(S_IWGRP | S_IWOTH);
  96    bool const creation_fail = (CreateAPTDirectoryIfNeeded(grand, partial) == false &&
  97          CreateAPTDirectoryIfNeeded(parent, partial) == false);
  98    umask(mode);
  99    if (creation_fail == true)
 100       return false;
 101
 102    std::string const SandboxUser = _config->Find("APT::Sandbox::User");
 103    if (getuid() == 0 && SandboxUser.empty() == false) // if we aren't root, we can't chown, so don't try it
 104    {
 105       struct passwd const * const pw = getpwnam(SandboxUser.c_str());
 106       struct group const * const gr = getgrnam("root");
 107       if (pw != NULL && gr != NULL)
 108       {
 109          // chown the partial dir
 110          if(chown(partial.c_str(), pw->pw_uid, gr->gr_gid) != 0)
 111             _error->WarningE("SetupAPTPartialDirectory", "chown to %s:root of directory %s failed", SandboxUser.c_str(), partial.c_str());
 112       }
 113    }
 114    if (chmod(partial.c_str(), 0700) != 0)
 115       _error->WarningE("SetupAPTPartialDirectory", "chmod 0700 of directory %s failed", partial.c_str());
 116
 117    return true;
 118 }
 119 bool pkgAcquire::Setup(pkgAcquireStatus *Progress, string const &Lock)
 120 {
 121    Log = Progress;
 122    if (Lock.empty())
 123    {
 124       string const listDir = _config->FindDir("Dir::State::lists");
 125       if (SetupAPTPartialDirectory(_config->FindDir("Dir::State"), listDir) == false)
 126          return _error->Errno("Acquire", _("List directory %spartial is missing."), listDir.c_str());
 127       string const archivesDir = _config->FindDir("Dir::Cache::Archives");
 128       if (SetupAPTPartialDirectory(_config->FindDir("Dir::Cache"), archivesDir) == false)
 129          return _error->Errno("Acquire", _("Archives directory %spartial is missing."), archivesDir.c_str());
 130       return true;
 131    }
 132    return GetLock(Lock);
 133 }
 134 bool pkgAcquire::GetLock(std::string const &Lock)
 135 {
 136    if (Lock.empty() == true)
 137       return false;
 138
 139    // check for existence and possibly create auxiliary directories
 140    string const listDir = _config->FindDir("Dir::State::lists");
 141    string const archivesDir = _config->FindDir("Dir::Cache::Archives");
 142
 143    if (Lock == listDir)
 144    {
 145       if (SetupAPTPartialDirectory(_config->FindDir("Dir::State"), listDir) == false)
 146          return _error->Errno("Acquire", _("List directory %spartial is missing."), listDir.c_str());
 147    }
 148    if (Lock == archivesDir)
 149    {
 150       if (SetupAPTPartialDirectory(_config->FindDir("Dir::Cache"), archivesDir) == false)
 151          return _error->Errno("Acquire", _("Archives directory %spartial is missing."), archivesDir.c_str());
 152    }
 153
 154    if (_config->FindB("Debug::NoLocking", false) == true)
 155       return true;
 156
 157    // Lock the directory this acquire object will work in
 158    if (LockFD != -1)
 159       close(LockFD);
 160    LockFD = ::GetLock(flCombine(Lock, "lock"));
 161    if (LockFD == -1)
 162       return _error->Error(_("Unable to lock directory %s"), Lock.c_str());
 163
 164    return true;
 165 }
 166                                                                         /*}}}*/
 167 // Acquire::~pkgAcquire - Destructor                                    /*{{{*/
 168 // ---------------------------------------------------------------------
 169 /* Free our memory, clean up the queues (destroy the workers) */
 170 pkgAcquire::~pkgAcquire()
 171 {
 172    Shutdown();
 173
 174    if (LockFD != -1)
 175       close(LockFD);
 176
 177    while (Configs != 0)
 178    {
 179       MethodConfig *Jnk = Configs;
 180       Configs = Configs->Next;
 181       delete Jnk;
 182    }
 183 }
 184                                                                         /*}}}*/
 185 // Acquire::Shutdown - Clean out the acquire object                     /*{{{*/
 186 // ---------------------------------------------------------------------
 187 /* */
 188 void pkgAcquire::Shutdown()
 189 {
 190    while (Items.empty() == false)
 191    {
 192       if (Items[0]->Status == Item::StatFetching)
 193          Items[0]->Status = Item::StatError;
 194       delete Items[0];
 195    }
 196
 197    while (Queues != 0)
 198    {
 199       Queue *Jnk = Queues;
 200       Queues = Queues->Next;
 201       delete Jnk;
 202    }
 203 }
 204                                                                         /*}}}*/
 205 // Acquire::Add - Add a new item                                        /*{{{*/
 206 // ---------------------------------------------------------------------
 207 /* This puts an item on the acquire list. This list is mainly for tracking
 208    item status */
 209 void pkgAcquire::Add(Item *Itm)
 210 {
 211    Items.push_back(Itm);
 212 }
 213                                                                         /*}}}*/
 214 // Acquire::Remove - Remove a item                                      /*{{{*/
 215 // ---------------------------------------------------------------------
 216 /* Remove an item from the acquire list. This is usually not used.. */
 217 void pkgAcquire::Remove(Item *Itm)
 218 {
 219    Dequeue(Itm);
 220
 221    for (ItemIterator I = Items.begin(); I != Items.end();)
 222    {
 223       if (*I == Itm)
 224       {
 225          Items.erase(I);
 226          I = Items.begin();
 227       }
 228       else
 229          ++I;
 230    }
 231 }
 232                                                                         /*}}}*/
 233 // Acquire::Add - Add a worker                                          /*{{{*/
 234 // ---------------------------------------------------------------------
 235 /* A list of workers is kept so that the select loop can direct their FD
 236    usage. */
 237 void pkgAcquire::Add(Worker *Work)
 238 {
 239    Work->NextAcquire = Workers;
 240    Workers = Work;
 241 }
 242                                                                         /*}}}*/
 243 // Acquire::Remove - Remove a worker                                    /*{{{*/
 244 // ---------------------------------------------------------------------
 245 /* A worker has died. This can not be done while the select loop is running
 246    as it would require that RunFds could handling a changing list state and
 247    it can't.. */
 248 void pkgAcquire::Remove(Worker *Work)
 249 {
 250    if (Running == true)
 251       abort();
 252
 253    Worker **I = &Workers;
 254    for (; *I != 0;)
 255    {
 256       if (*I == Work)
 257          *I = (*I)->NextAcquire;
 258       else
 259          I = &(*I)->NextAcquire;
 260    }
 261 }
 262                                                                         /*}}}*/
 263 // Acquire::Enqueue - Queue an URI for fetching                         /*{{{*/
 264 // ---------------------------------------------------------------------
 265 /* This is the entry point for an item. An item calls this function when
 266    it is constructed which creates a queue (based on the current queue
 267    mode) and puts the item in that queue. If the system is running then
 268    the queue might be started. */
 269 void pkgAcquire::Enqueue(ItemDesc &Item)
 270 {
 271    // Determine which queue to put the item in
 272    const MethodConfig *Config;
 273    string Name = QueueName(Item.URI,Config);
 274    if (Name.empty() == true)
 275       return;
 276
 277    // Find the queue structure
 278    Queue *I = Queues;
 279    for (; I != 0 && I->Name != Name; I = I->Next);
 280    if (I == 0)
 281    {
 282       I = new Queue(Name,this);
 283       I->Next = Queues;
 284       Queues = I;
 285
 286       if (Running == true)
 287          I->Startup();
 288    }
 289
 290    // See if this is a local only URI
 291    if (Config->LocalOnly == true && Item.Owner->Complete == false)
 292       Item.Owner->Local = true;
 293    Item.Owner->Status = Item::StatIdle;
 294
 295    // Queue it into the named queue
 296    if(I->Enqueue(Item))
 297       ToFetch++;
 298
 299    // Some trace stuff
 300    if (Debug == true)
 301    {
 302       clog << "Fetching " << Item.URI << endl;
 303       clog << " to " << Item.Owner->DestFile << endl;
 304       clog << " Queue is: " << Name << endl;
 305    }
 306 }
 307                                                                         /*}}}*/
 308 // Acquire::Dequeue - Remove an item from all queues                    /*{{{*/
 309 // ---------------------------------------------------------------------
 310 /* This is called when an item is finished being fetched. It removes it
 311    from all the queues */
 312 void pkgAcquire::Dequeue(Item *Itm)
 313 {
 314    Queue *I = Queues;
 315    bool Res = false;
 316    if (Debug == true)
 317       clog << "Dequeuing " << Itm->DestFile << endl;
 318
 319    for (; I != 0; I = I->Next)
 320    {
 321       if (I->Dequeue(Itm))
 322       {
 323          Res = true;
 324          if (Debug == true)
 325             clog << "Dequeued from " << I->Name << endl;
 326       }
 327    }
 328
 329    if (Res == true)
 330       ToFetch--;
 331 }
 332                                                                         /*}}}*/
 333 // Acquire::QueueName - Return the name of the queue for this URI       /*{{{*/
 334 // ---------------------------------------------------------------------
 335 /* The string returned depends on the configuration settings and the
 336    method parameters. Given something like http://foo.org/bar it can
 337    return http://foo.org or http */
 338 string pkgAcquire::QueueName(string Uri,MethodConfig const *&Config)
 339 {
 340    URI U(Uri);
 341
 342    Config = GetConfig(U.Access);
 343    if (Config == 0)
 344       return string();
 345
 346    /* Single-Instance methods get exactly one queue per URI. This is
 347       also used for the Access queue method  */
 348    if (Config->SingleInstance == true || QueueMode == QueueAccess)
 349       return U.Access;
 350
 351    string AccessSchema = U.Access + ':',
 352         FullQueueName = AccessSchema + U.Host;
 353    unsigned int Instances = 0, SchemaLength = AccessSchema.length();
 354
 355    Queue *I = Queues;
 356    for (; I != 0; I = I->Next) {
 357       // if the queue already exists, re-use it
 358       if (I->Name == FullQueueName)
 359          return FullQueueName;
 360
 361       if (I->Name.compare(0, SchemaLength, AccessSchema) == 0)
 362          Instances++;
 363    }
 364
 365    if (Debug) {
 366       clog << "Found " << Instances << " instances of " << U.Access << endl;
 367    }
 368
 369    if (Instances >= (unsigned int)_config->FindI("Acquire::QueueHost::Limit",10))
 370       return U.Access;
 371
 372    return FullQueueName;
 373 }
 374                                                                         /*}}}*/
 375 // Acquire::GetConfig - Fetch the configuration information             /*{{{*/
 376 // ---------------------------------------------------------------------
 377 /* This locates the configuration structure for an access method. If
 378    a config structure cannot be found a Worker will be created to
 379    retrieve it */
 380 pkgAcquire::MethodConfig *pkgAcquire::GetConfig(string Access)
 381 {
 382    // Search for an existing config
 383    MethodConfig *Conf;
 384    for (Conf = Configs; Conf != 0; Conf = Conf->Next)
 385       if (Conf->Access == Access)
 386          return Conf;
 387
 388    // Create the new config class
 389    Conf = new MethodConfig;
 390    Conf->Access = Access;
 391    Conf->Next = Configs;
 392    Configs = Conf;
 393
 394    // Create the worker to fetch the configuration
 395    Worker Work(Conf);
 396    if (Work.Start() == false)
 397       return 0;
 398
 399    /* if a method uses DownloadLimit, we switch to SingleInstance mode */
 400    if(_config->FindI("Acquire::"+Access+"::Dl-Limit",0) > 0)
 401       Conf->SingleInstance = true;
 402
 403    return Conf;
 404 }
 405                                                                         /*}}}*/
 406 // Acquire::SetFds - Deal with readable FDs                             /*{{{*/
 407 // ---------------------------------------------------------------------
 408 /* Collect FDs that have activity monitors into the fd sets */
 409 void pkgAcquire::SetFds(int &Fd,fd_set *RSet,fd_set *WSet)
 410 {
 411    for (Worker *I = Workers; I != 0; I = I->NextAcquire)
 412    {
 413       if (I->InReady == true && I->InFd >= 0)
 414       {
 415          if (Fd < I->InFd)
 416             Fd = I->InFd;
 417          FD_SET(I->InFd,RSet);
 418       }
 419       if (I->OutReady == true && I->OutFd >= 0)
 420       {
 421          if (Fd < I->OutFd)
 422             Fd = I->OutFd;
 423          FD_SET(I->OutFd,WSet);
 424       }
 425    }
 426 }
 427                                                                         /*}}}*/
 428 // Acquire::RunFds - Deal with active FDs                               /*{{{*/
 429 // ---------------------------------------------------------------------
 430 /* Dispatch active FDs over to the proper workers. It is very important
 431    that a worker never be erased while this is running! The queue class
 432    should never erase a worker except during shutdown processing. */
 433 void pkgAcquire::RunFds(fd_set *RSet,fd_set *WSet)
 434 {
 435    for (Worker *I = Workers; I != 0; I = I->NextAcquire)
 436    {
 437       if (I->InFd >= 0 && FD_ISSET(I->InFd,RSet) != 0)
 438          I->InFdReady();
 439       if (I->OutFd >= 0 && FD_ISSET(I->OutFd,WSet) != 0)
 440          I->OutFdReady();
 441    }
 442 }
 443                                                                         /*}}}*/
 444 // Acquire::Run - Run the fetch sequence                                /*{{{*/
 445 // ---------------------------------------------------------------------
 446 /* This runs the queues. It manages a select loop for all of the
 447    Worker tasks. The workers interact with the queues and items to
 448    manage the actual fetch. */
 449 pkgAcquire::RunResult pkgAcquire::Run(int PulseIntervall)
 450 {
 451    Running = true;
 452
 453    for (Queue *I = Queues; I != 0; I = I->Next)
 454       I->Startup();
 455
 456    if (Log != 0)
 457       Log->Start();
 458
 459    bool WasCancelled = false;
 460
 461    // Run till all things have been acquired
 462    struct timeval tv;
 463    tv.tv_sec = 0;
 464    tv.tv_usec = PulseIntervall;
 465    while (ToFetch > 0)
 466    {
 467       fd_set RFds;
 468       fd_set WFds;
 469       int Highest = 0;
 470       FD_ZERO(&RFds);
 471       FD_ZERO(&WFds);
 472       SetFds(Highest,&RFds,&WFds);
 473
 474       int Res;
 475       do
 476       {
 477          Res = select(Highest+1,&RFds,&WFds,0,&tv);
 478       }
 479       while (Res < 0 && errno == EINTR);
 480
 481       if (Res < 0)
 482       {
 483          _error->Errno("select","Select has failed");
 484          break;
 485       }
 486
 487       RunFds(&RFds,&WFds);
 488       if (_error->PendingError() == true)
 489          break;
 490
 491       // Timeout, notify the log class
 492       if (Res == 0 || (Log != 0 && Log->Update == true))
 493       {
 494          tv.tv_usec = PulseIntervall;
 495          for (Worker *I = Workers; I != 0; I = I->NextAcquire)
 496             I->Pulse();
 497          if (Log != 0 && Log->Pulse(this) == false)
 498          {
 499             WasCancelled = true;
 500             break;
 501          }
 502       }
 503    }
 504
 505    if (Log != 0)
 506       Log->Stop();
 507
 508    // Shut down the acquire bits
 509    Running = false;
 510    for (Queue *I = Queues; I != 0; I = I->Next)
 511       I->Shutdown(false);
 512
 513    // Shut down the items
 514    for (ItemIterator I = Items.begin(); I != Items.end(); ++I)
 515       (*I)->Finished();
 516
 517    if (_error->PendingError())
 518       return Failed;
 519    if (WasCancelled)
 520       return Cancelled;
 521    return Continue;
 522 }
 523                                                                         /*}}}*/
 524 // Acquire::Bump - Called when an item is dequeued                      /*{{{*/
 525 // ---------------------------------------------------------------------
 526 /* This routine bumps idle queues in hopes that they will be able to fetch
 527    the dequeued item */
 528 void pkgAcquire::Bump()
 529 {
 530    for (Queue *I = Queues; I != 0; I = I->Next)
 531       I->Bump();
 532 }
 533                                                                         /*}}}*/
 534 // Acquire::WorkerStep - Step to the next worker                        /*{{{*/
 535 // ---------------------------------------------------------------------
 536 /* Not inlined to advoid including acquire-worker.h */
 537 pkgAcquire::Worker *pkgAcquire::WorkerStep(Worker *I)
 538 {
 539    return I->NextAcquire;
 540 }
 541                                                                         /*}}}*/
 542 // Acquire::Clean - Cleans a directory                                  /*{{{*/
 543 // ---------------------------------------------------------------------
 544 /* This is a bit simplistic, it looks at every file in the dir and sees
 545    if it is part of the download set. */
 546 bool pkgAcquire::Clean(string Dir)
 547 {
 548    // non-existing directories are by definition clean…
 549    if (DirectoryExists(Dir) == false)
 550       return true;
 551
 552    if(Dir == "/")
 553       return _error->Error(_("Clean of %s is not supported"), Dir.c_str());
 554
 555    DIR *D = opendir(Dir.c_str());
 556    if (D == 0)
 557       return _error->Errno("opendir",_("Unable to read %s"),Dir.c_str());
 558
 559    string StartDir = SafeGetCWD();
 560    if (chdir(Dir.c_str()) != 0)
 561    {
 562       closedir(D);
 563       return _error->Errno("chdir",_("Unable to change to %s"),Dir.c_str());
 564    }
 565
 566    for (struct dirent *Dir = readdir(D); Dir != 0; Dir = readdir(D))
 567    {
 568       // Skip some files..
 569       if (strcmp(Dir->d_name,"lock") == 0 ||
 570           strcmp(Dir->d_name,"partial") == 0 ||
 571           strcmp(Dir->d_name,".") == 0 ||
 572           strcmp(Dir->d_name,"..") == 0)
 573          continue;
 574
 575       // Look in the get list
 576       ItemCIterator I = Items.begin();
 577       for (; I != Items.end(); ++I)
 578          if (flNotDir((*I)->DestFile) == Dir->d_name)
 579             break;
 580
 581       // Nothing found, nuke it
 582       if (I == Items.end())
 583          unlink(Dir->d_name);
 584    };
 585
 586    closedir(D);
 587    if (chdir(StartDir.c_str()) != 0)
 588       return _error->Errno("chdir",_("Unable to change to %s"),StartDir.c_str());
 589    return true;
 590 }
 591                                                                         /*}}}*/
 592 // Acquire::TotalNeeded - Number of bytes to fetch                      /*{{{*/
 593 // ---------------------------------------------------------------------
 594 /* This is the total number of bytes needed */
 595 APT_PURE unsigned long long pkgAcquire::TotalNeeded()
 596 {
 597    unsigned long long Total = 0;
 598    for (ItemCIterator I = ItemsBegin(); I != ItemsEnd(); ++I)
 599       Total += (*I)->FileSize;
 600    return Total;
 601 }
 602                                                                         /*}}}*/
 603 // Acquire::FetchNeeded - Number of bytes needed to get                 /*{{{*/
 604 // ---------------------------------------------------------------------
 605 /* This is the number of bytes that is not local */
 606 APT_PURE unsigned long long pkgAcquire::FetchNeeded()
 607 {
 608    unsigned long long Total = 0;
 609    for (ItemCIterator I = ItemsBegin(); I != ItemsEnd(); ++I)
 610       if ((*I)->Local == false)
 611          Total += (*I)->FileSize;
 612    return Total;
 613 }
 614                                                                         /*}}}*/
 615 // Acquire::PartialPresent - Number of partial bytes we already have    /*{{{*/
 616 // ---------------------------------------------------------------------
 617 /* This is the number of bytes that is not local */
 618 APT_PURE unsigned long long pkgAcquire::PartialPresent()
 619 {
 620   unsigned long long Total = 0;
 621    for (ItemCIterator I = ItemsBegin(); I != ItemsEnd(); ++I)
 622       if ((*I)->Local == false)
 623          Total += (*I)->PartialSize;
 624    return Total;
 625 }
 626                                                                         /*}}}*/
 627 // Acquire::UriBegin - Start iterator for the uri list                  /*{{{*/
 628 // ---------------------------------------------------------------------
 629 /* */
 630 pkgAcquire::UriIterator pkgAcquire::UriBegin()
 631 {
 632    return UriIterator(Queues);
 633 }
 634                                                                         /*}}}*/
 635 // Acquire::UriEnd - End iterator for the uri list                      /*{{{*/
 636 // ---------------------------------------------------------------------
 637 /* */
 638 pkgAcquire::UriIterator pkgAcquire::UriEnd()
 639 {
 640    return UriIterator(0);
 641 }
 642                                                                         /*}}}*/
 643 // Acquire::MethodConfig::MethodConfig - Constructor                    /*{{{*/
 644 // ---------------------------------------------------------------------
 645 /* */
 646 pkgAcquire::MethodConfig::MethodConfig() : d(NULL), Next(0), SingleInstance(false),
 647    Pipeline(false), SendConfig(false), LocalOnly(false), NeedsCleanup(false),
 648    Removable(false)
 649 {
 650 }
 651                                                                         /*}}}*/
 652 // Queue::Queue - Constructor                                           /*{{{*/
 653 // ---------------------------------------------------------------------
 654 /* */
 655 pkgAcquire::Queue::Queue(string const &name,pkgAcquire * const owner) : d(NULL), Next(0),
 656    Name(name), Items(0), Workers(0), Owner(owner), PipeDepth(0), MaxPipeDepth(1)
 657 {
 658 }
 659                                                                         /*}}}*/
 660 // Queue::~Queue - Destructor                                           /*{{{*/
 661 // ---------------------------------------------------------------------
 662 /* */
 663 pkgAcquire::Queue::~Queue()
 664 {
 665    Shutdown(true);
 666
 667    while (Items != 0)
 668    {
 669       QItem *Jnk = Items;
 670       Items = Items->Next;
 671       delete Jnk;
 672    }
 673 }
 674                                                                         /*}}}*/
 675 // Queue::Enqueue - Queue an item to the queue                          /*{{{*/
 676 // ---------------------------------------------------------------------
 677 /* */
 678 bool pkgAcquire::Queue::Enqueue(ItemDesc &Item)
 679 {
 680    QItem **I = &Items;
 681    // move to the end of the queue and check for duplicates here
 682    HashStringList const hsl = Item.Owner->GetExpectedHashes();
 683    for (; *I != 0; I = &(*I)->Next)
 684       if (Item.URI == (*I)->URI || hsl == (*I)->Owner->GetExpectedHashes())
 685       {
 686          if (_config->FindB("Debug::pkgAcquire::Worker",false) == true)
 687             std::cerr << " @ Queue: Action combined for " << Item.URI << " and " << (*I)->URI << std::endl;
 688          (*I)->Owners.push_back(Item.Owner);
 689          Item.Owner->Status = (*I)->Owner->Status;
 690          return false;
 691       }
 692
 693    // Create a new item
 694    QItem *Itm = new QItem;
 695    *Itm = Item;
 696    Itm->Next = 0;
 697    *I = Itm;
 698
 699    Item.Owner->QueueCounter++;
 700    if (Items->Next == 0)
 701       Cycle();
 702    return true;
 703 }
 704                                                                         /*}}}*/
 705 // Queue::Dequeue - Remove an item from the queue                       /*{{{*/
 706 // ---------------------------------------------------------------------
 707 /* We return true if we hit something */
 708 bool pkgAcquire::Queue::Dequeue(Item *Owner)
 709 {
 710    if (Owner->Status == pkgAcquire::Item::StatFetching)
 711       return _error->Error("Tried to dequeue a fetching object");
 712
 713    bool Res = false;
 714
 715    QItem **I = &Items;
 716    for (; *I != 0;)
 717    {
 718       if (Owner == (*I)->Owner)
 719       {
 720          QItem *Jnk= *I;
 721          *I = (*I)->Next;
 722          Owner->QueueCounter--;
 723          delete Jnk;
 724          Res = true;
 725       }
 726       else
 727          I = &(*I)->Next;
 728    }
 729
 730    return Res;
 731 }
 732                                                                         /*}}}*/
 733 // Queue::Startup - Start the worker processes                          /*{{{*/
 734 // ---------------------------------------------------------------------
 735 /* It is possible for this to be called with a pre-existing set of
 736    workers. */
 737 bool pkgAcquire::Queue::Startup()
 738 {
 739    if (Workers == 0)
 740    {
 741       URI U(Name);
 742       pkgAcquire::MethodConfig *Cnf = Owner->GetConfig(U.Access);
 743       if (Cnf == 0)
 744          return false;
 745
 746       Workers = new Worker(this,Cnf,Owner->Log);
 747       Owner->Add(Workers);
 748       if (Workers->Start() == false)
 749          return false;
 750
 751       /* When pipelining we commit 10 items. This needs to change when we
 752          added other source retry to have cycle maintain a pipeline depth
 753          on its own. */
 754       if (Cnf->Pipeline == true)
 755          MaxPipeDepth = _config->FindI("Acquire::Max-Pipeline-Depth",10);
 756       else
 757          MaxPipeDepth = 1;
 758    }
 759
 760    return Cycle();
 761 }
 762                                                                         /*}}}*/
 763 // Queue::Shutdown - Shutdown the worker processes                      /*{{{*/
 764 // ---------------------------------------------------------------------
 765 /* If final is true then all workers are eliminated, otherwise only workers
 766    that do not need cleanup are removed */
 767 bool pkgAcquire::Queue::Shutdown(bool Final)
 768 {
 769    // Delete all of the workers
 770    pkgAcquire::Worker **Cur = &Workers;
 771    while (*Cur != 0)
 772    {
 773       pkgAcquire::Worker *Jnk = *Cur;
 774       if (Final == true || Jnk->GetConf()->NeedsCleanup == false)
 775       {
 776          *Cur = Jnk->NextQueue;
 777          Owner->Remove(Jnk);
 778          delete Jnk;
 779       }
 780       else
 781          Cur = &(*Cur)->NextQueue;
 782    }
 783
 784    return true;
 785 }
 786                                                                         /*}}}*/
 787 // Queue::FindItem - Find a URI in the item list                        /*{{{*/
 788 // ---------------------------------------------------------------------
 789 /* */
 790 pkgAcquire::Queue::QItem *pkgAcquire::Queue::FindItem(string URI,pkgAcquire::Worker *Owner)
 791 {
 792    for (QItem *I = Items; I != 0; I = I->Next)
 793       if (I->URI == URI && I->Worker == Owner)
 794          return I;
 795    return 0;
 796 }
 797                                                                         /*}}}*/
 798 // Queue::ItemDone - Item has been completed                            /*{{{*/
 799 // ---------------------------------------------------------------------
 800 /* The worker signals this which causes the item to be removed from the
 801    queue. If this is the last queue instance then it is removed from the
 802    main queue too.*/
 803 bool pkgAcquire::Queue::ItemDone(QItem *Itm)
 804 {
 805    PipeDepth--;
 806    for (QItem::owner_iterator O = Itm->Owners.begin(); O != Itm->Owners.end(); ++O)
 807    {
 808       if ((*O)->Status == pkgAcquire::Item::StatFetching)
 809          (*O)->Status = pkgAcquire::Item::StatDone;
 810    }
 811
 812    if (Itm->Owner->QueueCounter <= 1)
 813       Owner->Dequeue(Itm->Owner);
 814    else
 815    {
 816       Dequeue(Itm->Owner);
 817       Owner->Bump();
 818    }
 819
 820    return Cycle();
 821 }
 822                                                                         /*}}}*/
 823 // Queue::Cycle - Queue new items into the method                       /*{{{*/
 824 // ---------------------------------------------------------------------
 825 /* This locates a new idle item and sends it to the worker. If pipelining
 826    is enabled then it keeps the pipe full. */
 827 bool pkgAcquire::Queue::Cycle()
 828 {
 829    if (Items == 0 || Workers == 0)
 830       return true;
 831
 832    if (PipeDepth < 0)
 833       return _error->Error("Pipedepth failure");
 834
 835    // Look for a queable item
 836    QItem *I = Items;
 837    while (PipeDepth < (signed)MaxPipeDepth)
 838    {
 839       for (; I != 0; I = I->Next)
 840          if (I->Owner->Status == pkgAcquire::Item::StatIdle)
 841             break;
 842
 843       // Nothing to do, queue is idle.
 844       if (I == 0)
 845          return true;
 846
 847       I->Worker = Workers;
 848       for (QItem::owner_iterator O = I->Owners.begin(); O != I->Owners.end(); ++O)
 849          (*O)->Status = pkgAcquire::Item::StatFetching;
 850       PipeDepth++;
 851       if (Workers->QueueItem(I) == false)
 852          return false;
 853    }
 854
 855    return true;
 856 }
 857                                                                         /*}}}*/
 858 // Queue::Bump - Fetch any pending objects if we are idle               /*{{{*/
 859 // ---------------------------------------------------------------------
 860 /* This is called when an item in multiple queues is dequeued */
 861 void pkgAcquire::Queue::Bump()
 862 {
 863    Cycle();
 864 }
 865                                                                         /*}}}*/
 866 HashStringList pkgAcquire::Queue::QItem::GetExpectedHashes() const      /*{{{*/
 867 {
 868    /* each Item can have multiple owners and each owner might have different
 869       hashes, even if that is unlikely in practice and if so at least some
 870       owners will later fail. There is one situation through which is not a
 871       failure and still needs this handling: Two owners who expect the same
 872       file, but one owner only knows the SHA1 while the other only knows SHA256. */
 873    HashStringList superhsl;
 874    for (pkgAcquire::Queue::QItem::owner_iterator O = Owners.begin(); O != Owners.end(); ++O)
 875    {
 876       HashStringList const hsl = (*O)->GetExpectedHashes();
 877       if (hsl.usable() == false)
 878          continue;
 879       if (superhsl.usable() == false)
 880          superhsl = hsl;
 881       else
 882       {
 883          // we merge both lists - if we find disagreement send no hashes
 884          HashStringList::const_iterator hs = hsl.begin();
 885          for (; hs != hsl.end(); ++hs)
 886             if (superhsl.push_back(*hs) == false)
 887                break;
 888          if (hs != hsl.end())
 889          {
 890             superhsl.clear();
 891             break;
 892          }
 893       }
 894    }
 895    return superhsl;
 896 }
 897                                                                         /*}}}*/
 898 APT_PURE unsigned long long pkgAcquire::Queue::QItem::GetMaximumSize() const    /*{{{*/
 899 {
 900    unsigned long long Maximum = std::numeric_limits<unsigned long long>::max();
 901    for (pkgAcquire::Queue::QItem::owner_iterator O = Owners.begin(); O != Owners.end(); ++O)
 902    {
 903       if ((*O)->FileSize == 0)
 904          continue;
 905       Maximum = std::min(Maximum, (*O)->FileSize);
 906    }
 907    if (Maximum == std::numeric_limits<unsigned long long>::max())
 908       return 0;
 909    return Maximum;
 910 }
 911                                                                         /*}}}*/
 912 void pkgAcquire::Queue::QItem::SyncDestinationFiles() const             /*{{{*/
 913 {
 914    /* ensure that the first owner has the best partial file of all and
 915       the rest have (potentially dangling) symlinks to it so that
 916       everything (like progress reporting) finds it easily */
 917    std::string superfile = Owner->DestFile;
 918    off_t supersize = 0;
 919    for (pkgAcquire::Queue::QItem::owner_iterator O = Owners.begin(); O != Owners.end(); ++O)
 920    {
 921       if ((*O)->DestFile == superfile)
 922          continue;
 923       struct stat file;
 924       if (lstat((*O)->DestFile.c_str(),&file) == 0)
 925       {
 926          if ((file.st_mode & S_IFREG) == 0)
 927             unlink((*O)->DestFile.c_str());
 928          else if (supersize < file.st_size)
 929          {
 930             supersize = file.st_size;
 931             unlink(superfile.c_str());
 932             rename((*O)->DestFile.c_str(), superfile.c_str());
 933          }
 934          else
 935             unlink((*O)->DestFile.c_str());
 936          if (symlink(superfile.c_str(), (*O)->DestFile.c_str()) != 0)
 937          {
 938             ; // not a problem per-se and no real alternative
 939          }
 940       }
 941    }
 942 }
 943                                                                         /*}}}*/
 944 std::string pkgAcquire::Queue::QItem::Custom600Headers() const          /*{{{*/
 945 {
 946    /* The others are relatively easy to merge, but this one?
 947       Lets not merge and see how far we can run with it…
 948       Likely, nobody will ever notice as all the items will
 949       be of the same class and hence generate the same headers. */
 950    return Owner->Custom600Headers();
 951 }
 952                                                                         /*}}}*/
 953
 954 // AcquireStatus::pkgAcquireStatus - Constructor                        /*{{{*/
 955 // ---------------------------------------------------------------------
 956 /* */
 957 pkgAcquireStatus::pkgAcquireStatus() : d(NULL), Percent(-1), Update(true), MorePulses(false)
 958 {
 959    Start();
 960 }
 961                                                                         /*}}}*/
 962 // AcquireStatus::Pulse - Called periodically                           /*{{{*/
 963 // ---------------------------------------------------------------------
 964 /* This computes some internal state variables for the derived classes to
 965    use. It generates the current downloaded bytes and total bytes to download
 966    as well as the current CPS estimate. */
 967 bool pkgAcquireStatus::Pulse(pkgAcquire *Owner)
 968 {
 969    TotalBytes = 0;
 970    CurrentBytes = 0;
 971    TotalItems = 0;
 972    CurrentItems = 0;
 973
 974    // Compute the total number of bytes to fetch
 975    unsigned int Unknown = 0;
 976    unsigned int Count = 0;
 977    bool UnfetchedReleaseFiles = false;
 978    for (pkgAcquire::ItemCIterator I = Owner->ItemsBegin();
 979         I != Owner->ItemsEnd();
 980         ++I, ++Count)
 981    {
 982       TotalItems++;
 983       if ((*I)->Status == pkgAcquire::Item::StatDone)
 984          ++CurrentItems;
 985
 986       // Totally ignore local items
 987       if ((*I)->Local == true)
 988          continue;
 989
 990       // see if the method tells us to expect more
 991       TotalItems += (*I)->ExpectedAdditionalItems;
 992
 993       // check if there are unfetched Release files
 994       if ((*I)->Complete == false && (*I)->ExpectedAdditionalItems > 0)
 995          UnfetchedReleaseFiles = true;
 996
 997       TotalBytes += (*I)->FileSize;
 998       if ((*I)->Complete == true)
 999          CurrentBytes += (*I)->FileSize;
1000       if ((*I)->FileSize == 0 && (*I)->Complete == false)
1001          ++Unknown;
1002    }
1003
1004    // Compute the current completion
1005    unsigned long long ResumeSize = 0;
1006    for (pkgAcquire::Worker *I = Owner->WorkersBegin(); I != 0;
1007         I = Owner->WorkerStep(I))
1008    {
1009       if (I->CurrentItem != 0 && I->CurrentItem->Owner->Complete == false)
1010       {
1011          CurrentBytes += I->CurrentSize;
1012          ResumeSize += I->ResumePoint;
1013
1014          // Files with unknown size always have 100% completion
1015          if (I->CurrentItem->Owner->FileSize == 0 &&
1016              I->CurrentItem->Owner->Complete == false)
1017             TotalBytes += I->CurrentSize;
1018       }
1019    }
1020
1021    // Normalize the figures and account for unknown size downloads
1022    if (TotalBytes <= 0)
1023       TotalBytes = 1;
1024    if (Unknown == Count)
1025       TotalBytes = Unknown;
1026
1027    // Wha?! Is not supposed to happen.
1028    if (CurrentBytes > TotalBytes)
1029       CurrentBytes = TotalBytes;
1030
1031    // debug
1032    if (_config->FindB("Debug::acquire::progress", false) == true)
1033       std::clog << " Bytes: "
1034                 << SizeToStr(CurrentBytes) << " / " << SizeToStr(TotalBytes)
1035                 << std::endl;
1036
1037    // Compute the CPS
1038    struct timeval NewTime;
1039    gettimeofday(&NewTime,0);
1040    if ((NewTime.tv_sec - Time.tv_sec == 6 && NewTime.tv_usec > Time.tv_usec) ||
1041        NewTime.tv_sec - Time.tv_sec > 6)
1042    {
1043       double Delta = NewTime.tv_sec - Time.tv_sec +
1044                      (NewTime.tv_usec - Time.tv_usec)/1000000.0;
1045
1046       // Compute the CPS value
1047       if (Delta < 0.01)
1048          CurrentCPS = 0;
1049       else
1050          CurrentCPS = ((CurrentBytes - ResumeSize) - LastBytes)/Delta;
1051       LastBytes = CurrentBytes - ResumeSize;
1052       ElapsedTime = (unsigned long long)Delta;
1053       Time = NewTime;
1054    }
1055
1056    double const OldPercent = Percent;
1057    // calculate the percentage, if we have too little data assume 1%
1058    if (TotalBytes > 0 && UnfetchedReleaseFiles)
1059       Percent = 0;
1060    else
1061       // use both files and bytes because bytes can be unreliable
1062       Percent = (0.8 * (CurrentBytes/float(TotalBytes)*100.0) +
1063                  0.2 * (CurrentItems/float(TotalItems)*100.0));
1064    double const DiffPercent = Percent - OldPercent;
1065    if (DiffPercent < 0.001 && _config->FindB("Acquire::Progress::Diffpercent", false) == true)
1066       return true;
1067
1068    int fd = _config->FindI("APT::Status-Fd",-1);
1069    if(fd > 0)
1070    {
1071       ostringstream status;
1072
1073       char msg[200];
1074       long i = CurrentItems < TotalItems ? CurrentItems + 1 : CurrentItems;
1075       unsigned long long ETA = 0;
1076       if(CurrentCPS > 0)
1077          ETA = (TotalBytes - CurrentBytes) / CurrentCPS;
1078
1079       // only show the ETA if it makes sense
1080       if (ETA > 0 && ETA < 172800 /* two days */ )
1081          snprintf(msg,sizeof(msg), _("Retrieving file %li of %li (%s remaining)"), i, TotalItems, TimeToStr(ETA).c_str());
1082       else
1083          snprintf(msg,sizeof(msg), _("Retrieving file %li of %li"), i, TotalItems);
1084
1085       // build the status str
1086       status << "dlstatus:" << i
1087              << ":"  << std::setprecision(3) << Percent
1088              << ":" << msg
1089              << endl;
1090
1091       std::string const dlstatus = status.str();
1092       FileFd::Write(fd, dlstatus.c_str(), dlstatus.size());
1093    }
1094
1095    return true;
1096 }
1097                                                                         /*}}}*/
1098 // AcquireStatus::Start - Called when the download is started           /*{{{*/
1099 // ---------------------------------------------------------------------
1100 /* We just reset the counters */
1101 void pkgAcquireStatus::Start()
1102 {
1103    gettimeofday(&Time,0);
1104    gettimeofday(&StartTime,0);
1105    LastBytes = 0;
1106    CurrentCPS = 0;
1107    CurrentBytes = 0;
1108    TotalBytes = 0;
1109    FetchedBytes = 0;
1110    ElapsedTime = 0;
1111    TotalItems = 0;
1112    CurrentItems = 0;
1113 }
1114                                                                         /*}}}*/
1115 // AcquireStatus::Stop - Finished downloading                           /*{{{*/
1116 // ---------------------------------------------------------------------
1117 /* This accurately computes the elapsed time and the total overall CPS. */
1118 void pkgAcquireStatus::Stop()
1119 {
1120    // Compute the CPS and elapsed time
1121    struct timeval NewTime;
1122    gettimeofday(&NewTime,0);
1123
1124    double Delta = NewTime.tv_sec - StartTime.tv_sec +
1125                   (NewTime.tv_usec - StartTime.tv_usec)/1000000.0;
1126
1127    // Compute the CPS value
1128    if (Delta < 0.01)
1129       CurrentCPS = 0;
1130    else
1131       CurrentCPS = FetchedBytes/Delta;
1132    LastBytes = CurrentBytes;
1133    ElapsedTime = (unsigned long long)Delta;
1134 }
1135                                                                         /*}}}*/
1136 // AcquireStatus::Fetched - Called when a byte set has been fetched     /*{{{*/
1137 // ---------------------------------------------------------------------
1138 /* This is used to get accurate final transfer rate reporting. */
1139 void pkgAcquireStatus::Fetched(unsigned long long Size,unsigned long long Resume)
1140 {
1141    FetchedBytes += Size - Resume;
1142 }
1143                                                                         /*}}}*/
1144
1145 pkgAcquire::UriIterator::UriIterator(pkgAcquire::Queue *Q) : d(NULL), CurQ(Q), CurItem(0)
1146 {
1147    while (CurItem == 0 && CurQ != 0)
1148    {
1149       CurItem = CurQ->Items;
1150       CurQ = CurQ->Next;
1151    }
1152 }
1153
1154 APT_CONST pkgAcquire::UriIterator::~UriIterator() {}
1155 APT_CONST pkgAcquire::MethodConfig::~MethodConfig() {}
1156 APT_CONST pkgAcquireStatus::~pkgAcquireStatus() {}