apt-pkg/acquire.cc

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