apt-pkg/acquire.cc

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