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