apt-pkg/acquire.cc

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