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    string name(U.Access + ':' + U.Host);
 241
 242    int parallel(_config->FindI("Acquire::"+U.Access+"::MaxParallel",8));
 243    if (parallel <= 0)
 244       return name;
 245
 246    typedef map<string, int> indexmap;
 247    static indexmap indices;
 248
 249    pair<indexmap::iterator, bool> cache(indices.insert(indexmap::value_type(name, -1)));
 250    if (cache.second || cache.first->second == -1) {
 251       int &index(indices[U.Access]);
 252       if (index >= parallel)
 253          index = 0;
 254       cache.first->second = index++;
 255    }
 256
 257    ostringstream value;
 258    value << U.Access << "::" << cache.first->second;
 259    return value.str();
 260 }
 261                                                                         /*}}}*/
 262 // Acquire::GetConfig - Fetch the configuration information             /*{{{*/
 263 // ---------------------------------------------------------------------
 264 /* This locates the configuration structure for an access method. If
 265    a config structure cannot be found a Worker will be created to
 266    retrieve it */
 267 pkgAcquire::MethodConfig *pkgAcquire::GetConfig(string Access)
 268 {
 269    // Search for an existing config
 270    MethodConfig *Conf;
 271    for (Conf = Configs; Conf != 0; Conf = Conf->Next)
 272       if (Conf->Access == Access)
 273          return Conf;
 274
 275    // Create the new config class
 276    Conf = new MethodConfig;
 277    Conf->Access = Access;
 278    Conf->Next = Configs;
 279    Configs = Conf;
 280
 281    // Create the worker to fetch the configuration
 282    Worker Work(Conf);
 283    if (Work.Start() == false)
 284       return 0;
 285
 286    /* if a method uses DownloadLimit, we switch to SingleInstance mode */
 287    if(_config->FindI("Acquire::"+Access+"::Dl-Limit",0) > 0)
 288       Conf->SingleInstance = true;
 289
 290    return Conf;
 291 }
 292                                                                         /*}}}*/
 293 // Acquire::SetFds - Deal with readable FDs                             /*{{{*/
 294 // ---------------------------------------------------------------------
 295 /* Collect FDs that have activity monitors into the fd sets */
 296 void pkgAcquire::SetFds(int &Fd,fd_set *RSet,fd_set *WSet)
 297 {
 298    for (Worker *I = Workers; I != 0; I = I->NextAcquire)
 299    {
 300       if (I->InReady == true && I->InFd >= 0)
 301       {
 302          if (Fd < I->InFd)
 303             Fd = I->InFd;
 304          FD_SET(I->InFd,RSet);
 305       }
 306       if (I->OutReady == true && I->OutFd >= 0)
 307       {
 308          if (Fd < I->OutFd)
 309             Fd = I->OutFd;
 310          FD_SET(I->OutFd,WSet);
 311       }
 312    }
 313 }
 314                                                                         /*}}}*/
 315 // Acquire::RunFds - Deal with active FDs                               /*{{{*/
 316 // ---------------------------------------------------------------------
 317 /* Dispatch active FDs over to the proper workers. It is very important
 318    that a worker never be erased while this is running! The queue class
 319    should never erase a worker except during shutdown processing. */
 320 void pkgAcquire::RunFds(fd_set *RSet,fd_set *WSet)
 321 {
 322    for (Worker *I = Workers; I != 0; I = I->NextAcquire)
 323    {
 324       if (I->InFd >= 0 && FD_ISSET(I->InFd,RSet) != 0)
 325          I->InFdReady();
 326       if (I->OutFd >= 0 && FD_ISSET(I->OutFd,WSet) != 0)
 327          I->OutFdReady();
 328    }
 329 }
 330                                                                         /*}}}*/
 331 // Acquire::Run - Run the fetch sequence                                /*{{{*/
 332 // ---------------------------------------------------------------------
 333 /* This runs the queues. It manages a select loop for all of the
 334    Worker tasks. The workers interact with the queues and items to
 335    manage the actual fetch. */
 336 pkgAcquire::RunResult pkgAcquire::Run(int PulseIntervall)
 337 {
 338    Running = true;
 339
 340    for (Queue *I = Queues; I != 0; I = I->Next)
 341       I->Startup();
 342
 343    if (Log != 0)
 344       Log->Start();
 345
 346    bool WasCancelled = false;
 347
 348    // Run till all things have been acquired
 349    struct timeval tv;
 350    tv.tv_sec = 0;
 351    tv.tv_usec = PulseIntervall;
 352    while (ToFetch > 0)
 353    {
 354       fd_set RFds;
 355       fd_set WFds;
 356       int Highest = 0;
 357       FD_ZERO(&RFds);
 358       FD_ZERO(&WFds);
 359       SetFds(Highest,&RFds,&WFds);
 360
 361       int Res;
 362       do
 363       {
 364          Res = select(Highest+1,&RFds,&WFds,0,&tv);
 365       }
 366       while (Res < 0 && errno == EINTR);
 367
 368       if (Res < 0)
 369       {
 370          _error->Errno("select","Select has failed");
 371          break;
 372       }
 373
 374       RunFds(&RFds,&WFds);
 375       if (_error->PendingError() == true)
 376          break;
 377
 378       // Timeout, notify the log class
 379       if (Res == 0 || (Log != 0 && Log->Update == true))
 380       {
 381          tv.tv_usec = PulseIntervall;
 382          for (Worker *I = Workers; I != 0; I = I->NextAcquire)
 383             I->Pulse();
 384          if (Log != 0 && Log->Pulse(this) == false)
 385          {
 386             WasCancelled = true;
 387             break;
 388          }
 389       }
 390    }
 391
 392    if (Log != 0)
 393       Log->Stop();
 394
 395    // Shut down the acquire bits
 396    Running = false;
 397    for (Queue *I = Queues; I != 0; I = I->Next)
 398       I->Shutdown(false);
 399
 400    // Shut down the items
 401    for (ItemIterator I = Items.begin(); I != Items.end(); I++)
 402       (*I)->Finished();
 403
 404    if (_error->PendingError())
 405       return Failed;
 406    if (WasCancelled)
 407       return Cancelled;
 408    return Continue;
 409 }
 410                                                                         /*}}}*/
 411 // Acquire::Bump - Called when an item is dequeued                      /*{{{*/
 412 // ---------------------------------------------------------------------
 413 /* This routine bumps idle queues in hopes that they will be able to fetch
 414    the dequeued item */
 415 void pkgAcquire::Bump()
 416 {
 417    for (Queue *I = Queues; I != 0; I = I->Next)
 418       I->Bump();
 419 }
 420                                                                         /*}}}*/
 421 // Acquire::WorkerStep - Step to the next worker                        /*{{{*/
 422 // ---------------------------------------------------------------------
 423 /* Not inlined to advoid including acquire-worker.h */
 424 pkgAcquire::Worker *pkgAcquire::WorkerStep(Worker *I)
 425 {
 426    return I->NextAcquire;
 427 };
 428                                                                         /*}}}*/
 429 // Acquire::Clean - Cleans a directory                                  /*{{{*/
 430 // ---------------------------------------------------------------------
 431 /* This is a bit simplistic, it looks at every file in the dir and sees
 432    if it is part of the download set. */
 433 bool pkgAcquire::Clean(string Dir)
 434 {
 435    DIR *D = opendir(Dir.c_str());
 436    if (D == 0)
 437       return _error->Errno("opendir",_("Unable to read %s"),Dir.c_str());
 438
 439    string StartDir = SafeGetCWD();
 440    if (chdir(Dir.c_str()) != 0)
 441    {
 442       closedir(D);
 443       return _error->Errno("chdir",_("Unable to change to %s"),Dir.c_str());
 444    }
 445
 446    for (struct dirent *Dir = readdir(D); Dir != 0; Dir = readdir(D))
 447    {
 448       // Skip some files..
 449       if (strcmp(Dir->d_name,"lock") == 0 ||
 450           strcmp(Dir->d_name,"partial") == 0 ||
 451           strcmp(Dir->d_name,".") == 0 ||
 452           strcmp(Dir->d_name,"..") == 0)
 453          continue;
 454
 455       // Look in the get list
 456       ItemCIterator I = Items.begin();
 457       for (; I != Items.end(); I++)
 458          if (flNotDir((*I)->DestFile) == Dir->d_name)
 459             break;
 460
 461       // Nothing found, nuke it
 462       if (I == Items.end())
 463          unlink(Dir->d_name);
 464    };
 465
 466    closedir(D);
 467    if (chdir(StartDir.c_str()) != 0)
 468       return _error->Errno("chdir",_("Unable to change to %s"),StartDir.c_str());
 469    return true;
 470 }
 471                                                                         /*}}}*/
 472 // Acquire::TotalNeeded - Number of bytes to fetch                      /*{{{*/
 473 // ---------------------------------------------------------------------
 474 /* This is the total number of bytes needed */
 475 double pkgAcquire::TotalNeeded()
 476 {
 477    double Total = 0;
 478    for (ItemCIterator I = ItemsBegin(); I != ItemsEnd(); I++)
 479       Total += (*I)->FileSize;
 480    return Total;
 481 }
 482                                                                         /*}}}*/
 483 // Acquire::FetchNeeded - Number of bytes needed to get                 /*{{{*/
 484 // ---------------------------------------------------------------------
 485 /* This is the number of bytes that is not local */
 486 double pkgAcquire::FetchNeeded()
 487 {
 488    double Total = 0;
 489    for (ItemCIterator I = ItemsBegin(); I != ItemsEnd(); I++)
 490       if ((*I)->Local == false)
 491          Total += (*I)->FileSize;
 492    return Total;
 493 }
 494                                                                         /*}}}*/
 495 // Acquire::PartialPresent - Number of partial bytes we already have    /*{{{*/
 496 // ---------------------------------------------------------------------
 497 /* This is the number of bytes that is not local */
 498 double pkgAcquire::PartialPresent()
 499 {
 500   double Total = 0;
 501    for (ItemCIterator I = ItemsBegin(); I != ItemsEnd(); I++)
 502       if ((*I)->Local == false)
 503          Total += (*I)->PartialSize;
 504    return Total;
 505 }
 506                                                                         /*}}}*/
 507 // Acquire::UriBegin - Start iterator for the uri list                  /*{{{*/
 508 // ---------------------------------------------------------------------
 509 /* */
 510 pkgAcquire::UriIterator pkgAcquire::UriBegin()
 511 {
 512    return UriIterator(Queues);
 513 }
 514                                                                         /*}}}*/
 515 // Acquire::UriEnd - End iterator for the uri list                      /*{{{*/
 516 // ---------------------------------------------------------------------
 517 /* */
 518 pkgAcquire::UriIterator pkgAcquire::UriEnd()
 519 {
 520    return UriIterator(0);
 521 }
 522                                                                         /*}}}*/
 523 // Acquire::MethodConfig::MethodConfig - Constructor                    /*{{{*/
 524 // ---------------------------------------------------------------------
 525 /* */
 526 pkgAcquire::MethodConfig::MethodConfig()
 527 {
 528    SingleInstance = false;
 529    Pipeline = false;
 530    SendConfig = false;
 531    LocalOnly = false;
 532    Removable = false;
 533    Next = 0;
 534 }
 535                                                                         /*}}}*/
 536 // Queue::Queue - Constructor                                           /*{{{*/
 537 // ---------------------------------------------------------------------
 538 /* */
 539 pkgAcquire::Queue::Queue(string Name,pkgAcquire *Owner) : Name(Name),
 540             Owner(Owner)
 541 {
 542    Items = 0;
 543    Next = 0;
 544    Workers = 0;
 545    MaxPipeDepth = 1;
 546    PipeDepth = 0;
 547 }
 548                                                                         /*}}}*/
 549 // Queue::~Queue - Destructor                                           /*{{{*/
 550 // ---------------------------------------------------------------------
 551 /* */
 552 pkgAcquire::Queue::~Queue()
 553 {
 554    Shutdown(true);
 555
 556    while (Items != 0)
 557    {
 558       QItem *Jnk = Items;
 559       Items = Items->Next;
 560       delete Jnk;
 561    }
 562 }
 563                                                                         /*}}}*/
 564 // Queue::Enqueue - Queue an item to the queue                          /*{{{*/
 565 // ---------------------------------------------------------------------
 566 /* */
 567 bool pkgAcquire::Queue::Enqueue(ItemDesc &Item)
 568 {
 569    QItem **I = &Items;
 570    // move to the end of the queue and check for duplicates here
 571    for (; *I != 0; I = &(*I)->Next)
 572       if (Item.URI == (*I)->URI)
 573       {
 574          Item.Owner->Status = Item::StatDone;
 575          return false;
 576       }
 577
 578    // Create a new item
 579    QItem *Itm = new QItem;
 580    *Itm = Item;
 581    Itm->Next = 0;
 582    *I = Itm;
 583
 584    Item.Owner->QueueCounter++;
 585    if (Items->Next == 0)
 586       Cycle();
 587    return true;
 588 }
 589                                                                         /*}}}*/
 590 // Queue::Dequeue - Remove an item from the queue                       /*{{{*/
 591 // ---------------------------------------------------------------------
 592 /* We return true if we hit something */
 593 bool pkgAcquire::Queue::Dequeue(Item *Owner)
 594 {
 595    if (Owner->Status == pkgAcquire::Item::StatFetching)
 596       return _error->Error("Tried to dequeue a fetching object");
 597
 598    bool Res = false;
 599
 600    QItem **I = &Items;
 601    for (; *I != 0;)
 602    {
 603       if ((*I)->Owner == Owner)
 604       {
 605          QItem *Jnk= *I;
 606          *I = (*I)->Next;
 607          Owner->QueueCounter--;
 608          delete Jnk;
 609          Res = true;
 610       }
 611       else
 612          I = &(*I)->Next;
 613    }
 614
 615    return Res;
 616 }
 617                                                                         /*}}}*/
 618 // Queue::Startup - Start the worker processes                          /*{{{*/
 619 // ---------------------------------------------------------------------
 620 /* It is possible for this to be called with a pre-existing set of
 621    workers. */
 622 bool pkgAcquire::Queue::Startup()
 623 {
 624    if (Workers == 0)
 625    {
 626       URI U(Name);
 627       pkgAcquire::MethodConfig *Cnf = Owner->GetConfig(U.Access);
 628       if (Cnf == 0)
 629          return false;
 630
 631       Workers = new Worker(this,Cnf,Owner->Log);
 632       Owner->Add(Workers);
 633       if (Workers->Start() == false)
 634          return false;
 635
 636       /* When pipelining we commit 10 items. This needs to change when we
 637          added other source retry to have cycle maintain a pipeline depth
 638          on its own. */
 639       if (Cnf->Pipeline == true)
 640          MaxPipeDepth = _config->FindI("Acquire::Max-Pipeline-Depth",10);
 641       else
 642          MaxPipeDepth = 1;
 643    }
 644
 645    return Cycle();
 646 }
 647                                                                         /*}}}*/
 648 // Queue::Shutdown - Shutdown the worker processes                      /*{{{*/
 649 // ---------------------------------------------------------------------
 650 /* If final is true then all workers are eliminated, otherwise only workers
 651    that do not need cleanup are removed */
 652 bool pkgAcquire::Queue::Shutdown(bool Final)
 653 {
 654    // Delete all of the workers
 655    pkgAcquire::Worker **Cur = &Workers;
 656    while (*Cur != 0)
 657    {
 658       pkgAcquire::Worker *Jnk = *Cur;
 659       if (Final == true || Jnk->GetConf()->NeedsCleanup == false)
 660       {
 661          *Cur = Jnk->NextQueue;
 662          Owner->Remove(Jnk);
 663          delete Jnk;
 664       }
 665       else
 666          Cur = &(*Cur)->NextQueue;
 667    }
 668
 669    return true;
 670 }
 671                                                                         /*}}}*/
 672 // Queue::FindItem - Find a URI in the item list                        /*{{{*/
 673 // ---------------------------------------------------------------------
 674 /* */
 675 pkgAcquire::Queue::QItem *pkgAcquire::Queue::FindItem(string URI,pkgAcquire::Worker *Owner)
 676 {
 677    for (QItem *I = Items; I != 0; I = I->Next)
 678       if (I->URI == URI && I->Worker == Owner)
 679          return I;
 680    return 0;
 681 }
 682                                                                         /*}}}*/
 683 // Queue::ItemDone - Item has been completed                            /*{{{*/
 684 // ---------------------------------------------------------------------
 685 /* The worker signals this which causes the item to be removed from the
 686    queue. If this is the last queue instance then it is removed from the
 687    main queue too.*/
 688 bool pkgAcquire::Queue::ItemDone(QItem *Itm)
 689 {
 690    PipeDepth--;
 691    if (Itm->Owner->Status == pkgAcquire::Item::StatFetching)
 692       Itm->Owner->Status = pkgAcquire::Item::StatDone;
 693
 694    if (Itm->Owner->QueueCounter <= 1)
 695       Owner->Dequeue(Itm->Owner);
 696    else
 697    {
 698       Dequeue(Itm->Owner);
 699       Owner->Bump();
 700    }
 701
 702    return Cycle();
 703 }
 704                                                                         /*}}}*/
 705 // Queue::Cycle - Queue new items into the method                       /*{{{*/
 706 // ---------------------------------------------------------------------
 707 /* This locates a new idle item and sends it to the worker. If pipelining
 708    is enabled then it keeps the pipe full. */
 709 bool pkgAcquire::Queue::Cycle()
 710 {
 711    if (Items == 0 || Workers == 0)
 712       return true;
 713
 714    if (PipeDepth < 0)
 715       return _error->Error("Pipedepth failure");
 716
 717    // Look for a queable item
 718    QItem *I = Items;
 719    while (PipeDepth < (signed)MaxPipeDepth)
 720    {
 721       for (; I != 0; I = I->Next)
 722          if (I->Owner->Status == pkgAcquire::Item::StatIdle)
 723             break;
 724
 725       // Nothing to do, queue is idle.
 726       if (I == 0)
 727          return true;
 728
 729       I->Worker = Workers;
 730       I->Owner->Status = pkgAcquire::Item::StatFetching;
 731       PipeDepth++;
 732       if (Workers->QueueItem(I) == false)
 733          return false;
 734    }
 735
 736    return true;
 737 }
 738                                                                         /*}}}*/
 739 // Queue::Bump - Fetch any pending objects if we are idle               /*{{{*/
 740 // ---------------------------------------------------------------------
 741 /* This is called when an item in multiple queues is dequeued */
 742 void pkgAcquire::Queue::Bump()
 743 {
 744    Cycle();
 745 }
 746                                                                         /*}}}*/
 747 // AcquireStatus::pkgAcquireStatus - Constructor                        /*{{{*/
 748 // ---------------------------------------------------------------------
 749 /* */
 750 pkgAcquireStatus::pkgAcquireStatus() : Update(true), MorePulses(false)
 751 {
 752    Start();
 753 }
 754                                                                         /*}}}*/
 755 // AcquireStatus::Pulse - Called periodically                           /*{{{*/
 756 // ---------------------------------------------------------------------
 757 /* This computes some internal state variables for the derived classes to
 758    use. It generates the current downloaded bytes and total bytes to download
 759    as well as the current CPS estimate. */
 760 bool pkgAcquireStatus::Pulse(pkgAcquire *Owner)
 761 {
 762    TotalBytes = 0;
 763    CurrentBytes = 0;
 764    TotalItems = 0;
 765    CurrentItems = 0;
 766
 767    // Compute the total number of bytes to fetch
 768    unsigned int Unknown = 0;
 769    unsigned int Count = 0;
 770    for (pkgAcquire::ItemCIterator I = Owner->ItemsBegin(); I != Owner->ItemsEnd();
 771         I++, Count++)
 772    {
 773       TotalItems++;
 774       if ((*I)->Status == pkgAcquire::Item::StatDone)
 775          CurrentItems++;
 776
 777       // Totally ignore local items
 778       if ((*I)->Local == true)
 779          continue;
 780
 781       TotalBytes += (*I)->FileSize;
 782       if ((*I)->Complete == true)
 783          CurrentBytes += (*I)->FileSize;
 784       if ((*I)->FileSize == 0 && (*I)->Complete == false)
 785          Unknown++;
 786    }
 787
 788    // Compute the current completion
 789    unsigned long ResumeSize = 0;
 790    for (pkgAcquire::Worker *I = Owner->WorkersBegin(); I != 0;
 791         I = Owner->WorkerStep(I))
 792       if (I->CurrentItem != 0 && I->CurrentItem->Owner->Complete == false)
 793       {
 794          CurrentBytes += I->CurrentSize;
 795          ResumeSize += I->ResumePoint;
 796
 797          // Files with unknown size always have 100% completion
 798          if (I->CurrentItem->Owner->FileSize == 0 &&
 799              I->CurrentItem->Owner->Complete == false)
 800             TotalBytes += I->CurrentSize;
 801       }
 802
 803    // Normalize the figures and account for unknown size downloads
 804    if (TotalBytes <= 0)
 805       TotalBytes = 1;
 806    if (Unknown == Count)
 807       TotalBytes = Unknown;
 808
 809    // Wha?! Is not supposed to happen.
 810    if (CurrentBytes > TotalBytes)
 811       CurrentBytes = TotalBytes;
 812
 813    // Compute the CPS
 814    struct timeval NewTime;
 815    gettimeofday(&NewTime,0);
 816    if ((NewTime.tv_sec - Time.tv_sec == 6 && NewTime.tv_usec > Time.tv_usec) ||
 817        NewTime.tv_sec - Time.tv_sec > 6)
 818    {
 819       double Delta = NewTime.tv_sec - Time.tv_sec +
 820                      (NewTime.tv_usec - Time.tv_usec)/1000000.0;
 821
 822       // Compute the CPS value
 823       if (Delta < 0.01)
 824          CurrentCPS = 0;
 825       else
 826          CurrentCPS = ((CurrentBytes - ResumeSize) - LastBytes)/Delta;
 827       LastBytes = CurrentBytes - ResumeSize;
 828       ElapsedTime = (unsigned long)Delta;
 829       Time = NewTime;
 830    }
 831
 832    int fd = _config->FindI("APT::Status-Fd",-1);
 833    if(fd > 0)
 834    {
 835       ostringstream status;
 836
 837       char msg[200];
 838       long i = CurrentItems < TotalItems ? CurrentItems + 1 : CurrentItems;
 839       unsigned long ETA =
 840          (unsigned long)((TotalBytes - CurrentBytes) / CurrentCPS);
 841
 842       // only show the ETA if it makes sense
 843       if (ETA > 0 && ETA < 172800 /* two days */ )
 844          snprintf(msg,sizeof(msg), _("Retrieving file %li of %li (%s remaining)"), i, TotalItems, TimeToStr(ETA).c_str());
 845       else
 846          snprintf(msg,sizeof(msg), _("Retrieving file %li of %li"), i, TotalItems);
 847
 848
 849
 850       // build the status str
 851       status << "dlstatus:" << i
 852              << ":"  << (CurrentBytes/float(TotalBytes)*100.0)
 853              << ":" << msg
 854              << endl;
 855       write(fd, status.str().c_str(), status.str().size());
 856    }
 857
 858    return true;
 859 }
 860                                                                         /*}}}*/
 861 // AcquireStatus::Start - Called when the download is started           /*{{{*/
 862 // ---------------------------------------------------------------------
 863 /* We just reset the counters */
 864 void pkgAcquireStatus::Start()
 865 {
 866    gettimeofday(&Time,0);
 867    gettimeofday(&StartTime,0);
 868    LastBytes = 0;
 869    CurrentCPS = 0;
 870    CurrentBytes = 0;
 871    TotalBytes = 0;
 872    FetchedBytes = 0;
 873    ElapsedTime = 0;
 874    TotalItems = 0;
 875    CurrentItems = 0;
 876 }
 877                                                                         /*}}}*/
 878 // AcquireStatus::Stop - Finished downloading                           /*{{{*/
 879 // ---------------------------------------------------------------------
 880 /* This accurately computes the elapsed time and the total overall CPS. */
 881 void pkgAcquireStatus::Stop()
 882 {
 883    // Compute the CPS and elapsed time
 884    struct timeval NewTime;
 885    gettimeofday(&NewTime,0);
 886
 887    double Delta = NewTime.tv_sec - StartTime.tv_sec +
 888                   (NewTime.tv_usec - StartTime.tv_usec)/1000000.0;
 889
 890    // Compute the CPS value
 891    if (Delta < 0.01)
 892       CurrentCPS = 0;
 893    else
 894       CurrentCPS = FetchedBytes/Delta;
 895    LastBytes = CurrentBytes;
 896    ElapsedTime = (unsigned int)Delta;
 897 }
 898                                                                         /*}}}*/
 899 // AcquireStatus::Fetched - Called when a byte set has been fetched     /*{{{*/
 900 // ---------------------------------------------------------------------
 901 /* This is used to get accurate final transfer rate reporting. */
 902 void pkgAcquireStatus::Fetched(unsigned long Size,unsigned long Resume)
 903 {
 904    FetchedBytes += Size - Resume;
 905 }
 906                                                                         /*}}}*/