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 schedual 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 controled by how the queue
  11    name is derived from the URI.
  12
  13    ##################################################################### */
  14                                                                         /*}}}*/
  15 // Include Files                                                        /*{{{*/
  16 #ifdef __GNUG__
  17 #pragma implementation "apt-pkg/acquire.h"
  18 #endif
  19 #include <apt-pkg/acquire.h>
  20 #include <apt-pkg/acquire-item.h>
  21 #include <apt-pkg/acquire-worker.h>
  22 #include <apt-pkg/configuration.h>
  23 #include <apt-pkg/error.h>
  24 #include <apt-pkg/strutl.h>
  25
  26 #include <apti18n.h>
  27
  28 #include <iostream>
  29
  30 #include <dirent.h>
  31 #include <sys/time.h>
  32 #include <errno.h>
  33 #include <sys/stat.h>
  34                                                                         /*}}}*/
  35
  36 using namespace std;
  37
  38 // Acquire::pkgAcquire - Constructor                                    /*{{{*/
  39 // ---------------------------------------------------------------------
  40 /* We grab some runtime state from the configuration space */
  41 pkgAcquire::pkgAcquire(pkgAcquireStatus *Log) : Log(Log)
  42 {
  43    Queues = 0;
  44    Configs = 0;
  45    Workers = 0;
  46    ToFetch = 0;
  47    Running = false;
  48
  49    string Mode = _config->Find("Acquire::Queue-Mode","host");
  50    if (strcasecmp(Mode.c_str(),"host") == 0)
  51       QueueMode = QueueHost;
  52    if (strcasecmp(Mode.c_str(),"access") == 0)
  53       QueueMode = QueueAccess;
  54
  55    Debug = _config->FindB("Debug::pkgAcquire",false);
  56
  57    // This is really a stupid place for this
  58    struct stat St;
  59    if (stat((_config->FindDir("Dir::State::lists") + "partial/").c_str(),&St) != 0 ||
  60        S_ISDIR(St.st_mode) == 0)
  61       _error->Error(_("Lists directory %spartial is missing."),
  62                     _config->FindDir("Dir::State::lists").c_str());
  63    if (stat((_config->FindDir("Dir::Cache::Archives") + "partial/").c_str(),&St) != 0 ||
  64        S_ISDIR(St.st_mode) == 0)
  65       _error->Error(_("Archive directory %spartial is missing."),
  66                     _config->FindDir("Dir::Cache::Archives").c_str());
  67 }
  68                                                                         /*}}}*/
  69 // Acquire::~pkgAcquire - Destructor                                    /*{{{*/
  70 // ---------------------------------------------------------------------
  71 /* Free our memory, clean up the queues (destroy the workers) */
  72 pkgAcquire::~pkgAcquire()
  73 {
  74    Shutdown();
  75
  76    while (Configs != 0)
  77    {
  78       MethodConfig *Jnk = Configs;
  79       Configs = Configs->Next;
  80       delete Jnk;
  81    }
  82 }
  83                                                                         /*}}}*/
  84 // Acquire::Shutdown - Clean out the acquire object                     /*{{{*/
  85 // ---------------------------------------------------------------------
  86 /* */
  87 void pkgAcquire::Shutdown()
  88 {
  89    while (Items.size() != 0)
  90    {
  91       if (Items[0]->Status == Item::StatFetching)
  92          Items[0]->Status = Item::StatError;
  93       delete Items[0];
  94    }
  95
  96    while (Queues != 0)
  97    {
  98       Queue *Jnk = Queues;
  99       Queues = Queues->Next;
 100       delete Jnk;
 101    }
 102 }
 103                                                                         /*}}}*/
 104 // Acquire::Add - Add a new item                                        /*{{{*/
 105 // ---------------------------------------------------------------------
 106 /* This puts an item on the acquire list. This list is mainly for tracking
 107    item status */
 108 void pkgAcquire::Add(Item *Itm)
 109 {
 110    Items.push_back(Itm);
 111 }
 112                                                                         /*}}}*/
 113 // Acquire::Remove - Remove a item                                      /*{{{*/
 114 // ---------------------------------------------------------------------
 115 /* Remove an item from the acquire list. This is usually not used.. */
 116 void pkgAcquire::Remove(Item *Itm)
 117 {
 118    Dequeue(Itm);
 119
 120    for (ItemIterator I = Items.begin(); I != Items.end();)
 121    {
 122       if (*I == Itm)
 123       {
 124          Items.erase(I);
 125          I = Items.begin();
 126       }
 127       else
 128          I++;
 129    }
 130 }
 131                                                                         /*}}}*/
 132 // Acquire::Add - Add a worker                                          /*{{{*/
 133 // ---------------------------------------------------------------------
 134 /* A list of workers is kept so that the select loop can direct their FD
 135    usage. */
 136 void pkgAcquire::Add(Worker *Work)
 137 {
 138    Work->NextAcquire = Workers;
 139    Workers = Work;
 140 }
 141                                                                         /*}}}*/
 142 // Acquire::Remove - Remove a worker                                    /*{{{*/
 143 // ---------------------------------------------------------------------
 144 /* A worker has died. This can not be done while the select loop is running
 145    as it would require that RunFds could handling a changing list state and
 146    it cant.. */
 147 void pkgAcquire::Remove(Worker *Work)
 148 {
 149    if (Running == true)
 150       abort();
 151
 152    Worker **I = &Workers;
 153    for (; *I != 0;)
 154    {
 155       if (*I == Work)
 156          *I = (*I)->NextAcquire;
 157       else
 158          I = &(*I)->NextAcquire;
 159    }
 160 }
 161                                                                         /*}}}*/
 162 // Acquire::Enqueue - Queue an URI for fetching                         /*{{{*/
 163 // ---------------------------------------------------------------------
 164 /* This is the entry point for an item. An item calls this function when
 165    it is constructed which creates a queue (based on the current queue
 166    mode) and puts the item in that queue. If the system is running then
 167    the queue might be started. */
 168 void pkgAcquire::Enqueue(ItemDesc &Item)
 169 {
 170    // Determine which queue to put the item in
 171    const MethodConfig *Config;
 172    string Name = QueueName(Item.URI,Config);
 173    if (Name.empty() == true)
 174       return;
 175
 176    // Find the queue structure
 177    Queue *I = Queues;
 178    for (; I != 0 && I->Name != Name; I = I->Next);
 179    if (I == 0)
 180    {
 181       I = new Queue(Name,this);
 182       I->Next = Queues;
 183       Queues = I;
 184
 185       if (Running == true)
 186          I->Startup();
 187    }
 188
 189    // See if this is a local only URI
 190    if (Config->LocalOnly == true && Item.Owner->Complete == false)
 191       Item.Owner->Local = true;
 192    Item.Owner->Status = Item::StatIdle;
 193
 194    // Queue it into the named queue
 195    I->Enqueue(Item);
 196    ToFetch++;
 197
 198    // Some trace stuff
 199    if (Debug == true)
 200    {
 201       clog << "Fetching " << Item.URI << endl;
 202       clog << " to " << Item.Owner->DestFile << endl;
 203       clog << " Queue is: " << Name << endl;
 204    }
 205 }
 206                                                                         /*}}}*/
 207 // Acquire::Dequeue - Remove an item from all queues                    /*{{{*/
 208 // ---------------------------------------------------------------------
 209 /* This is called when an item is finished being fetched. It removes it
 210    from all the queues */
 211 void pkgAcquire::Dequeue(Item *Itm)
 212 {
 213    Queue *I = Queues;
 214    bool Res = false;
 215    for (; I != 0; I = I->Next)
 216       Res |= I->Dequeue(Itm);
 217
 218    if (Debug == true)
 219       clog << "Dequeuing " << Itm->DestFile << endl;
 220    if (Res == true)
 221       ToFetch--;
 222 }
 223                                                                         /*}}}*/
 224 // Acquire::QueueName - Return the name of the queue for this URI       /*{{{*/
 225 // ---------------------------------------------------------------------
 226 /* The string returned depends on the configuration settings and the
 227    method parameters. Given something like http://foo.org/bar it can
 228    return http://foo.org or http */
 229 string pkgAcquire::QueueName(string Uri,MethodConfig const *&Config)
 230 {
 231    URI U(Uri);
 232
 233    Config = GetConfig(U.Access);
 234    if (Config == 0)
 235       return string();
 236
 237    /* Single-Instance methods get exactly one queue per URI. This is
 238       also used for the Access queue method  */
 239    if (Config->SingleInstance == true || QueueMode == QueueAccess)
 240        return U.Access;
 241
 242    return U.Access + ':' + U.Host;
 243 }
 244                                                                         /*}}}*/
 245 // Acquire::GetConfig - Fetch the configuration information             /*{{{*/
 246 // ---------------------------------------------------------------------
 247 /* This locates the configuration structure for an access method. If
 248    a config structure cannot be found a Worker will be created to
 249    retrieve it */
 250 pkgAcquire::MethodConfig *pkgAcquire::GetConfig(string Access)
 251 {
 252    // Search for an existing config
 253    MethodConfig *Conf;
 254    for (Conf = Configs; Conf != 0; Conf = Conf->Next)
 255       if (Conf->Access == Access)
 256          return Conf;
 257
 258    // Create the new config class
 259    Conf = new MethodConfig;
 260    Conf->Access = Access;
 261    Conf->Next = Configs;
 262    Configs = Conf;
 263
 264    // Create the worker to fetch the configuration
 265    Worker Work(Conf);
 266    if (Work.Start() == false)
 267       return 0;
 268
 269    return Conf;
 270 }
 271                                                                         /*}}}*/
 272 // Acquire::SetFds - Deal with readable FDs                             /*{{{*/
 273 // ---------------------------------------------------------------------
 274 /* Collect FDs that have activity monitors into the fd sets */
 275 void pkgAcquire::SetFds(int &Fd,fd_set *RSet,fd_set *WSet)
 276 {
 277    for (Worker *I = Workers; I != 0; I = I->NextAcquire)
 278    {
 279       if (I->InReady == true && I->InFd >= 0)
 280       {
 281          if (Fd < I->InFd)
 282             Fd = I->InFd;
 283          FD_SET(I->InFd,RSet);
 284       }
 285       if (I->OutReady == true && I->OutFd >= 0)
 286       {
 287          if (Fd < I->OutFd)
 288             Fd = I->OutFd;
 289          FD_SET(I->OutFd,WSet);
 290       }
 291    }
 292 }
 293                                                                         /*}}}*/
 294 // Acquire::RunFds - Deal with active FDs                               /*{{{*/
 295 // ---------------------------------------------------------------------
 296 /* Dispatch active FDs over to the proper workers. It is very important
 297    that a worker never be erased while this is running! The queue class
 298    should never erase a worker except during shutdown processing. */
 299 void pkgAcquire::RunFds(fd_set *RSet,fd_set *WSet)
 300 {
 301    for (Worker *I = Workers; I != 0; I = I->NextAcquire)
 302    {
 303       if (I->InFd >= 0 && FD_ISSET(I->InFd,RSet) != 0)
 304          I->InFdReady();
 305       if (I->OutFd >= 0 && FD_ISSET(I->OutFd,WSet) != 0)
 306          I->OutFdReady();
 307    }
 308 }
 309                                                                         /*}}}*/
 310 // Acquire::Run - Run the fetch sequence                                /*{{{*/
 311 // ---------------------------------------------------------------------
 312 /* This runs the queues. It manages a select loop for all of the
 313    Worker tasks. The workers interact with the queues and items to
 314    manage the actual fetch. */
 315 pkgAcquire::RunResult pkgAcquire::Run()
 316 {
 317    Running = true;
 318
 319    for (Queue *I = Queues; I != 0; I = I->Next)
 320       I->Startup();
 321
 322    if (Log != 0)
 323       Log->Start();
 324
 325    bool WasCancelled = false;
 326
 327    // Run till all things have been acquired
 328    struct timeval tv;
 329    tv.tv_sec = 0;
 330    tv.tv_usec = 500000;
 331    while (ToFetch > 0)
 332    {
 333       fd_set RFds;
 334       fd_set WFds;
 335       int Highest = 0;
 336       FD_ZERO(&RFds);
 337       FD_ZERO(&WFds);
 338       SetFds(Highest,&RFds,&WFds);
 339
 340       int Res;
 341       do
 342       {
 343          Res = select(Highest+1,&RFds,&WFds,0,&tv);
 344       }
 345       while (Res < 0 && errno == EINTR);
 346
 347       if (Res < 0)
 348       {
 349          _error->Errno("select","Select has failed");
 350          break;
 351       }
 352
 353       RunFds(&RFds,&WFds);
 354       if (_error->PendingError() == true)
 355          break;
 356
 357       // Timeout, notify the log class
 358       if (Res == 0 || (Log != 0 && Log->Update == true))
 359       {
 360          tv.tv_usec = 500000;
 361          for (Worker *I = Workers; I != 0; I = I->NextAcquire)
 362             I->Pulse();
 363          if (Log != 0 && Log->Pulse(this) == false)
 364          {
 365             WasCancelled = true;
 366             break;
 367          }
 368       }
 369    }
 370
 371    if (Log != 0)
 372       Log->Stop();
 373
 374    // Shut down the acquire bits
 375    Running = false;
 376    for (Queue *I = Queues; I != 0; I = I->Next)
 377       I->Shutdown(false);
 378
 379    // Shut down the items
 380    for (ItemIterator I = Items.begin(); I != Items.end(); I++)
 381       (*I)->Finished();
 382
 383    if (_error->PendingError())
 384       return Failed;
 385    if (WasCancelled)
 386       return Cancelled;
 387    return Continue;
 388 }
 389                                                                         /*}}}*/
 390 // Acquire::Bump - Called when an item is dequeued                      /*{{{*/
 391 // ---------------------------------------------------------------------
 392 /* This routine bumps idle queues in hopes that they will be able to fetch
 393    the dequeued item */
 394 void pkgAcquire::Bump()
 395 {
 396    for (Queue *I = Queues; I != 0; I = I->Next)
 397       I->Bump();
 398 }
 399                                                                         /*}}}*/
 400 // Acquire::WorkerStep - Step to the next worker                        /*{{{*/
 401 // ---------------------------------------------------------------------
 402 /* Not inlined to advoid including acquire-worker.h */
 403 pkgAcquire::Worker *pkgAcquire::WorkerStep(Worker *I)
 404 {
 405    return I->NextAcquire;
 406 };
 407                                                                         /*}}}*/
 408 // Acquire::Clean - Cleans a directory                                  /*{{{*/
 409 // ---------------------------------------------------------------------
 410 /* This is a bit simplistic, it looks at every file in the dir and sees
 411    if it is part of the download set. */
 412 bool pkgAcquire::Clean(string Dir)
 413 {
 414    DIR *D = opendir(Dir.c_str());
 415    if (D == 0)
 416       return _error->Errno("opendir",_("Unable to read %s"),Dir.c_str());
 417
 418    string StartDir = SafeGetCWD();
 419    if (chdir(Dir.c_str()) != 0)
 420    {
 421       closedir(D);
 422       return _error->Errno("chdir",_("Unable to change to %s"),Dir.c_str());
 423    }
 424
 425    for (struct dirent *Dir = readdir(D); Dir != 0; Dir = readdir(D))
 426    {
 427       // Skip some files..
 428       if (strcmp(Dir->d_name,"lock") == 0 ||
 429           strcmp(Dir->d_name,"partial") == 0 ||
 430           strcmp(Dir->d_name,".") == 0 ||
 431           strcmp(Dir->d_name,"..") == 0)
 432          continue;
 433
 434       // Look in the get list
 435       ItemCIterator I = Items.begin();
 436       for (; I != Items.end(); I++)
 437          if (flNotDir((*I)->DestFile) == Dir->d_name)
 438             break;
 439
 440       // Nothing found, nuke it
 441       if (I == Items.end())
 442          unlink(Dir->d_name);
 443    };
 444
 445    chdir(StartDir.c_str());
 446    closedir(D);
 447    return true;
 448 }
 449                                                                         /*}}}*/
 450 // Acquire::TotalNeeded - Number of bytes to fetch                      /*{{{*/
 451 // ---------------------------------------------------------------------
 452 /* This is the total number of bytes needed */
 453 double pkgAcquire::TotalNeeded()
 454 {
 455    double Total = 0;
 456    for (ItemCIterator I = ItemsBegin(); I != ItemsEnd(); I++)
 457       Total += (*I)->FileSize;
 458    return Total;
 459 }
 460                                                                         /*}}}*/
 461 // Acquire::FetchNeeded - Number of bytes needed to get                 /*{{{*/
 462 // ---------------------------------------------------------------------
 463 /* This is the number of bytes that is not local */
 464 double pkgAcquire::FetchNeeded()
 465 {
 466    double Total = 0;
 467    for (ItemCIterator I = ItemsBegin(); I != ItemsEnd(); I++)
 468       if ((*I)->Local == false)
 469          Total += (*I)->FileSize;
 470    return Total;
 471 }
 472                                                                         /*}}}*/
 473 // Acquire::PartialPresent - Number of partial bytes we already have    /*{{{*/
 474 // ---------------------------------------------------------------------
 475 /* This is the number of bytes that is not local */
 476 double pkgAcquire::PartialPresent()
 477 {
 478   double Total = 0;
 479    for (ItemCIterator I = ItemsBegin(); I != ItemsEnd(); I++)
 480       if ((*I)->Local == false)
 481          Total += (*I)->PartialSize;
 482    return Total;
 483 }
 484
 485 // Acquire::UriBegin - Start iterator for the uri list                  /*{{{*/
 486 // ---------------------------------------------------------------------
 487 /* */
 488 pkgAcquire::UriIterator pkgAcquire::UriBegin()
 489 {
 490    return UriIterator(Queues);
 491 }
 492                                                                         /*}}}*/
 493 // Acquire::UriEnd - End iterator for the uri list                      /*{{{*/
 494 // ---------------------------------------------------------------------
 495 /* */
 496 pkgAcquire::UriIterator pkgAcquire::UriEnd()
 497 {
 498    return UriIterator(0);
 499 }
 500                                                                         /*}}}*/
 501
 502 // Acquire::MethodConfig::MethodConfig - Constructor                    /*{{{*/
 503 // ---------------------------------------------------------------------
 504 /* */
 505 pkgAcquire::MethodConfig::MethodConfig()
 506 {
 507    SingleInstance = false;
 508    Pipeline = false;
 509    SendConfig = false;
 510    LocalOnly = false;
 511    Removable = false;
 512    Next = 0;
 513 }
 514                                                                         /*}}}*/
 515
 516 // Queue::Queue - Constructor                                           /*{{{*/
 517 // ---------------------------------------------------------------------
 518 /* */
 519 pkgAcquire::Queue::Queue(string Name,pkgAcquire *Owner) : Name(Name),
 520             Owner(Owner)
 521 {
 522    Items = 0;
 523    Next = 0;
 524    Workers = 0;
 525    MaxPipeDepth = 1;
 526    PipeDepth = 0;
 527 }
 528                                                                         /*}}}*/
 529 // Queue::~Queue - Destructor                                           /*{{{*/
 530 // ---------------------------------------------------------------------
 531 /* */
 532 pkgAcquire::Queue::~Queue()
 533 {
 534    Shutdown(true);
 535
 536    while (Items != 0)
 537    {
 538       QItem *Jnk = Items;
 539       Items = Items->Next;
 540       delete Jnk;
 541    }
 542 }
 543                                                                         /*}}}*/
 544 // Queue::Enqueue - Queue an item to the queue                          /*{{{*/
 545 // ---------------------------------------------------------------------
 546 /* */
 547 void pkgAcquire::Queue::Enqueue(ItemDesc &Item)
 548 {
 549    QItem **I = &Items;
 550    for (; *I != 0; I = &(*I)->Next);
 551
 552    // Create a new item
 553    QItem *Itm = new QItem;
 554    *Itm = Item;
 555    Itm->Next = 0;
 556    *I = Itm;
 557
 558    Item.Owner->QueueCounter++;
 559    if (Items->Next == 0)
 560       Cycle();
 561 }
 562                                                                         /*}}}*/
 563 // Queue::Dequeue - Remove an item from the queue                       /*{{{*/
 564 // ---------------------------------------------------------------------
 565 /* We return true if we hit something */
 566 bool pkgAcquire::Queue::Dequeue(Item *Owner)
 567 {
 568    if (Owner->Status == pkgAcquire::Item::StatFetching)
 569       return _error->Error("Tried to dequeue a fetching object");
 570
 571    bool Res = false;
 572
 573    QItem **I = &Items;
 574    for (; *I != 0;)
 575    {
 576       if ((*I)->Owner == Owner)
 577       {
 578          QItem *Jnk= *I;
 579          *I = (*I)->Next;
 580          Owner->QueueCounter--;
 581          delete Jnk;
 582          Res = true;
 583       }
 584       else
 585          I = &(*I)->Next;
 586    }
 587
 588    return Res;
 589 }
 590                                                                         /*}}}*/
 591 // Queue::Startup - Start the worker processes                          /*{{{*/
 592 // ---------------------------------------------------------------------
 593 /* It is possible for this to be called with a pre-existing set of
 594    workers. */
 595 bool pkgAcquire::Queue::Startup()
 596 {
 597    if (Workers == 0)
 598    {
 599       URI U(Name);
 600       pkgAcquire::MethodConfig *Cnf = Owner->GetConfig(U.Access);
 601       if (Cnf == 0)
 602          return false;
 603
 604       Workers = new Worker(this,Cnf,Owner->Log);
 605       Owner->Add(Workers);
 606       if (Workers->Start() == false)
 607          return false;
 608
 609       /* When pipelining we commit 10 items. This needs to change when we
 610          added other source retry to have cycle maintain a pipeline depth
 611          on its own. */
 612       if (Cnf->Pipeline == true)
 613          MaxPipeDepth = 10;
 614       else
 615          MaxPipeDepth = 1;
 616    }
 617
 618    return Cycle();
 619 }
 620                                                                         /*}}}*/
 621 // Queue::Shutdown - Shutdown the worker processes                      /*{{{*/
 622 // ---------------------------------------------------------------------
 623 /* If final is true then all workers are eliminated, otherwise only workers
 624    that do not need cleanup are removed */
 625 bool pkgAcquire::Queue::Shutdown(bool Final)
 626 {
 627    // Delete all of the workers
 628    pkgAcquire::Worker **Cur = &Workers;
 629    while (*Cur != 0)
 630    {
 631       pkgAcquire::Worker *Jnk = *Cur;
 632       if (Final == true || Jnk->GetConf()->NeedsCleanup == false)
 633       {
 634          *Cur = Jnk->NextQueue;
 635          Owner->Remove(Jnk);
 636          delete Jnk;
 637       }
 638       else
 639          Cur = &(*Cur)->NextQueue;
 640    }
 641
 642    return true;
 643 }
 644                                                                         /*}}}*/
 645 // Queue::FindItem - Find a URI in the item list                        /*{{{*/
 646 // ---------------------------------------------------------------------
 647 /* */
 648 pkgAcquire::Queue::QItem *pkgAcquire::Queue::FindItem(string URI,pkgAcquire::Worker *Owner)
 649 {
 650    for (QItem *I = Items; I != 0; I = I->Next)
 651       if (I->URI == URI && I->Worker == Owner)
 652          return I;
 653    return 0;
 654 }
 655                                                                         /*}}}*/
 656 // Queue::ItemDone - Item has been completed                            /*{{{*/
 657 // ---------------------------------------------------------------------
 658 /* The worker signals this which causes the item to be removed from the
 659    queue. If this is the last queue instance then it is removed from the
 660    main queue too.*/
 661 bool pkgAcquire::Queue::ItemDone(QItem *Itm)
 662 {
 663    PipeDepth--;
 664    if (Itm->Owner->Status == pkgAcquire::Item::StatFetching)
 665       Itm->Owner->Status = pkgAcquire::Item::StatDone;
 666
 667    if (Itm->Owner->QueueCounter <= 1)
 668       Owner->Dequeue(Itm->Owner);
 669    else
 670    {
 671       Dequeue(Itm->Owner);
 672       Owner->Bump();
 673    }
 674
 675    return Cycle();
 676 }
 677                                                                         /*}}}*/
 678 // Queue::Cycle - Queue new items into the method                       /*{{{*/
 679 // ---------------------------------------------------------------------
 680 /* This locates a new idle item and sends it to the worker. If pipelining
 681    is enabled then it keeps the pipe full. */
 682 bool pkgAcquire::Queue::Cycle()
 683 {
 684    if (Items == 0 || Workers == 0)
 685       return true;
 686
 687    if (PipeDepth < 0)
 688       return _error->Error("Pipedepth failure");
 689
 690    // Look for a queable item
 691    QItem *I = Items;
 692    while (PipeDepth < (signed)MaxPipeDepth)
 693    {
 694       for (; I != 0; I = I->Next)
 695          if (I->Owner->Status == pkgAcquire::Item::StatIdle)
 696             break;
 697
 698       // Nothing to do, queue is idle.
 699       if (I == 0)
 700          return true;
 701
 702       I->Worker = Workers;
 703       I->Owner->Status = pkgAcquire::Item::StatFetching;
 704       PipeDepth++;
 705       if (Workers->QueueItem(I) == false)
 706          return false;
 707    }
 708
 709    return true;
 710 }
 711                                                                         /*}}}*/
 712 // Queue::Bump - Fetch any pending objects if we are idle               /*{{{*/
 713 // ---------------------------------------------------------------------
 714 /* This is called when an item in multiple queues is dequeued */
 715 void pkgAcquire::Queue::Bump()
 716 {
 717    Cycle();
 718 }
 719                                                                         /*}}}*/
 720
 721 // AcquireStatus::pkgAcquireStatus - Constructor                        /*{{{*/
 722 // ---------------------------------------------------------------------
 723 /* */
 724 pkgAcquireStatus::pkgAcquireStatus() : Update(true), MorePulses(false)
 725 {
 726    Start();
 727 }
 728                                                                         /*}}}*/
 729 // AcquireStatus::Pulse - Called periodically                           /*{{{*/
 730 // ---------------------------------------------------------------------
 731 /* This computes some internal state variables for the derived classes to
 732    use. It generates the current downloaded bytes and total bytes to download
 733    as well as the current CPS estimate. */
 734 bool pkgAcquireStatus::Pulse(pkgAcquire *Owner)
 735 {
 736    TotalBytes = 0;
 737    CurrentBytes = 0;
 738    TotalItems = 0;
 739    CurrentItems = 0;
 740
 741    // Compute the total number of bytes to fetch
 742    unsigned int Unknown = 0;
 743    unsigned int Count = 0;
 744    for (pkgAcquire::ItemCIterator I = Owner->ItemsBegin(); I != Owner->ItemsEnd();
 745         I++, Count++)
 746    {
 747       TotalItems++;
 748       if ((*I)->Status == pkgAcquire::Item::StatDone)
 749          CurrentItems++;
 750
 751       // Totally ignore local items
 752       if ((*I)->Local == true)
 753          continue;
 754
 755       TotalBytes += (*I)->FileSize;
 756       if ((*I)->Complete == true)
 757          CurrentBytes += (*I)->FileSize;
 758       if ((*I)->FileSize == 0 && (*I)->Complete == false)
 759          Unknown++;
 760    }
 761
 762    // Compute the current completion
 763    unsigned long ResumeSize = 0;
 764    for (pkgAcquire::Worker *I = Owner->WorkersBegin(); I != 0;
 765         I = Owner->WorkerStep(I))
 766       if (I->CurrentItem != 0 && I->CurrentItem->Owner->Complete == false)
 767       {
 768          CurrentBytes += I->CurrentSize;
 769          ResumeSize += I->ResumePoint;
 770
 771          // Files with unknown size always have 100% completion
 772          if (I->CurrentItem->Owner->FileSize == 0 &&
 773              I->CurrentItem->Owner->Complete == false)
 774             TotalBytes += I->CurrentSize;
 775       }
 776
 777    // Normalize the figures and account for unknown size downloads
 778    if (TotalBytes <= 0)
 779       TotalBytes = 1;
 780    if (Unknown == Count)
 781       TotalBytes = Unknown;
 782
 783    // Wha?! Is not supposed to happen.
 784    if (CurrentBytes > TotalBytes)
 785       CurrentBytes = TotalBytes;
 786
 787    // Compute the CPS
 788    struct timeval NewTime;
 789    gettimeofday(&NewTime,0);
 790    if (NewTime.tv_sec - Time.tv_sec == 6 && NewTime.tv_usec > Time.tv_usec ||
 791        NewTime.tv_sec - Time.tv_sec > 6)
 792    {
 793       double Delta = NewTime.tv_sec - Time.tv_sec +
 794                      (NewTime.tv_usec - Time.tv_usec)/1000000.0;
 795
 796       // Compute the CPS value
 797       if (Delta < 0.01)
 798          CurrentCPS = 0;
 799       else
 800          CurrentCPS = ((CurrentBytes - ResumeSize) - LastBytes)/Delta;
 801       LastBytes = CurrentBytes - ResumeSize;
 802       ElapsedTime = (unsigned long)Delta;
 803       Time = NewTime;
 804    }
 805
 806    return true;
 807 }
 808                                                                         /*}}}*/
 809 // AcquireStatus::Start - Called when the download is started           /*{{{*/
 810 // ---------------------------------------------------------------------
 811 /* We just reset the counters */
 812 void pkgAcquireStatus::Start()
 813 {
 814    gettimeofday(&Time,0);
 815    gettimeofday(&StartTime,0);
 816    LastBytes = 0;
 817    CurrentCPS = 0;
 818    CurrentBytes = 0;
 819    TotalBytes = 0;
 820    FetchedBytes = 0;
 821    ElapsedTime = 0;
 822    TotalItems = 0;
 823    CurrentItems = 0;
 824 }
 825                                                                         /*}}}*/
 826 // AcquireStatus::Stop - Finished downloading                           /*{{{*/
 827 // ---------------------------------------------------------------------
 828 /* This accurately computes the elapsed time and the total overall CPS. */
 829 void pkgAcquireStatus::Stop()
 830 {
 831    // Compute the CPS and elapsed time
 832    struct timeval NewTime;
 833    gettimeofday(&NewTime,0);
 834
 835    double Delta = NewTime.tv_sec - StartTime.tv_sec +
 836                   (NewTime.tv_usec - StartTime.tv_usec)/1000000.0;
 837
 838    // Compute the CPS value
 839    if (Delta < 0.01)
 840       CurrentCPS = 0;
 841    else
 842       CurrentCPS = FetchedBytes/Delta;
 843    LastBytes = CurrentBytes;
 844    ElapsedTime = (unsigned int)Delta;
 845 }
 846                                                                         /*}}}*/
 847 // AcquireStatus::Fetched - Called when a byte set has been fetched     /*{{{*/
 848 // ---------------------------------------------------------------------
 849 /* This is used to get accurate final transfer rate reporting. */
 850 void pkgAcquireStatus::Fetched(unsigned long Size,unsigned long Resume)
 851 {
 852    FetchedBytes += Size - Resume;
 853 }
 854                                                                         /*}}}*/