apt-pkg/acquire.cc

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