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