apt-pkg/acquire.cc

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