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