apt-pkg/acquire.cc

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