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