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